polyc-facts 2026.9.0

//! The `conversation-core/v1` positioned fold.
//!
//! [`crate::committed`] owns the commit-scoping rule this module reuses: a turn
//! is committed when both its `turn_start` and its `turn_complete` marker are
//! present. This module answers a different question. It projects one source
//! partition's committed turns and message text into row facts that carry the
//! REAL journal position of every event.
//!
//! # Why a second entry point
//!
//! [`crate::committed::committed_message_facts`] carries an `ordinal`: an index
//! into the slice it was handed. That coordinate is correct for search ordering
//! and wrong for a projection row. A projected row is addressed by the position
//! the journal assigned, so a consumer can join it against any other fact in
//! the same partition. The two folds also disagree on two rules on purpose:
//!
//! - an empty decoded text block emits a row here, matching the current public
//!   `messages` table, and emits none there, matching the search index;
//! - a relevant message payload that cannot be decoded refuses the whole fold
//!   here. Publishing a generation while silently dropping one of its rows
//!   would make an incomplete artifact indistinguishable from a complete one.
//!
//! # What this module does not own
//!
//! Excision and text withholding are separate shared contracts, and
//! [`prepare_conversation_core`] is the one place that names the order they
//! apply in. It returns the excised position set rather than hiding it, because
//! a caller that needs to report a removal needs that set. The fold itself is
//! pure over whatever events it is given.
//!
//! Physical encoding is not here either. This module emits logical row facts.
//! `polyc-projection` owns the family's logical schema and its fingerprint, and
//! the projector owns Arrow and Parquet.

use std::collections::{BTreeMap, BTreeSet};

use polyc_eventlog_model::{Event, TrustTag};
use polyc_proto::proto::polychrome::agent::v1::Message;
use polyc_proto::{events_decode::try_decode_event_payload, kinds};

use crate::committed::committed_turn_ids;
use crate::excision::{excised_positions, strip_excised, verified_excisions_matching};
use crate::message_content::{MessageContent, fold_message_content};
use crate::withholding::{withheld_turn_ids_positioned, withhold_paused_turn_text_positioned};

/// The two event kinds `conversation-core/v1` reads message text from.
///
/// Exactly the pair the current `messages` table folds. A tool call, a tool
/// result, and a reasoning thought are E2 scope and emit no v1 row.
const MESSAGE_KIND_BASES: [&str; 2] = [kinds::USER_MSG, kinds::OUTPUT_MSG];

/// One committed turn's boundary facts in one source partition.
///
/// [`Self::first_position`] is deliberately distinct from
/// [`Self::start_position`]. Turn-ordering consumers order a committed turn by
/// the minimum position of ANY event that carries its id, and the commit
/// contract permits a completion marker before a start marker and permits
/// duplicates of either. Collapsing the two would silently change that order.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommittedTurnFact {
    /// The committed turn.
    pub turn_id: uuid::Uuid,
    /// The minimum journal position of any event carrying this turn id.
    pub first_position: u64,
    /// The minimum journal position carrying this turn's `turn_start`.
    pub start_position: u64,
    /// The maximum journal position carrying this turn's `turn_complete`.
    pub complete_position: u64,
}

/// One text block from a `user_msg` or `output_msg` in a committed turn.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConversationMessageFact {
    /// The event's real journal append position.
    pub position: u64,
    /// The committed turn the message belongs to.
    pub turn_id: uuid::Uuid,
    /// The wire `Message.role`, verbatim.
    pub role: String,
    /// The wire `Message.internal_only`, verbatim.
    pub internal_only: bool,
    /// The full, untruncated text block, empty text included.
    pub text: String,
    /// The event's provenance tag.
    pub trust: TrustTag,
}

/// One source partition's complete `conversation-core/v1` row facts.
///
/// Both vectors are ordered: turns by `turn_id`, messages by `position`. The
/// order is part of the contract, because a generation's bytes must be a
/// function of its source prefix alone.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ConversationCoreFacts {
    /// Committed turn boundaries, ordered by turn id.
    pub turns: Vec<CommittedTurnFact>,
    /// Committed message text, ordered by journal position.
    pub messages: Vec<ConversationMessageFact>,
}

/// Why one source prefix cannot produce a complete `conversation-core/v1` fold.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ConversationCoreError {
    /// A `user_msg` or `output_msg` in a committed turn carries bytes that are
    /// not a `Message`.
    ///
    /// The fold refuses rather than skipping the row. A skipped row would leave
    /// an incomplete generation that reads exactly like a complete one.
    #[error("position {position} carries a {kind} payload that is not a message: {reason}")]
    UndecodableMessage {
        /// The refusing event's journal position.
        position: u64,
        /// The refusing event's kind base.
        kind: &'static str,
        /// What the decoder reported.
        reason: String,
    },
    /// Two events claim the same journal position.
    ///
    /// A journal position is dense and unique within one incarnation, so a
    /// repeat means the caller assembled a prefix from more than one source or
    /// delivered the same commit twice with different content.
    #[error("position {position} appears more than once in one source prefix")]
    RepeatedPosition {
        /// The repeated position.
        position: u64,
    },
}

/// What [`prepare_conversation_core`] removed or withheld before the fold ran.
///
/// The removed position set travels back to the caller rather than staying
/// inside, because a caller reporting "this was removed" needs it and cannot
/// re-derive it from the stripped events.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PreparedSource {
    /// Journal positions a verified excision removed.
    pub excised: BTreeSet<u64>,
    /// Turns whose model narration a `turn_text_withheld` marker withheld.
    pub withheld: BTreeSet<uuid::Uuid>,
}

/// A lifecycle event that can invalidate facts a generation already published.
///
/// Neither is an ordinary tail commit. Both change what the authoritative view
/// says about POSITIONS THAT ARE ALREADY IN A PUBLISHED GENERATION, so applying
/// one incrementally would leave a current generation that contradicts its own
/// source. A consumer that sees one stops, rebuilds from a fresh authoritative
/// view, and publishes a higher generation before acknowledging the cursor that
/// carried it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum LifecycleBarrier {
    /// A verified excision removed content.
    Excision,
}

/// Returns the lifecycle barrier a chunk carries, where it carries one.
///
/// Only the marker's presence is reported. Whether the excision it names
/// verifies, and which positions it removes, are
/// [`prepare_conversation_core`]'s business over the rebuilt prefix — an
/// unverified marker still means the source changed under a published
/// generation, and the rebuild is what settles what it changed.
#[must_use]
pub fn lifecycle_barrier(events: &[(u64, Event)]) -> Option<LifecycleBarrier> {
    events
        .iter()
        .any(|(_, event)| kinds::base(&event.kind) == kinds::TAINT_EXCISION)
        .then_some(LifecycleBarrier::Excision)
}

/// Applies excision and text withholding to one partition's prefix, in order.
///
/// Excision runs first: a verified excision marker is an ordinary append, so
/// nothing later in the pipeline would notice it, and stripping before the fold
/// is what keeps removed text out of a published artifact rather than relying
/// on a reader-side filter. Withholding runs second, over the stripped events,
/// and flips `internal_only` on the model narration of every paused turn.
///
/// Both mutations are the existing shared contracts. This function fixes their
/// order and nothing else.
///
/// `partition` is the logical State partition name. The excision marker carries
/// a bare conversation id, so the comparison adds the `conv-` prefix the
/// partition name carries.
pub fn prepare_conversation_core(events: &mut [(u64, Event)], partition: &str) -> PreparedSource {
    let excisions = verified_excisions_matching(events, partition, |excision| {
        format!("conv-{}", excision.conversation_id) == partition
    });
    let excised = excised_positions(events, &excisions);
    strip_excised(events, &excised);

    let withheld = withheld_turn_ids_positioned(events);
    if !withheld.is_empty() {
        withhold_paused_turn_text_positioned(events, &withheld);
    }

    PreparedSource { excised, withheld }
}

/// Projects one prepared source prefix into `conversation-core/v1` row facts.
///
/// The input is one source partition's events with their real journal
/// positions, already prepared by [`prepare_conversation_core`]. The output is
/// a function of that input alone: the same prefix always folds to the same
/// rows, in the same order.
///
/// A turn appears only when the prefix contains both of its markers, so a turn
/// still open at the end of the prefix contributes no rows at all. It appears
/// in a later generation whose prefix reaches its completion marker.
///
/// # Errors
///
/// Returns [`ConversationCoreError::RepeatedPosition`] when one position
/// appears twice, and [`ConversationCoreError::UndecodableMessage`] when a
/// relevant, non-empty message payload cannot be decoded. Content that decodes
/// but carries no v1 fact — a thought, a tool call, a tool result, an absent
/// content block — is not an error; it simply emits no row.
pub fn fold_conversation_core(
    events: &[(u64, Event)],
) -> Result<ConversationCoreFacts, ConversationCoreError> {
    let mut seen: BTreeSet<u64> = BTreeSet::new();
    for (position, _) in events {
        if !seen.insert(*position) {
            return Err(ConversationCoreError::RepeatedPosition {
                position: *position,
            });
        }
    }

    let committed = committed_turn_ids(events.iter().map(|(_, event)| event));

    // One pass over the prefix fills both tables. `boundaries` accumulates the
    // three positions each committed turn's row needs; a turn id that is not
    // committed is never inserted, so an open turn cannot leak a partial row.
    let mut boundaries: BTreeMap<uuid::Uuid, CommittedTurnFact> = BTreeMap::new();
    let mut messages = Vec::new();

    for (position, event) in events {
        let (base, turn_uuid) = kinds::parse(&event.kind);
        let Some(turn_id) = turn_uuid.filter(|id| committed.contains(id)) else {
            continue;
        };

        // Both sentinels below are replaced by real marker positions before the
        // loop ends: a turn is in `committed` only when the prefix carries at
        // least one of each marker, and every marker reaches this branch.
        let fact = boundaries.entry(turn_id).or_insert(CommittedTurnFact {
            turn_id,
            first_position: *position,
            start_position: u64::MAX,
            complete_position: 0,
        });
        fact.first_position = fact.first_position.min(*position);
        if base == kinds::TURN_START {
            fact.start_position = fact.start_position.min(*position);
        }
        if base == kinds::TURN_COMPLETE {
            fact.complete_position = fact.complete_position.max(*position);
        }

        if !MESSAGE_KIND_BASES.contains(&base) {
            continue;
        }
        if event.payload.is_empty() {
            // An empty payload decodes to a message with no content block,
            // which carries no v1 fact. Saying so here keeps the refusal below
            // about bytes that claim to be a message and are not.
            continue;
        }
        let message = try_decode_event_payload::<Message>(&event.payload).map_err(|error| {
            ConversationCoreError::UndecodableMessage {
                position: *position,
                kind: if base == kinds::USER_MSG {
                    kinds::USER_MSG
                } else {
                    kinds::OUTPUT_MSG
                },
                reason: error.to_string(),
            }
        })?;
        let folded = fold_message_content(&message, *position, None, event.trust.as_str());
        let MessageContent::Text(text) = folded.content else {
            continue;
        };
        messages.push(ConversationMessageFact {
            position: *position,
            turn_id,
            role: message.role,
            internal_only: message.internal_only,
            text: text.text,
            trust: event.trust,
        });
    }

    Ok(ConversationCoreFacts {
        turns: boundaries.into_values().collect(),
        messages,
    })
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use buffa::Message as _;
    use polyc_proto::proto::polychrome::agent::v1::{Content, Message, TextContent, content};

    use super::*;

    fn turn() -> uuid::Uuid {
        uuid::Uuid::from_u128(0x1111_2222_3333_4444_5555_6666_7777_8888)
    }

    fn other_turn() -> uuid::Uuid {
        uuid::Uuid::from_u128(0x9999_aaaa_bbbb_cccc_dddd_eeee_ffff_0000)
    }

    fn marker(base: &str, id: uuid::Uuid) -> Event {
        Event::new(kinds::tagged(base, &id), Vec::new())
    }

    fn text_payload(text: &str) -> Vec<u8> {
        Message {
            role: "model".into(),
            content: buffa::MessageField::some(Content {
                r#type: Some(content::Type::Text(Box::new(TextContent {
                    text: text.to_owned(),
                    ..Default::default()
                }))),
                ..Default::default()
            }),
            internal_only: false,
            ..Default::default()
        }
        .encode_to_vec()
    }

    fn message(base: &str, id: uuid::Uuid, text: &str, trust: TrustTag) -> Event {
        Event::with_trust(kinds::tagged(base, &id), text_payload(text), trust)
    }

    fn committed_pair(id: uuid::Uuid, start: u64, complete: u64) -> Vec<(u64, Event)> {
        vec![
            (start, marker(kinds::TURN_START, id)),
            (complete, marker(kinds::TURN_COMPLETE, id)),
        ]
    }

    #[test]
    fn both_markers_are_required() {
        let events = vec![(1, marker(kinds::TURN_START, turn()))];
        let facts = fold_conversation_core(&events).unwrap();
        assert!(facts.turns.is_empty());
    }

    #[test]
    fn duplicate_markers_choose_minimum_start_and_maximum_completion() {
        let mut events = committed_pair(turn(), 5, 9);
        events.push((3, marker(kinds::TURN_START, turn())));
        events.push((11, marker(kinds::TURN_COMPLETE, turn())));
        events.sort_by_key(|(position, _)| *position);

        let facts = fold_conversation_core(&events).unwrap();
        let [only] = facts.turns.as_slice() else {
            panic!("one committed turn: {:?}", facts.turns);
        };
        assert_eq!(only.start_position, 3);
        assert_eq!(only.complete_position, 11);
        assert_eq!(only.first_position, 3);
    }

    #[test]
    fn first_position_precedes_both_boundary_positions() {
        // A message arrives before the turn's own start marker: legal, and the
        // reason `first_position` is not `start_position`.
        let mut events = vec![(
            2,
            message(kinds::USER_MSG, turn(), "hi", TrustTag::TrustedUser),
        )];
        events.extend(committed_pair(turn(), 4, 8));

        let facts = fold_conversation_core(&events).unwrap();
        let [only] = facts.turns.as_slice() else {
            panic!("one committed turn");
        };
        assert_eq!(only.first_position, 2);
        assert_eq!(only.start_position, 4);
        assert_eq!(only.complete_position, 8);
    }

    #[test]
    fn message_rows_carry_the_real_journal_position() {
        let mut events = committed_pair(turn(), 100, 400);
        events.push((
            250,
            message(kinds::OUTPUT_MSG, turn(), "spoke", TrustTag::Unspecified),
        ));
        events.sort_by_key(|(position, _)| *position);

        let facts = fold_conversation_core(&events).unwrap();
        let [only] = facts.messages.as_slice() else {
            panic!("one message row");
        };
        assert_eq!(only.position, 250);
        assert_eq!(only.turn_id, turn());
        assert_eq!(only.role, "model");
        assert!(!only.internal_only);
        assert_eq!(only.text, "spoke");
        assert_eq!(only.trust, TrustTag::Unspecified);
    }

    #[test]
    fn an_empty_text_block_still_emits_one_row() {
        let mut events = committed_pair(turn(), 1, 3);
        events.insert(
            1,
            (
                2,
                message(kinds::OUTPUT_MSG, turn(), "", TrustTag::Unspecified),
            ),
        );

        let facts = fold_conversation_core(&events).unwrap();
        assert_eq!(facts.messages.len(), 1);
        assert_eq!(facts.messages[0].text, "");
    }

    #[test]
    fn an_uncommitted_turns_text_emits_no_row() {
        let events = vec![
            (1, marker(kinds::TURN_START, turn())),
            (
                2,
                message(
                    kinds::USER_MSG,
                    turn(),
                    "never committed",
                    TrustTag::TrustedUser,
                ),
            ),
        ];
        let facts = fold_conversation_core(&events).unwrap();
        assert!(facts.messages.is_empty());
        assert!(facts.turns.is_empty());
    }

    #[test]
    fn a_message_with_no_turn_suffix_emits_no_row() {
        let mut events = committed_pair(turn(), 1, 3);
        events.insert(
            1,
            (
                2,
                Event::new(kinds::USER_MSG.to_owned(), text_payload("orphan")),
            ),
        );
        let facts = fold_conversation_core(&events).unwrap();
        assert!(facts.messages.is_empty());
    }

    #[test]
    fn one_partitions_marker_cannot_complete_another_partitions_turn() {
        // Both folds run per partition, so cross-partition completion is
        // impossible by construction: this asserts the per-call contract that
        // makes it so — a prefix carrying only one marker commits nothing.
        let first = vec![(1, marker(kinds::TURN_START, turn()))];
        let second = vec![(1, marker(kinds::TURN_COMPLETE, turn()))];
        assert!(fold_conversation_core(&first).unwrap().turns.is_empty());
        assert!(fold_conversation_core(&second).unwrap().turns.is_empty());
    }

    #[test]
    fn an_undecodable_relevant_payload_refuses_the_fold() {
        let mut events = committed_pair(turn(), 1, 3);
        events.insert(
            1,
            (
                2,
                Event::new(kinds::tagged(kinds::OUTPUT_MSG, &turn()), vec![0xff; 8]),
            ),
        );
        let error = fold_conversation_core(&events).unwrap_err();
        assert!(matches!(
            error,
            ConversationCoreError::UndecodableMessage { position: 2, .. }
        ));
    }

    #[test]
    fn an_undecodable_irrelevant_payload_is_ignored() {
        let mut events = committed_pair(turn(), 1, 3);
        events.insert(1, (2, Event::new("usage".to_owned(), vec![0xff; 8])));
        assert!(fold_conversation_core(&events).is_ok());
    }

    #[test]
    fn a_repeated_position_refuses_the_fold() {
        let events = vec![
            (1, marker(kinds::TURN_START, turn())),
            (1, marker(kinds::TURN_COMPLETE, turn())),
        ];
        assert!(matches!(
            fold_conversation_core(&events).unwrap_err(),
            ConversationCoreError::RepeatedPosition { position: 1 }
        ));
    }

    #[test]
    fn rows_are_ordered_by_turn_id_and_position() {
        let mut events = committed_pair(other_turn(), 1, 2);
        events.extend(committed_pair(turn(), 3, 6));
        events.push((
            5,
            message(kinds::OUTPUT_MSG, turn(), "b", TrustTag::Unspecified),
        ));
        events.push((
            4,
            message(kinds::USER_MSG, turn(), "a", TrustTag::TrustedUser),
        ));
        events.sort_by_key(|(position, _)| *position);

        let facts = fold_conversation_core(&events).unwrap();
        let ids: Vec<_> = facts.turns.iter().map(|turn| turn.turn_id).collect();
        let mut sorted = ids.clone();
        sorted.sort_unstable();
        assert_eq!(ids, sorted);
        let positions: Vec<_> = facts.messages.iter().map(|m| m.position).collect();
        assert_eq!(positions, vec![4, 5]);
    }

    #[test]
    fn the_same_prefix_folds_to_the_same_rows() {
        let mut events = committed_pair(turn(), 1, 4);
        events.push((
            2,
            message(kinds::USER_MSG, turn(), "ask", TrustTag::TrustedUser),
        ));
        events.push((
            3,
            message(kinds::OUTPUT_MSG, turn(), "answer", TrustTag::Unspecified),
        ));
        events.sort_by_key(|(position, _)| *position);

        assert_eq!(
            fold_conversation_core(&events).unwrap(),
            fold_conversation_core(&events).unwrap()
        );
    }

    #[test]
    fn text_matches_the_search_fold_wherever_that_fold_emits_a_row() {
        // Parity against the behavior that exists today. The search fold drops
        // empty text and keys by slice ordinal; everything it DOES emit must
        // carry the same turn id and the same untruncated text here.
        let mut events = committed_pair(turn(), 1, 5);
        events.push((
            2,
            message(kinds::USER_MSG, turn(), "ask", TrustTag::TrustedUser),
        ));
        events.push((
            3,
            message(kinds::OUTPUT_MSG, turn(), "", TrustTag::Unspecified),
        ));
        events.push((
            4,
            message(kinds::OUTPUT_MSG, turn(), "answer", TrustTag::Unspecified),
        ));
        events.sort_by_key(|(position, _)| *position);

        let bare: Vec<Event> = events.iter().map(|(_, event)| event.clone()).collect();
        let search = crate::committed::committed_message_facts(&bare);
        let projected = fold_conversation_core(&events).unwrap();

        let search_pairs: Vec<(String, String)> = search
            .iter()
            .map(|fact| (fact.turn_id.clone(), fact.text.clone()))
            .collect();
        let projected_pairs: Vec<(String, String)> = projected
            .messages
            .iter()
            .filter(|row| !row.text.is_empty())
            .map(|row| (row.turn_id.to_string(), row.text.clone()))
            .collect();
        assert_eq!(search_pairs, projected_pairs);
        // And the projected fold keeps the row the search fold drops.
        assert_eq!(projected.messages.len(), search.len() + 1);
    }

    #[test]
    fn withholding_flips_internal_only_before_the_fold_emits_a_row() {
        let mut events = committed_pair(turn(), 1, 4);
        events.push((
            2,
            message(
                kinds::OUTPUT_MSG,
                turn(),
                "narration",
                TrustTag::Unspecified,
            ),
        ));
        events.push((3, marker(kinds::TURN_TEXT_WITHHELD, turn())));
        events.sort_by_key(|(position, _)| *position);

        let before = fold_conversation_core(&events).unwrap();
        assert!(!before.messages[0].internal_only);

        let prepared = prepare_conversation_core(&mut events, "conv-x");
        assert!(prepared.withheld.contains(&turn()));
        let after = fold_conversation_core(&events).unwrap();
        assert!(after.messages[0].internal_only);
        assert_eq!(after.messages[0].text, "narration");
    }

    #[test]
    fn preparing_reports_the_positions_it_did_not_remove() {
        let mut events = committed_pair(turn(), 1, 3);
        events.insert(
            1,
            (
                2,
                message(kinds::OUTPUT_MSG, turn(), "kept", TrustTag::Unspecified),
            ),
        );
        let prepared = prepare_conversation_core(&mut events, "conv-x");
        assert!(prepared.excised.is_empty());
        assert!(prepared.withheld.is_empty());
        assert_eq!(fold_conversation_core(&events).unwrap().messages.len(), 1);
    }
}