feat(split): leakage-safe cutoff snapshots and relation co-partitions

CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to TEPP are documented here. The format follows Keep a Chang
 
 ### Added
 
+- `corpus_split` leakage-safe knowledge-cutoff snapshots, relation-connected co-partition groups, rolling-origin windows, and group-normalized ESS weight contracts.
 - `persistence_postgres` bitemporal foundation: multi-word migration contracts, knowledge-cutoff eligibility, and in-memory as-known-at / as-valid-at document replay (live SQLx/PostgreSQL execution remains accepted-target).
 - `event_core` mention/instance separation with explicit promotion, typed roles, event-time validity, and fail-closed mention-as-instance refusal.
 - `membership_core` time-varying weighted multiple-membership network with contextual roles, event-time validity, and atomistic-fallacy prevention contracts.

crates/corpus_split/Cargo.toml

@@ -13,5 +13,12 @@ keywords.workspace = true
 categories.workspace = true
 publish = false
 
+[dependencies]
+temporal_core = { path = "../temporal_core", version = "0.1.0" }
+uuid.workspace = true
+
+[dev-dependencies]
+uuid.workspace = true
+
 [lints]
 workspace = true

crates/corpus_split/src/connected_group.rs

@@ -0,0 +1,214 @@
+//! Relation-connected leakage groups.
+
+use crate::CorpusSplitError;
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+use uuid::Uuid;
+
+/// Governed link kinds that force co-partitioning.
+#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub enum LeakageLinkKind {
+    /// Document revision of another document.
+    Revision,
+    /// Translation of another document.
+    Translation,
+    /// Copied or template-derived variant.
+    CopiedVariant,
+    /// Shared event episode membership.
+    SameEpisode,
+}
+
+/// Undirected leakage link between two document identities.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct LeakageLink {
+    /// First endpoint.
+    pub left: Uuid,
+    /// Second endpoint.
+    pub right: Uuid,
+    /// Governed link kind.
+    pub kind: LeakageLinkKind,
+}
+
+/// Union of documents that must remain in one split partition.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ConnectedGroup {
+    members: BTreeSet<Uuid>,
+}
+
+impl ConnectedGroup {
+    /// Borrow member identities.
+    #[must_use]
+    pub fn members(&self) -> &BTreeSet<Uuid> {
+        &self.members
+    }
+
+    /// Return the number of members.
+    #[must_use]
+    pub fn len(&self) -> usize {
+        self.members.len()
+    }
+
+    /// Return whether the group is empty.
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.members.is_empty()
+    }
+}
+
+/// Build connected components over leakage links restricted to `universe`.
+#[must_use]
+pub fn build_connected_groups(universe: &[Uuid], links: &[LeakageLink]) -> Vec<ConnectedGroup> {
+    let allowed: BTreeSet<Uuid> = universe.iter().copied().collect();
+    let mut adjacency: BTreeMap<Uuid, BTreeSet<Uuid>> = BTreeMap::new();
+    for id in &allowed {
+        adjacency.entry(*id).or_default();
+    }
+    for link in links {
+        if !allowed.contains(&link.left) || !allowed.contains(&link.right) {
+            continue;
+        }
+        if link.left == link.right {
+            continue;
+        }
+        adjacency.entry(link.left).or_default().insert(link.right);
+        adjacency.entry(link.right).or_default().insert(link.left);
+    }
+
+    let mut seen = BTreeSet::new();
+    let mut groups = Vec::new();
+    for start in universe {
+        if !seen.insert(*start) {
+            continue;
+        }
+        let mut members = BTreeSet::new();
+        let mut queue = VecDeque::from([*start]);
+        while let Some(node) = queue.pop_front() {
+            // Each node is enqueued at most once via `seen`, so membership insert always succeeds.
+            members.insert(node);
+            // Every universe identity is pre-inserted into `adjacency`.
+            for neighbor in &adjacency[&node] {
+                if seen.insert(*neighbor) {
+                    queue.push_back(*neighbor);
+                }
+            }
+        }
+        groups.push(ConnectedGroup { members });
+    }
+    groups
+}
+
+/// Reject a proposed partition map that separates any connected group.
+///
+/// # Errors
+///
+/// Returns [`CorpusSplitError::RelationLeakage`] when linked members diverge.
+pub fn assert_no_group_leakage(
+    groups: &[ConnectedGroup],
+    partition_by_document: &BTreeMap<Uuid, u8>,
+) -> Result<(), CorpusSplitError> {
+    for group in groups {
+        let mut partition = None;
+        for member in &group.members {
+            let Some(assigned) = partition_by_document.get(member) else {
+                continue;
+            };
+            match partition {
+                None => partition = Some(*assigned),
+                Some(existing) if existing == *assigned => {}
+                Some(_) => return Err(CorpusSplitError::RelationLeakage),
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        ConnectedGroup, LeakageLink, LeakageLinkKind, assert_no_group_leakage,
+        build_connected_groups,
+    };
+    use crate::CorpusSplitError;
+    use std::collections::{BTreeMap, BTreeSet};
+    use uuid::Uuid;
+
+    #[test]
+    fn revisions_and_translations_form_one_group() {
+        let a = Uuid::now_v7();
+        let b = Uuid::now_v7();
+        let c = Uuid::now_v7();
+        let groups = build_connected_groups(
+            &[a, b, c],
+            &[
+                LeakageLink {
+                    left: a,
+                    right: b,
+                    kind: LeakageLinkKind::Revision,
+                },
+                LeakageLink {
+                    left: b,
+                    right: c,
+                    kind: LeakageLinkKind::Translation,
+                },
+            ],
+        );
+        assert_eq!(groups.len(), 1);
+        assert_eq!(groups[0].len(), 3);
+        assert!(!groups[0].is_empty());
+        assert_eq!(groups[0].members(), &BTreeSet::from([a, b, c]));
+    }
+
+    #[test]
+    fn cross_partition_assignment_is_rejected() {
+        let a = Uuid::now_v7();
+        let b = Uuid::now_v7();
+        let groups = build_connected_groups(
+            &[a, b],
+            &[LeakageLink {
+                left: a,
+                right: b,
+                kind: LeakageLinkKind::CopiedVariant,
+            }],
+        );
+        let mut map = BTreeMap::new();
+        map.insert(a, 0);
+        map.insert(b, 1);
+        assert_eq!(
+            assert_no_group_leakage(&groups, &map),
+            Err(CorpusSplitError::RelationLeakage)
+        );
+        map.insert(b, 0);
+        assert_no_group_leakage(&groups, &map).expect("same partition");
+        let empty = ConnectedGroup {
+            members: BTreeSet::new(),
+        };
+        assert!(empty.is_empty());
+        // Self-links and out-of-universe endpoints are ignored.
+        let lonely = Uuid::now_v7();
+        let outside = Uuid::now_v7();
+        let isolated = build_connected_groups(
+            &[lonely],
+            &[
+                LeakageLink {
+                    left: lonely,
+                    right: lonely,
+                    kind: LeakageLinkKind::Revision,
+                },
+                LeakageLink {
+                    left: lonely,
+                    right: outside,
+                    kind: LeakageLinkKind::Translation,
+                },
+                LeakageLink {
+                    left: outside,
+                    right: lonely,
+                    kind: LeakageLinkKind::CopiedVariant,
+                },
+            ],
+        );
+        assert_eq!(isolated.len(), 1);
+        assert_eq!(isolated[0].len(), 1);
+        // Unassigned members do not induce leakage.
+        let partial = BTreeMap::from([(a, 0_u8)]);
+        assert_no_group_leakage(&groups, &partial).expect("partial assignment");
+    }
+}

crates/corpus_split/src/document.rs

@@ -0,0 +1,24 @@
+//! Document observations eligible for cutoff-aware snapshots.
+
+use temporal_core::AvailableTime;
+use uuid::Uuid;
+
+/// One document observation with availability provenance.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CorpusDocument {
+    /// Stable analytical document identity.
+    pub document_id: Uuid,
+    /// When the document became available as evidence.
+    pub available_time: AvailableTime,
+}
+
+impl CorpusDocument {
+    /// Construct a document observation.
+    #[must_use]
+    pub const fn new(document_id: Uuid, available_time: AvailableTime) -> Self {
+        Self {
+            document_id,
+            available_time,
+        }
+    }
+}

crates/corpus_split/src/error.rs

@@ -0,0 +1,56 @@
+//! Fail-closed corpus-split errors.
+
+use std::fmt;
+
+/// A fail-closed corpus-split domain error.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[non_exhaustive]
+pub enum CorpusSplitError {
+    /// A partition assignment would separate linked records.
+    RelationLeakage,
+    /// A document was unavailable at the requested knowledge cutoff.
+    UnavailableAtCutoff,
+    /// A duplicate document identity was rejected.
+    DuplicateDocumentIdentity,
+    /// Split proportions or seeds were invalid.
+    InvalidSplitConfiguration,
+}
+
+impl fmt::Display for CorpusSplitError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let message = match self {
+            Self::RelationLeakage => "relation-aware split leakage",
+            Self::UnavailableAtCutoff => "document unavailable at knowledge cutoff",
+            Self::DuplicateDocumentIdentity => "duplicate document identity",
+            Self::InvalidSplitConfiguration => "invalid split configuration",
+        };
+        formatter.write_str(message)
+    }
+}
+
+impl std::error::Error for CorpusSplitError {}
+
+#[cfg(test)]
+mod tests {
+    use super::CorpusSplitError;
+
+    #[test]
+    fn messages_are_stable() {
+        assert_eq!(
+            CorpusSplitError::RelationLeakage.to_string(),
+            "relation-aware split leakage"
+        );
+        assert_eq!(
+            CorpusSplitError::UnavailableAtCutoff.to_string(),
+            "document unavailable at knowledge cutoff"
+        );
+        assert_eq!(
+            CorpusSplitError::DuplicateDocumentIdentity.to_string(),
+            "duplicate document identity"
+        );
+        assert_eq!(
+            CorpusSplitError::InvalidSplitConfiguration.to_string(),
+            "invalid split configuration"
+        );
+    }
+}

crates/corpus_split/src/lib.rs

@@ -2,5 +2,66 @@
 #![deny(missing_docs)]
 //! Leakage-safe corpus snapshots and relation-aware data splits.
 //!
-//! This crate intentionally exposes no production behavior in the workspace-foundation
-//! slice. Domain APIs are introduced test-first in the corresponding implementation task.
+//! TEPP historical analyses may only consume documents whose availability time
+//! does not exceed the declared knowledge cutoff. Train/validation/test
+//! partitions keep revisions, translations, copied variants, and shared
+//! episodes co-located so relation-aware leakage cannot invent independence.
+
+mod connected_group;
+mod document;
+mod error;
+mod rolling_origin;
+mod snapshot;
+mod weights;
+
+use temporal_core::{AvailableTime, KnowledgeCutoff};
+
+/// Knowledge-cutoff eligibility for snapshot membership.
+#[must_use]
+pub fn cutoff_eligible(available_time: &AvailableTime, knowledge_cutoff: &KnowledgeCutoff) -> bool {
+    available_time.instant() <= knowledge_cutoff.instant()
+}
+
+/// Relation-connected leakage group.
+pub use connected_group::ConnectedGroup;
+/// Undirected leakage link.
+pub use connected_group::LeakageLink;
+/// Governed leakage link vocabulary.
+pub use connected_group::LeakageLinkKind;
+/// Reject partitions that separate linked groups.
+pub use connected_group::assert_no_group_leakage;
+/// Build connected components over leakage links.
+pub use connected_group::build_connected_groups;
+/// Document observation with availability provenance.
+pub use document::CorpusDocument;
+/// Fail-closed corpus-split errors.
+pub use error::CorpusSplitError;
+/// Rolling-origin train/test window.
+pub use rolling_origin::RollingOriginWindow;
+/// Build ordered rolling-origin windows.
+pub use rolling_origin::rolling_origin_windows;
+/// Cutoff-filtered corpus snapshot.
+pub use snapshot::CorpusSnapshot;
+/// Kish effective sample size.
+pub use weights::effective_sample_size;
+/// Group-normalized observation weights.
+pub use weights::group_normalized_weights;
+
+#[cfg(test)]
+mod tests {
+    use super::cutoff_eligible;
+    use temporal_core::{AvailableTime, KnowledgeCutoff};
+
+    #[test]
+    fn cutoff_boundary_is_inclusive() {
+        let stamp = "2026-05-01T00:00:00Z";
+        assert!(cutoff_eligible(
+            &AvailableTime::parse_rfc3339(stamp).expect("a"),
+            &KnowledgeCutoff::parse_rfc3339(stamp).expect("c")
+        ));
+        assert!(!cutoff_eligible(
+            &AvailableTime::parse_rfc3339("2026-05-02T00:00:00Z").expect("a"),
+            &KnowledgeCutoff::parse_rfc3339(stamp).expect("c")
+        ));
+    }
+}