mempill_core/ports/
persistence.rs

1#![allow(missing_docs)]
2//! PersistencePort — INSERT-only, agent_id-first persistence abstraction.
3//!
4//! All methods take `agent_id` as the primary parameter (not a filter).
5//! Must enforce: single-writer per agent_id; append-only; atomic commit unit.
6
7use mempill_types::{
8    AgentId, Claim, ClaimEdge, ClaimRef, LedgerEntry, TransactionTime, ValidityAssertion,
9};
10
11/// An opaque transaction handle scoped to exactly one agent_id.
12/// No cross-agent transaction is possible (atomic commit unit is per-agent_id).
13pub trait Txn: Send + 'static {
14    fn agent_id(&self) -> &AgentId;
15}
16
17/// The persistence port — INSERT-only, agent_id-first.
18/// All methods take `agent_id` as the primary parameter (not a filter).
19/// Must enforce: single-writer per agent_id; append-only; atomic commit unit.
20pub trait PersistencePort: Send + Sync + 'static {
21    type Transaction: Txn;
22    type Error: std::error::Error + Send + Sync + 'static;
23
24    /// Begin an atomic unit scoped to one agent_id. No cross-agent transaction allowed.
25    fn begin_atomic(&self, agent_id: &AgentId) -> Result<Self::Transaction, Self::Error>;
26
27    fn append_claim(
28        &self,
29        txn: &mut Self::Transaction,
30        claim: &Claim,
31    ) -> Result<ClaimRef, Self::Error>;
32
33    fn append_validity_assertion(
34        &self,
35        txn: &mut Self::Transaction,
36        assertion: &ValidityAssertion,
37    ) -> Result<(), Self::Error>;
38
39    fn append_ledger_entry(
40        &self,
41        txn: &mut Self::Transaction,
42        entry: &LedgerEntry,
43    ) -> Result<(), Self::Error>;
44
45    fn append_claim_edge(
46        &self,
47        txn: &mut Self::Transaction,
48        edge: &ClaimEdge,
49    ) -> Result<(), Self::Error>;
50
51    fn commit(&self, txn: Self::Transaction) -> Result<(), Self::Error>;
52    fn rollback(&self, txn: Self::Transaction) -> Result<(), Self::Error>;
53
54    // ── Read operations (non-mutating w.r.t. belief and history — I1, I3) ──
55
56    /// Load all claims on the given (agent_id, subject, predicate) subject-line,
57    /// ordered by `tx_time ASC` (oldest first — callers fold in tx_time order).
58    ///
59    /// # Transaction-time cutoff (`as_of_tx_time`)
60    ///
61    /// When `as_of_tx_time` is `Some(T)`, only claims with `transaction_time <= T`
62    /// are returned. This enforces correct bi-temporal tx-time semantics: a claim
63    /// ingested after the query's as-of point did not exist at that point and must
64    /// not be visible to the fold.
65    ///
66    /// When `as_of_tx_time` is `None`, all claims are returned (current view). Use
67    /// `None` on all **write-path** callers (ingest, reconcile, adjudication) so
68    /// that incumbent detection always sees the full current state; passing `Some`
69    /// there would break conflict detection and succession.
70    fn load_subject_line(
71        &self,
72        agent_id: &AgentId,
73        subject: &str,
74        predicate: &str,
75        as_of_tx_time: Option<chrono::DateTime<chrono::Utc>>,
76    ) -> Result<Vec<Claim>, Self::Error>;
77
78    fn load_claim(
79        &self,
80        agent_id: &AgentId,
81        claim_ref: &ClaimRef,
82    ) -> Result<Option<Claim>, Self::Error>;
83
84    fn load_validity_assertions_for(
85        &self,
86        agent_id: &AgentId,
87        claim_ref: &ClaimRef,
88    ) -> Result<Vec<ValidityAssertion>, Self::Error>;
89
90    fn load_ledger(
91        &self,
92        agent_id: &AgentId,
93        from: Option<&TransactionTime>,
94        limit: usize,
95    ) -> Result<Vec<LedgerEntry>, Self::Error>;
96
97    /// Load ALL ledger entries for the given claim refs, with no row cap.
98    ///
99    /// Intended for the read path (query_memory / query_history): builds the
100    /// disposition map scoped to exactly the claims on a subject-line, avoiding
101    /// the agent-wide capped scan that caused silent wrong-belief at scale.
102    ///
103    /// # Empty input
104    ///
105    /// When `claim_refs` is empty this method MUST return `Ok(vec![])` immediately
106    /// without issuing any SQL (an empty `IN ()` clause is a syntax error on most
107    /// backends).
108    ///
109    /// # No row cap
110    ///
111    /// Unlike `load_ledger`, this method applies no `LIMIT`. Subject-lines are
112    /// small (typically 1–100 claims), so the result set is bounded naturally.
113    ///
114    /// # Transaction-time cutoff (`as_of_tx_time`)
115    ///
116    /// When `as_of_tx_time` is `Some(T)`, only ledger entries with
117    /// `recorded_at <= T` are returned. This is required for correct bi-temporal
118    /// tx-time travel: the disposition map must not see entries recorded after the
119    /// query's as-of point, otherwise a post-T supersession would incorrectly exclude
120    /// a claim that was still live at T.
121    ///
122    /// When `as_of_tx_time` is `None`, all entries are returned (preserves the
123    /// current behaviour for callers that want the full history or the latest view).
124    fn load_ledger_for_claims(
125        &self,
126        agent_id: &AgentId,
127        claim_refs: &[ClaimRef],
128        as_of_tx_time: Option<chrono::DateTime<chrono::Utc>>,
129    ) -> Result<Vec<LedgerEntry>, Self::Error>;
130
131    fn load_edges_for(
132        &self,
133        agent_id: &AgentId,
134        claim_ref: &ClaimRef,
135    ) -> Result<Vec<ClaimEdge>, Self::Error>;
136
137    /// Load the set of claims this agent served as injected context in the current session (for Amplification Guard entailment check).
138    fn load_injected_claims(
139        &self,
140        agent_id: &AgentId,
141    ) -> Result<Vec<ClaimRef>, Self::Error>;
142
143    /// Recursive CTE lineage traversal — returns the full `DerivedFrom` ancestry for a claim.
144    fn load_lineage(
145        &self,
146        agent_id: &AgentId,
147        claim_ref: &ClaimRef,
148    ) -> Result<Vec<ClaimEdge>, Self::Error>;
149
150    /// Return all distinct predicates stored for `(agent_id, subject)`.
151    ///
152    /// When `as_of_tx_time` is `Some(T)`, only predicates that have at least one claim
153    /// with `tx_time <= T` are returned (bi-temporal tx-time cutoff).  When `None`, all
154    /// predicates are returned (current view).
155    ///
156    /// The returned `Vec<String>` is NOT guaranteed to be sorted — the caller sorts if needed.
157    /// The result set is small (bounded by the number of distinct predicates per subject),
158    /// so no pagination is provided.
159    fn list_predicates_for_subject(
160        &self,
161        agent_id: &AgentId,
162        subject: &str,
163        as_of_tx_time: Option<chrono::DateTime<chrono::Utc>>,
164    ) -> Result<Vec<String>, Self::Error>;
165
166    /// Whether the store requires a global write serialization lock across all agent_ids.
167    ///
168    /// SQLite: true (single connection, no concurrent transactions possible).
169    /// Postgres: false (pool provides concurrent transactions; advisory lock per agent_id).
170    ///
171    /// EngineHandle consults this at write-path entry to decide whether to acquire
172    /// `store_write_lock`. Default = true (safe fallback for unknown adapters).
173    fn requires_global_write_serialization(&self) -> bool {
174        true
175    }
176}
mempill_core/ports/persistence.rs

mempill_core/ports/
persistence.rs
