polyc-facts 2026.10.2

//! The `persona-memory/v1` fold (6A-Q).
//!
//! Projects one `persona-{id}-mem` partition's records, under one incarnation,
//! into the family's thirteen tables (`crates/projection/src/family.rs`). Pure
//! over the events it is given: the same prefix always folds to the same
//! rows, in the same order, and this module reads no conversation journal and
//! no other persona's partition.
//!
//! # The owner/portable split
//!
//! A `memory_added` row always lands in [`PersonaMemoryFacts::facts`]. When
//! its closed `scope` folds as [`FactScope::Portable`], the SAME row is
//! duplicated into [`PersonaMemoryFacts::portable_facts`] — never derived
//! later by a query-time filter, so a participant's catalog registration
//! physically never has to open the owner-only segment to find the portable
//! rows. The same duplication applies to invalidations, keyed off the
//! CLOSED FACT's own folded scope (not the invalidation's own — it carries
//! none), and to a fact's `entities` list.
//!
//! # Identity never reaches a Visible row
//!
//! [`MemoryProvenanceIdentityRow`] is the only row shape that carries a
//! persona's provenance identity (provider, external id, display name). No
//! field of any other row type carries it — `scripts/check_persona_memory_family.py`'s
//! byte sweep (N-11) holds this against the family's published columns, and
//! [`fold_persona_memory`]'s own row types are what make it true: there is no
//! field to leak from, not just a value nobody set.

use std::collections::BTreeMap;

use polyc_eventlog_model::Event;
use polyc_proto::events_decode::try_decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::{
    ConversationSummaryEvent, MemoryAddedEvent, MemoryCorroboratedEvent, MemoryDurability,
    MemoryExtractedEvent, MemoryInvalidatedEvent, MemoryProvenance, MemoryScope,
    ProfileDocUpdatedEvent,
};

/// Closed `scope` vocabulary a row's `scope`/`fact_scope` column takes.
pub const SCOPE_DIRECT: &str = "direct";
/// See [`SCOPE_DIRECT`].
pub const SCOPE_PORTABLE: &str = "portable";
/// See [`SCOPE_DIRECT`].
///
/// Also the fold's answer for an invalidation of a fact this build cannot
/// resolve — its true scope cannot be proven portable, so it is never
/// duplicated into a `memory_portable_*` table.
pub const SCOPE_UNSPECIFIED: &str = "unspecified";

/// Closed `durability` vocabulary a fact row's `durability` column takes.
pub const DURABILITY_DURABLE: &str = "durable";
/// See [`DURABILITY_DURABLE`].
pub const DURABILITY_SESSION: &str = "session";
/// See [`DURABILITY_DURABLE`].
///
/// The fold's fail-closed answer for an absent or unrecognized discriminant,
/// matching `polyc_persona_memory::apply`'s own rule: only an explicit
/// durable classification is durable.
pub const DURABILITY_UNSPECIFIED: &str = "unspecified";

/// One `memory_added` row's folded scope, as a closed Rust value the rest of
/// this module matches on exhaustively.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FactScope {
    Direct,
    Portable,
    Unspecified,
}

impl FactScope {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Direct => SCOPE_DIRECT,
            Self::Portable => SCOPE_PORTABLE,
            Self::Unspecified => SCOPE_UNSPECIFIED,
        }
    }

    fn of(scope: buffa::EnumValue<MemoryScope>) -> Self {
        match scope.as_known() {
            Some(MemoryScope::Direct) => Self::Direct,
            Some(MemoryScope::Portable) => Self::Portable,
            Some(MemoryScope::Unspecified) | None => Self::Unspecified,
        }
    }
}

fn durability_str(durability: buffa::EnumValue<MemoryDurability>) -> &'static str {
    match durability.as_known() {
        Some(MemoryDurability::Durable) => DURABILITY_DURABLE,
        Some(MemoryDurability::Session) => DURABILITY_SESSION,
        Some(MemoryDurability::Unspecified) | None => DURABILITY_UNSPECIFIED,
    }
}

/// The provenance triple every owner-visible and portable row carries.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[allow(
    clippy::struct_field_names,
    reason = "conversation_id/turn_exec_id/turn_id are the wire and column names this triple mirrors"
)]
pub struct ProvenanceTriple {
    /// The conversation the record is attributed to.
    pub conversation_id: String,
    /// The turn (exec id) whose transcript produced the record, hyphenated
    /// as the persona-memory authority wrote it.
    pub turn_exec_id: String,
    /// [`Self::turn_exec_id`], normalized to the simple-hex form the
    /// conversation partition's own tables key `turn_id` on. Empty when
    /// [`Self::turn_exec_id`] is empty or not a UUID.
    pub turn_id: String,
}

/// Normalizes a persona-memory partition's hyphenated turn-exec UUID onto the
/// conversation partition's simple-hex `turn_id` form.
///
/// The two partitions' own tables must key identically for a reader to join
/// them, and this is the one place that normalization happens — moved into
/// the fold from `crate::control_plane`'s pre-cutover splice (6A-Q,
/// `02-DESIGN.md` §2.2), so every reader of this family sees the same joined
/// form rather than re-deriving it.
fn normalize_turn_exec_id(raw: &str) -> String {
    uuid::Uuid::parse_str(raw).map_or_else(|_| String::new(), |id| id.as_simple().to_string())
}

fn provenance_triple(provenance: Option<&MemoryProvenance>) -> ProvenanceTriple {
    let Some(provenance) = provenance else {
        return ProvenanceTriple::default();
    };
    ProvenanceTriple {
        conversation_id: provenance.conversation_id.clone(),
        turn_id: normalize_turn_exec_id(&provenance.turn_exec_id),
        turn_exec_id: provenance.turn_exec_id.clone(),
    }
}

/// The identity carried by a fact or invalidation's provenance —
/// `memory_provenance_identity` rows only, never a Visible one (see this
/// module's doc).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ProvenanceIdentity {
    /// The edge provider ("slack", "telegram", …).
    pub provider: String,
    /// The provider scope qualifying the id.
    pub scope: String,
    /// The provider-native stable identifier.
    pub external_id: String,
    /// The display name as last observed at the edge.
    pub display_name: String,
}

fn provenance_identity(provenance: Option<&MemoryProvenance>) -> Option<ProvenanceIdentity> {
    let identity = provenance?.identity.as_option()?;
    Some(ProvenanceIdentity {
        provider: identity.provider.clone(),
        scope: identity.scope.clone(),
        external_id: identity.external_id.clone(),
        display_name: identity.display_name.clone(),
    })
}

/// One `memory_facts`/`memory_portable_facts` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryFactRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The journal-unique fact id.
    pub fact_id: String,
    /// The fact's text.
    pub text: String,
    /// The entities the fact mentions.
    pub entities: Vec<String>,
    /// The record's provenance triple.
    pub provenance: ProvenanceTriple,
    /// When the fact was true/stated.
    pub event_time_ms: u64,
    /// When the extraction pass persisted it.
    pub ingested_at_ms: u64,
    /// Start of the validity interval.
    pub valid_from_ms: u64,
    /// Extractor confidence in basis points.
    pub confidence_bps: u32,
    /// The closed scope this row folded under.
    scope: FactScope,
    /// The closed durability vocabulary value.
    pub durability: &'static str,
}

impl MemoryFactRow {
    /// Returns the closed `scope` column value.
    #[must_use]
    pub const fn scope(&self) -> &'static str {
        self.scope.as_str()
    }

    /// Reports whether this row also lands in the `memory_portable_*` tables.
    #[must_use]
    const fn is_portable(&self) -> bool {
        matches!(self.scope, FactScope::Portable)
    }
}

/// One `memory_invalidations`/`memory_portable_invalidations` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryInvalidationRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The fact whose validity interval closes.
    pub fact_id: String,
    /// End of the validity interval.
    pub invalid_at_ms: u64,
    /// Why the fact was invalidated.
    pub reason: String,
    /// The fact superseding this one, when the invalidation named one.
    pub superseded_by_fact_id: String,
    /// The record's provenance triple.
    pub provenance: ProvenanceTriple,
    /// The CLOSED fact's own folded scope — `unspecified` when the closed
    /// fact is unknown to this fold (routes to the owner table only).
    fact_scope: FactScope,
}

impl MemoryInvalidationRow {
    /// Returns the closed `fact_scope` column value.
    #[must_use]
    pub const fn fact_scope(&self) -> &'static str {
        self.fact_scope.as_str()
    }

    /// Reports whether this row also lands in the `memory_portable_*` tables.
    #[must_use]
    const fn is_portable(&self) -> bool {
        matches!(self.fact_scope, FactScope::Portable)
    }
}

/// One `memory_lists`/`memory_portable_lists` element.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryListRow {
    /// The event's real journal append position.
    pub position: u64,
    /// Which owner row this element belongs to: `"fact"`, `"corroboration"`,
    /// or `"profile_rewrite"`.
    pub owner_kind: &'static str,
    /// The owning row's own id (a fact id, or the profile rewrite's
    /// `updated_at_ms` rendered as a string, since a profile rewrite mints
    /// no id of its own).
    pub owner_id: String,
    /// Which list this element belongs to: `"entities"`, `"corroborators"`,
    /// or `"source_fact_ids"`.
    pub list_kind: &'static str,
    /// The element's position within its list.
    pub index: u64,
    /// The element's text.
    pub text_value: String,
    /// Whether the owning fact folded as portable — decides whether this
    /// element is ALSO duplicated into `memory_portable_lists`. Only ever
    /// `true` for an `entities` element of a portable fact.
    portable: bool,
}

/// One `memory_corroborations` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryCorroborationRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The survivor fact this delta grows.
    pub fact_id: String,
    /// The survivor's origin provenance.
    pub origin: ProvenanceTriple,
    /// When the merge was recorded.
    pub at_ms: u64,
}

/// One `memory_extractions` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryExtractionRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The processed turn's conversation.
    pub conversation_id: String,
    /// The processed turn's exec id.
    pub turn_exec_id: String,
    /// When the pass completed.
    pub extracted_at_ms: u64,
    /// How many facts the pass added.
    pub facts_added: u32,
    /// How many facts the pass invalidated.
    pub facts_invalidated: u32,
}

/// One `memory_profile_rewrites` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryProfileRewriteRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The rendered document text.
    pub document: String,
    /// When the consolidation ran.
    pub updated_at_ms: u64,
    /// The active facts the document was distilled from — also folded into
    /// `memory_lists` under `list_kind = "source_fact_ids"`.
    pub source_fact_ids: Vec<String>,
}

/// One `memory_summaries` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemorySummaryRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The summarized conversation.
    pub conversation_id: String,
    /// The one-line summary.
    pub summary_text: String,
    /// Turn wall-clock of the summarized activity.
    pub event_time_ms: u64,
    /// The event's own closed scope column (informational only — this table
    /// has no portable sibling; see `02-DESIGN.md` §2.2).
    pub scope: &'static str,
}

/// One `memory_provenance_identity` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryProvenanceIdentityRow {
    /// The event's real journal append position.
    pub position: u64,
    /// Which record this identity belongs to: `"fact"` or `"invalidation"`.
    pub record_kind: &'static str,
    /// The owning record's fact id.
    pub record_id: String,
    /// The identity carried by that record's provenance.
    pub identity: ProvenanceIdentity,
}

/// One `memory_fences` row — payload-free by construction.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MemoryFenceRow {
    /// The event's real journal append position.
    pub position: u64,
}

/// One `memory_unknown` row: a record kind this build cannot name.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryUnknownRow {
    /// The event's real journal append position.
    pub position: u64,
    /// The unrecognized kind string, verbatim.
    pub kind: String,
}

/// Why one partition prefix cannot produce a complete `persona-memory/v1`
/// fold.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum PersonaMemoryFoldError {
    /// A record of a KNOWN kind carries bytes that do not decode as that
    /// kind's message. Refuses the whole fold rather than skipping the row —
    /// a skipped row would leave an incomplete generation indistinguishable
    /// from a complete one.
    #[error("position {position} carries a {kind} payload that will not decode: {reason}")]
    UndecodableRecord {
        /// The refusing event's journal position.
        position: u64,
        /// The refusing event's kind.
        kind: &'static str,
        /// What the decoder reported.
        reason: String,
    },
    /// Two events claim the same journal position.
    #[error("position {position} appears more than once in one source prefix")]
    RepeatedPosition {
        /// The repeated position.
        position: u64,
    },
}

/// One partition's complete `persona-memory/v1` row facts.
///
/// Every vector is ordered by journal position (list rows: by owner row
/// position, then index) — the order is part of the contract, since a
/// generation's bytes must be a function of its source prefix alone.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PersonaMemoryFacts {
    /// Every `memory_added` row, every scope.
    pub facts: Vec<MemoryFactRow>,
    /// Every `memory_invalidated` row.
    pub invalidations: Vec<MemoryInvalidationRow>,
    /// Every list element of an owner-visible row.
    pub lists: Vec<MemoryListRow>,
    /// Every `memory_corroborated` row.
    pub corroborations: Vec<MemoryCorroborationRow>,
    /// Every `memory_extracted` row.
    pub extractions: Vec<MemoryExtractionRow>,
    /// Every `profile_doc_updated` row.
    pub profile_rewrites: Vec<MemoryProfileRewriteRow>,
    /// Every `conversation_summary` row.
    pub summaries: Vec<MemorySummaryRow>,
    /// Every fact's or invalidation's provenance identity, Fleet-only.
    pub provenance_identities: Vec<MemoryProvenanceIdentityRow>,
    /// Every `memory_fence_barrier` row, Fleet-only.
    pub fences: Vec<MemoryFenceRow>,
    /// Every record of an unrecognized kind, Fleet-only.
    pub unknown: Vec<MemoryUnknownRow>,
}

impl PersonaMemoryFacts {
    /// Returns every `memory_portable_facts` row: [`Self::facts`] filtered to
    /// `scope = portable`.
    #[must_use]
    pub fn portable_facts(&self) -> Vec<&MemoryFactRow> {
        self.facts.iter().filter(|row| row.is_portable()).collect()
    }

    /// Returns every `memory_portable_invalidations` row: [`Self::invalidations`]
    /// filtered to `fact_scope = portable`.
    #[must_use]
    pub fn portable_invalidations(&self) -> Vec<&MemoryInvalidationRow> {
        self.invalidations
            .iter()
            .filter(|row| row.is_portable())
            .collect()
    }

    /// Returns every `memory_portable_lists` element: an `entities` element
    /// of a portable fact.
    #[must_use]
    pub fn portable_lists(&self) -> Vec<&MemoryListRow> {
        self.lists.iter().filter(|row| row.portable).collect()
    }
}

/// The payload-free `memory_fence_barrier` kind (`persona-memory-host`'s own
/// constant, mirrored here rather than imported — a foundation crate depends
/// only on other foundations, and this string is the wire contract, not an
/// implementation detail either side may change unilaterally).
const KIND_MEMORY_FENCE_BARRIER: &str = "memory_fence_barrier";

/// Every kind this fold recognizes. A kind outside this set folds to
/// `memory_unknown` rather than refusing the generation — the authority may
/// carry a record kind this build predates, and the honest answer is "this
/// build cannot name it," not a refused generation.
const KNOWN_KINDS: [&str; 7] = [
    kinds::MEMORY_ADDED,
    kinds::MEMORY_INVALIDATED,
    kinds::MEMORY_CORROBORATED,
    kinds::PROFILE_DOC_UPDATED,
    kinds::CONVERSATION_SUMMARY,
    kinds::MEMORY_EXTRACTED,
    KIND_MEMORY_FENCE_BARRIER,
];

/// Projects one partition's prefix into `persona-memory/v1` row facts.
///
/// # Errors
///
/// Returns [`PersonaMemoryFoldError::RepeatedPosition`] when one position
/// appears twice, and [`PersonaMemoryFoldError::UndecodableRecord`] when a
/// record of a recognized kind carries bytes that do not decode as that
/// kind's message.
#[allow(clippy::too_many_lines)]
pub fn fold_persona_memory(
    events: &[(u64, Event)],
) -> Result<PersonaMemoryFacts, PersonaMemoryFoldError> {
    let mut seen = std::collections::BTreeSet::new();
    for (position, _) in events {
        if !seen.insert(*position) {
            return Err(PersonaMemoryFoldError::RepeatedPosition {
                position: *position,
            });
        }
    }

    // The closed scope of every fact seen so far, by fact id — what an
    // invalidation's own `fact_scope` column resolves against. A fact
    // invalidated before this fold ever saw its `memory_added` (the prefix
    // starts mid-history, which cannot happen for a FULL replay from origin,
    // but this fold makes no assumption about where its prefix starts)
    // resolves `Unspecified`, which routes the invalidation to the owner
    // table only — never guessed as portable.
    let mut fact_scopes: BTreeMap<String, FactScope> = BTreeMap::new();

    let mut facts = Vec::new();
    let mut invalidations = Vec::new();
    let mut lists = Vec::new();
    let mut corroborations = Vec::new();
    let mut extractions = Vec::new();
    let mut profile_rewrites = Vec::new();
    let mut summaries = Vec::new();
    let mut provenance_identities = Vec::new();
    let mut fences = Vec::new();
    let mut unknown = Vec::new();

    for (position, event) in events {
        let position = *position;
        let base = kinds::base(&event.kind);
        if !KNOWN_KINDS.contains(&base) {
            unknown.push(MemoryUnknownRow {
                position,
                kind: event.kind.clone(),
            });
            continue;
        }
        if base == kinds::MEMORY_ADDED {
            let added = decode::<MemoryAddedEvent>(position, "memory_added", &event.payload)?;
            let provenance = added.provenance.as_option();
            let scope = FactScope::of(added.scope);
            fact_scopes.insert(added.fact_id.clone(), scope);
            if let Some(identity) = provenance_identity(provenance) {
                provenance_identities.push(MemoryProvenanceIdentityRow {
                    position,
                    record_kind: "fact",
                    record_id: added.fact_id.clone(),
                    identity,
                });
            }
            for (index, entity) in added.entities.iter().enumerate() {
                lists.push(MemoryListRow {
                    position,
                    owner_kind: "fact",
                    owner_id: added.fact_id.clone(),
                    list_kind: "entities",
                    index: index as u64,
                    text_value: entity.clone(),
                    portable: matches!(scope, FactScope::Portable),
                });
            }
            facts.push(MemoryFactRow {
                position,
                fact_id: added.fact_id,
                text: added.text,
                entities: added.entities,
                provenance: provenance_triple(provenance),
                event_time_ms: added.event_time_ms,
                ingested_at_ms: added.ingested_at_ms,
                valid_from_ms: added.valid_from_ms,
                confidence_bps: added.confidence_bps,
                scope,
                durability: durability_str(added.durability),
            });
        } else if base == kinds::MEMORY_INVALIDATED {
            let invalidated =
                decode::<MemoryInvalidatedEvent>(position, "memory_invalidated", &event.payload)?;
            let provenance = invalidated.provenance.as_option();
            if let Some(identity) = provenance_identity(provenance) {
                provenance_identities.push(MemoryProvenanceIdentityRow {
                    position,
                    record_kind: "invalidation",
                    record_id: invalidated.fact_id.clone(),
                    identity,
                });
            }
            let fact_scope = fact_scopes
                .get(&invalidated.fact_id)
                .copied()
                .unwrap_or(FactScope::Unspecified);
            invalidations.push(MemoryInvalidationRow {
                position,
                fact_id: invalidated.fact_id,
                invalid_at_ms: invalidated.invalid_at_ms,
                reason: invalidated.reason,
                superseded_by_fact_id: invalidated.superseded_by_fact_id,
                provenance: provenance_triple(provenance),
                fact_scope,
            });
        } else if base == kinds::MEMORY_CORROBORATED {
            let corroborated =
                decode::<MemoryCorroboratedEvent>(position, "memory_corroborated", &event.payload)?;
            for (index, corroborator) in corroborated.corroborating.iter().enumerate() {
                lists.push(MemoryListRow {
                    position,
                    owner_kind: "corroboration",
                    owner_id: corroborated.fact_id.clone(),
                    list_kind: "corroborators",
                    index: index as u64,
                    text_value: corroborator.conversation_id.clone(),
                    portable: false,
                });
            }
            corroborations.push(MemoryCorroborationRow {
                position,
                fact_id: corroborated.fact_id,
                origin: provenance_triple(corroborated.origin.as_option()),
                at_ms: corroborated.at_ms,
            });
        } else if base == kinds::PROFILE_DOC_UPDATED {
            let rewrite =
                decode::<ProfileDocUpdatedEvent>(position, "profile_doc_updated", &event.payload)?;
            for (index, fact_id) in rewrite.source_fact_ids.iter().enumerate() {
                lists.push(MemoryListRow {
                    position,
                    owner_kind: "profile_rewrite",
                    owner_id: rewrite.updated_at_ms.to_string(),
                    list_kind: "source_fact_ids",
                    index: index as u64,
                    text_value: fact_id.clone(),
                    portable: false,
                });
            }
            profile_rewrites.push(MemoryProfileRewriteRow {
                position,
                document: rewrite.document,
                updated_at_ms: rewrite.updated_at_ms,
                source_fact_ids: rewrite.source_fact_ids,
            });
        } else if base == kinds::CONVERSATION_SUMMARY {
            let summary = decode::<ConversationSummaryEvent>(
                position,
                "conversation_summary",
                &event.payload,
            )?;
            summaries.push(MemorySummaryRow {
                position,
                conversation_id: summary.conversation_id,
                summary_text: summary.summary_text,
                event_time_ms: summary.event_time_ms,
                scope: FactScope::of(summary.scope).as_str(),
            });
        } else if base == kinds::MEMORY_EXTRACTED {
            let extracted =
                decode::<MemoryExtractedEvent>(position, "memory_extracted", &event.payload)?;
            extractions.push(MemoryExtractionRow {
                position,
                conversation_id: extracted.conversation_id,
                turn_exec_id: extracted.turn_exec_id,
                extracted_at_ms: extracted.extracted_at_ms,
                facts_added: extracted.facts_added,
                facts_invalidated: extracted.facts_invalidated,
            });
        } else {
            debug_assert_eq!(base, KIND_MEMORY_FENCE_BARRIER);
            fences.push(MemoryFenceRow { position });
        }
    }

    // An invalidation is folded before its closed fact's final scope is
    // known only when the fact was added AFTER it in this same prefix —
    // impossible for a real authority (an invalidation always names an
    // already-added fact), so no second pass is needed. `fact_scopes` is
    // built incrementally above in position order, which already covers
    // every legal history.

    Ok(PersonaMemoryFacts {
        facts,
        invalidations,
        lists,
        corroborations,
        extractions,
        profile_rewrites,
        summaries,
        provenance_identities,
        fences,
        unknown,
    })
}

fn decode<T: buffa::Message + Default>(
    position: u64,
    kind: &'static str,
    payload: &[u8],
) -> Result<T, PersonaMemoryFoldError> {
    if payload.is_empty() {
        return Ok(T::default());
    }
    try_decode_event_payload::<T>(payload).map_err(|error| {
        PersonaMemoryFoldError::UndecodableRecord {
            position,
            kind,
            reason: error.to_string(),
        }
    })
}

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

    use buffa::Message as _;
    use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;

    use super::*;

    fn provenance(conversation_id: &str, turn_exec_id: &str) -> MemoryProvenance {
        MemoryProvenance {
            conversation_id: conversation_id.to_owned(),
            turn_exec_id: turn_exec_id.to_owned(),
            identity: buffa::MessageField::some(ExternalIdentity {
                provider: "slack".to_owned(),
                external_id: "U123".to_owned(),
                display_name: "Alex".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    fn added(fact_id: &str, scope: MemoryScope, entities: Vec<String>) -> Event {
        let event = MemoryAddedEvent {
            fact_id: fact_id.to_owned(),
            text: "likes tea".to_owned(),
            entities,
            provenance: buffa::MessageField::some(provenance(
                "conv-a",
                "01950000-0000-7000-8000-00000000aaaa",
            )),
            event_time_ms: 1,
            ingested_at_ms: 2,
            valid_from_ms: 1,
            confidence_bps: 9000,
            scope: buffa::EnumValue::from(scope),
            durability: buffa::EnumValue::from(MemoryDurability::Durable),
            ..Default::default()
        };
        Event::new(kinds::MEMORY_ADDED, event.encode_to_vec())
    }

    fn invalidated(fact_id: &str) -> Event {
        let event = MemoryInvalidatedEvent {
            fact_id: fact_id.to_owned(),
            invalid_at_ms: 5,
            reason: "user-corrected".to_owned(),
            provenance: buffa::MessageField::some(provenance(
                "conv-a",
                "01950000-0000-7000-8000-00000000aaaa",
            )),
            ..Default::default()
        };
        Event::new(kinds::MEMORY_INVALIDATED, event.encode_to_vec())
    }

    #[test]
    fn a_direct_fact_never_reaches_the_portable_tables() {
        let events = vec![(1, added("f1", MemoryScope::Direct, vec!["tea".to_owned()]))];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.facts.len(), 1);
        assert_eq!(facts.facts[0].scope(), SCOPE_DIRECT);
        assert!(facts.portable_facts().is_empty());
        assert!(facts.portable_lists().is_empty());
    }

    /// Mutation: drop the `is_portable` filter in `portable_facts` and this
    /// reds — a `Direct` row would leak into the participant-visible table.
    #[test]
    fn a_portable_fact_is_duplicated_into_the_portable_table_and_no_other() {
        let events = vec![(
            1,
            added("f1", MemoryScope::Portable, vec!["tea".to_owned()]),
        )];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.facts.len(), 1);
        let portable = facts.portable_facts();
        assert_eq!(portable.len(), 1);
        assert_eq!(portable[0].fact_id, "f1");
        let portable_lists = facts.portable_lists();
        assert_eq!(portable_lists.len(), 1);
        assert_eq!(portable_lists[0].text_value, "tea");
    }

    #[test]
    fn an_invalidation_of_a_portable_fact_is_also_portable() {
        let events = vec![
            (1, added("f1", MemoryScope::Portable, vec![])),
            (2, invalidated("f1")),
        ];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.portable_invalidations().len(), 1);
    }

    /// Mutation: default `fact_scope` to `Portable` instead of `Unspecified`
    /// for an unresolved fact, and this reds — the invalidation would leak
    /// into the portable table for a fact this fold never saw added.
    #[test]
    fn an_invalidation_of_an_unresolved_fact_is_never_portable() {
        let events = vec![(1, invalidated("ghost"))];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.invalidations.len(), 1);
        assert_eq!(facts.invalidations[0].fact_scope(), SCOPE_UNSPECIFIED);
        assert!(facts.portable_invalidations().is_empty());
    }

    #[test]
    fn an_invalidation_of_a_direct_fact_stays_owner_only() {
        let events = vec![
            (1, added("f1", MemoryScope::Direct, vec![])),
            (2, invalidated("f1")),
        ];
        let facts = fold_persona_memory(&events).unwrap();
        assert!(facts.portable_invalidations().is_empty());
    }

    #[test]
    fn provenance_identity_never_reaches_a_fact_or_invalidation_row() {
        let events = vec![(1, added("f1", MemoryScope::Portable, vec![]))];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.provenance_identities.len(), 1);
        assert_eq!(facts.provenance_identities[0].identity.provider, "slack");
        // `MemoryFactRow` structurally has no identity field to check against
        // — this assertion documents the guarantee at the type level.
        let _: MemoryFactRow = facts.facts.into_iter().next().unwrap();
    }

    #[test]
    fn an_unrecognized_kind_folds_to_memory_unknown_not_an_error() {
        let events = vec![(1, Event::new("some_future_kind", Vec::new()))];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.unknown.len(), 1);
        assert_eq!(facts.unknown[0].kind, "some_future_kind");
    }

    #[test]
    fn a_fence_barrier_folds_with_no_payload() {
        let events = vec![(1, Event::new(KIND_MEMORY_FENCE_BARRIER, Vec::new()))];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(facts.fences.len(), 1);
    }

    #[test]
    fn a_repeated_position_refuses_the_fold() {
        let events = vec![
            (1, added("f1", MemoryScope::Direct, vec![])),
            (1, invalidated("f1")),
        ];
        assert!(matches!(
            fold_persona_memory(&events).unwrap_err(),
            PersonaMemoryFoldError::RepeatedPosition { position: 1 }
        ));
    }

    #[test]
    fn an_undecodable_known_kind_payload_refuses_the_fold() {
        let events = vec![(1, Event::new(kinds::MEMORY_ADDED, vec![0xff; 8]))];
        let error = fold_persona_memory(&events).unwrap_err();
        assert!(matches!(
            error,
            PersonaMemoryFoldError::UndecodableRecord { position: 1, .. }
        ));
    }

    #[test]
    fn the_turn_exec_id_normalizes_to_the_conversation_partitions_simple_hex_form() {
        let events = vec![(1, added("f1", MemoryScope::Direct, vec![]))];
        let facts = fold_persona_memory(&events).unwrap();
        assert_eq!(
            facts.facts[0].provenance.turn_id,
            "0195000000007000800000000000aaaa"
        );
        assert_eq!(
            facts.facts[0].provenance.turn_exec_id,
            "01950000-0000-7000-8000-00000000aaaa"
        );
    }

    #[test]
    fn the_same_prefix_folds_to_the_same_rows() {
        let events = vec![(
            1,
            added("f1", MemoryScope::Portable, vec!["tea".to_owned()]),
        )];
        assert_eq!(
            fold_persona_memory(&events).unwrap(),
            fold_persona_memory(&events).unwrap()
        );
    }
}