perspt_store/store/
psp9_ledger.rs

1use super::*;
2
3/// One durable PSP-9 ledger record (system 14). The SDK's `Ledger` owns the
4/// chain semantics; this row keeps the canonical bytes for resume and audit
5/// replay.
6#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
7pub struct Psp9LedgerRow {
8    pub session_id: String,
9    pub sequence: i64,
10    pub event_json: String,
11    pub prev_hash: String,
12    pub hash: String,
13}
14
15#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
16pub struct Psp9VerdictRow {
17    pub session_id: String,
18    pub candidate_id: String,
19    pub validator_id: String,
20    pub stratum: String,
21    pub missed: bool,
22    pub unsafe_label: Option<bool>,
23    pub evidence_hash: String,
24}
25
26#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
27pub struct Psp9CalibrationEpochRow {
28    pub epoch_id: String,
29    pub stratum: String,
30    pub target_rho: f64,
31    pub threshold: Option<f64>,
32    pub state: String,
33    pub sample_count: i64,
34}
35
36/// (sample_id, score, delayed unsafe label, audit_selected).
37pub type Psp9SampleRow = (String, f64, Option<bool>, bool);
38
39#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
40pub struct Psp9ExternalEffectRow {
41    pub idempotency_key: String,
42    pub intent_hash: String,
43    pub intent_json: String,
44    pub result_json: Option<String>,
45    pub status: String,
46}
47
48impl SessionStore {
49    pub fn put_psp9_artifact(
50        &self,
51        content_hash: &str,
52        content: &[u8],
53        media_type: &str,
54    ) -> Result<()> {
55        self.conn.lock().unwrap().execute(
56            "INSERT INTO psp9_artifacts (content_hash, content, byte_len, media_type) \
57             VALUES (?, ?, ?, ?) ON CONFLICT (content_hash) DO NOTHING",
58            duckdb::params![content_hash, content, content.len() as i64, media_type],
59        )?;
60        Ok(())
61    }
62
63    pub fn get_psp9_artifact(&self, content_hash: &str) -> Result<Option<Vec<u8>>> {
64        let conn = self.conn.lock().unwrap();
65        let mut statement =
66            conn.prepare("SELECT content FROM psp9_artifacts WHERE content_hash = ?")?;
67        let mut rows = statement.query([content_hash])?;
68        Ok(rows.next()?.map(|row| row.get(0)).transpose()?)
69    }
70
71    pub fn record_psp9_checkpoint(
72        &self,
73        session_id: &str,
74        covered_event_root: &str,
75        checkpoint_json: &str,
76    ) -> Result<()> {
77        self.conn.lock().unwrap().execute(
78            "INSERT INTO psp9_context_checkpoints \
79             (session_id, covered_event_root, checkpoint_json) VALUES (?, ?, ?) \
80             ON CONFLICT (session_id, covered_event_root) DO NOTHING",
81            duckdb::params![session_id, covered_event_root, checkpoint_json],
82        )?;
83        Ok(())
84    }
85
86    /// The newest checkpoint JSON for a session, if any.
87    pub fn latest_psp9_checkpoint(&self, session_id: &str) -> Result<Option<String>> {
88        let conn = self.conn.lock().unwrap();
89        let mut statement = conn.prepare(
90            "SELECT checkpoint_json FROM psp9_context_checkpoints WHERE session_id = ? \
91             ORDER BY created_at DESC, covered_event_root DESC LIMIT 1",
92        )?;
93        let mut rows = statement.query([session_id])?;
94        Ok(rows.next()?.map(|row| row.get(0)).transpose()?)
95    }
96
97    pub fn record_psp9_verdict(&self, row: &Psp9VerdictRow) -> Result<()> {
98        self.conn.lock().unwrap().execute(
99            "INSERT INTO psp9_verdicts (session_id, candidate_id, validator_id, stratum, \
100             missed, unsafe_label, evidence_hash) VALUES (?, ?, ?, ?, ?, ?, ?) \
101             ON CONFLICT (session_id, candidate_id, validator_id) DO UPDATE SET \
102             unsafe_label = excluded.unsafe_label, evidence_hash = excluded.evidence_hash",
103            duckdb::params![
104                row.session_id,
105                row.candidate_id,
106                row.validator_id,
107                row.stratum,
108                row.missed,
109                row.unsafe_label,
110                row.evidence_hash
111            ],
112        )?;
113        Ok(())
114    }
115
116    /// Per-validator verdict history for one session, in insertion order.
117    pub fn get_psp9_verdicts(&self, session_id: &str) -> Result<Vec<Psp9VerdictRow>> {
118        let conn = self.conn.lock().unwrap();
119        let mut statement = conn.prepare(
120            "SELECT session_id, candidate_id, validator_id, stratum, missed, \
121             unsafe_label, evidence_hash FROM psp9_verdicts WHERE session_id = ?",
122        )?;
123        let rows = statement.query_map([session_id], |row| {
124            Ok(Psp9VerdictRow {
125                session_id: row.get(0)?,
126                candidate_id: row.get(1)?,
127                validator_id: row.get(2)?,
128                stratum: row.get(3)?,
129                missed: row.get(4)?,
130                unsafe_label: row.get(5)?,
131                evidence_hash: row.get(6)?,
132            })
133        })?;
134        rows.collect::<std::result::Result<Vec<_>, _>>()
135            .map_err(Into::into)
136    }
137
138    pub fn record_psp9_calibration_epoch(&self, row: &Psp9CalibrationEpochRow) -> Result<()> {
139        self.conn.lock().unwrap().execute(
140            "INSERT INTO psp9_calibration_epochs \
141             (epoch_id, stratum, target_rho, threshold, state, sample_count) \
142             VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (epoch_id) DO NOTHING",
143            duckdb::params![
144                row.epoch_id,
145                row.stratum,
146                row.target_rho,
147                row.threshold,
148                row.state,
149                row.sample_count
150            ],
151        )?;
152        Ok(())
153    }
154
155    /// Return the newest immutable epoch for exactly one serialized stratum.
156    /// A caller must never reuse an epoch across a changed stratum.
157    pub fn latest_psp9_calibration_epoch(
158        &self,
159        stratum: &str,
160    ) -> Result<Option<Psp9CalibrationEpochRow>> {
161        let conn = self.conn.lock().unwrap();
162        let mut statement = conn.prepare(
163            "SELECT epoch_id, stratum, target_rho, threshold, state, sample_count \
164             FROM psp9_calibration_epochs WHERE stratum = ? \
165             ORDER BY created_at DESC, epoch_id DESC LIMIT 1",
166        )?;
167        let mut rows = statement.query([stratum])?;
168        rows.next()?
169            .map(|row| {
170                Ok(Psp9CalibrationEpochRow {
171                    epoch_id: row.get(0)?,
172                    stratum: row.get(1)?,
173                    target_rho: row.get(2)?,
174                    threshold: row.get(3)?,
175                    state: row.get(4)?,
176                    sample_count: row.get(5)?,
177                })
178            })
179            .transpose()
180    }
181
182    pub fn psp9_calibration_samples(&self, epoch_id: &str) -> Result<Vec<Psp9SampleRow>> {
183        let conn = self.conn.lock().unwrap();
184        let mut statement = conn.prepare(
185            "SELECT sample_id, score, unsafe_label, audit_selected \
186             FROM psp9_calibration_samples WHERE epoch_id = ? ORDER BY created_at, sample_id",
187        )?;
188        let rows = statement.query_map([epoch_id], |row| {
189            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
190        })?;
191        rows.collect::<std::result::Result<Vec<_>, _>>()
192            .map_err(Into::into)
193    }
194
195    pub fn record_psp9_calibration_sample(
196        &self,
197        epoch_id: &str,
198        sample_id: &str,
199        score: f64,
200        unsafe_label: Option<bool>,
201        audit_selected: bool,
202    ) -> Result<()> {
203        self.conn.lock().unwrap().execute(
204            "INSERT INTO psp9_calibration_samples \
205             (epoch_id, sample_id, score, unsafe_label, audit_selected) VALUES (?, ?, ?, ?, ?) \
206             ON CONFLICT (epoch_id, sample_id) DO NOTHING",
207            duckdb::params![epoch_id, sample_id, score, unsafe_label, audit_selected],
208        )?;
209        Ok(())
210    }
211
212    /// Ingest one delayed audit label. Returns how many rows were labeled
213    /// (0 when the sample is unknown or already labeled — labels are
214    /// single-assignment, matching the ledger's write-once discipline).
215    pub fn label_psp9_calibration_sample(&self, sample_id: &str, is_unsafe: bool) -> Result<usize> {
216        let updated = self.conn.lock().unwrap().execute(
217            "UPDATE psp9_calibration_samples SET unsafe_label = ? \
218             WHERE sample_id = ? AND unsafe_label IS NULL",
219            duckdb::params![is_unsafe, sample_id],
220        )?;
221        Ok(updated)
222    }
223
224    /// Single-assignment delayed label on every verdict row for a
225    /// candidate — the same oracle labels calibration samples and validator
226    /// verdicts in one pass (PSP-9 system 8).
227    pub fn label_psp9_verdicts(&self, candidate_id: &str, is_unsafe: bool) -> Result<usize> {
228        let updated = self.conn.lock().unwrap().execute(
229            "UPDATE psp9_verdicts SET unsafe_label = ? \
230             WHERE candidate_id = ? AND unsafe_label IS NULL",
231            duckdb::params![is_unsafe, candidate_id],
232        )?;
233        Ok(updated)
234    }
235
236    /// Labeled verdict rows across all sessions, for the independence
237    /// estimator (matched strata are joined by candidate id downstream).
238    pub fn labeled_psp9_verdicts(&self) -> Result<Vec<Psp9VerdictRow>> {
239        let conn = self.conn.lock().unwrap();
240        let mut statement = conn.prepare(
241            "SELECT session_id, candidate_id, validator_id, stratum, missed, \
242             unsafe_label, evidence_hash FROM psp9_verdicts \
243             WHERE unsafe_label IS NOT NULL ORDER BY candidate_id, validator_id",
244        )?;
245        let rows = statement.query_map([], |row| {
246            Ok(Psp9VerdictRow {
247                session_id: row.get(0)?,
248                candidate_id: row.get(1)?,
249                validator_id: row.get(2)?,
250                stratum: row.get(3)?,
251                missed: row.get(4)?,
252                unsafe_label: row.get(5)?,
253                evidence_hash: row.get(6)?,
254            })
255        })?;
256        Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
257    }
258
259    /// Audit-selected samples still waiting for their delayed label.
260    pub fn pending_psp9_audit_samples(&self, limit: usize) -> Result<Vec<(String, String, f64)>> {
261        let conn = self.conn.lock().unwrap();
262        let mut statement = conn.prepare(
263            "SELECT epoch_id, sample_id, score FROM psp9_calibration_samples \
264             WHERE audit_selected AND unsafe_label IS NULL \
265             ORDER BY created_at LIMIT ?",
266        )?;
267        let rows =
268            statement.query_map([limit], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?;
269        rows.collect::<std::result::Result<Vec<_>, _>>()
270            .map_err(Into::into)
271    }
272
273    /// Every labeled sample across all epochs of one serialized stratum, for
274    /// threshold recomputation. Unlabeled samples never count toward the
275    /// conformal floor.
276    pub fn labeled_psp9_samples_for_stratum(&self, stratum: &str) -> Result<Vec<(f64, bool)>> {
277        let conn = self.conn.lock().unwrap();
278        let mut statement = conn.prepare(
279            "SELECT s.score, s.unsafe_label \
280             FROM psp9_calibration_samples s \
281             JOIN psp9_calibration_epochs e ON e.epoch_id = s.epoch_id \
282             WHERE e.stratum = ? AND s.unsafe_label IS NOT NULL \
283             ORDER BY s.created_at, s.sample_id",
284        )?;
285        let rows = statement.query_map([stratum], |row| Ok((row.get(0)?, row.get(1)?)))?;
286        rows.collect::<std::result::Result<Vec<_>, _>>()
287            .map_err(Into::into)
288    }
289
290    /// Newest-first calibration epochs across all strata.
291    pub fn all_psp9_calibration_epochs(
292        &self,
293        limit: usize,
294    ) -> Result<Vec<Psp9CalibrationEpochRow>> {
295        let conn = self.conn.lock().unwrap();
296        let mut statement = conn.prepare(
297            "SELECT epoch_id, stratum, target_rho, threshold, state, sample_count \
298             FROM psp9_calibration_epochs ORDER BY created_at DESC, epoch_id DESC LIMIT ?",
299        )?;
300        let rows = statement.query_map([limit], |row| {
301            Ok(Psp9CalibrationEpochRow {
302                epoch_id: row.get(0)?,
303                stratum: row.get(1)?,
304                target_rho: row.get(2)?,
305                threshold: row.get(3)?,
306                state: row.get(4)?,
307                sample_count: row.get(5)?,
308            })
309        })?;
310        rows.collect::<std::result::Result<Vec<_>, _>>()
311            .map_err(Into::into)
312    }
313
314    /// The stratum an epoch belongs to.
315    pub fn psp9_epoch_stratum(&self, epoch_id: &str) -> Result<Option<String>> {
316        let conn = self.conn.lock().unwrap();
317        let mut statement =
318            conn.prepare("SELECT stratum FROM psp9_calibration_epochs WHERE epoch_id = ?")?;
319        let mut rows = statement.query([epoch_id])?;
320        Ok(rows.next()?.map(|row| row.get(0)).transpose()?)
321    }
322
323    /// Append one PSP-9 ledger record durably.
324    pub fn record_psp9_event(&self, row: &Psp9LedgerRow) -> Result<()> {
325        self.conn.lock().unwrap().execute(
326            "INSERT INTO psp9_ledger_events (session_id, sequence, event_json, prev_hash, hash) \
327             VALUES (?, ?, ?, ?, ?)",
328            duckdb::params![
329                row.session_id,
330                row.sequence,
331                row.event_json,
332                row.prev_hash,
333                row.hash
334            ],
335        )?;
336        Ok(())
337    }
338
339    /// Load a session's PSP-9 event stream in sequence order.
340    pub fn get_psp9_events(&self, session_id: &str) -> Result<Vec<Psp9LedgerRow>> {
341        let conn = self.conn.lock().unwrap();
342        let mut stmt = conn.prepare(
343            "SELECT session_id, sequence, event_json, prev_hash, hash \
344             FROM psp9_ledger_events WHERE session_id = ? ORDER BY sequence",
345        )?;
346        let rows = stmt.query_map([session_id], |row| {
347            Ok(Psp9LedgerRow {
348                session_id: row.get(0)?,
349                sequence: row.get(1)?,
350                event_json: row.get(2)?,
351                prev_hash: row.get(3)?,
352                hash: row.get(4)?,
353            })
354        })?;
355        let mut results = Vec::new();
356        for row in rows {
357            results.push(row?);
358        }
359        Ok(results)
360    }
361
362    pub fn initialize_authority_epoch(&self, session_id: &str, epoch: u64) -> Result<()> {
363        self.conn.lock().unwrap().execute(
364            "INSERT INTO psp9_authority_epochs (session_id, epoch) VALUES (?, ?) \
365             ON CONFLICT (session_id) DO NOTHING",
366            duckdb::params![session_id, epoch as i64],
367        )?;
368        Ok(())
369    }
370
371    pub fn authority_epoch(&self, session_id: &str) -> Result<u64> {
372        let conn = self.conn.lock().unwrap();
373        let mut statement =
374            conn.prepare("SELECT epoch FROM psp9_authority_epochs WHERE session_id = ?")?;
375        let mut rows = statement.query([session_id])?;
376        let row = rows.next()?.context("authority epoch is not initialized")?;
377        let epoch: i64 = row.get(0)?;
378        Ok(epoch as u64)
379    }
380
381    /// Execute one synchronous durable effect while holding the database's
382    /// authority-epoch write lock. Revocation uses the same row and therefore
383    /// cannot commit between the epoch check and `operation`.
384    pub fn with_authority_epoch<T>(
385        &self,
386        session_id: &str,
387        expected_epoch: u64,
388        operation: impl FnOnce() -> Result<T>,
389    ) -> Result<T> {
390        let mut conn = self.conn.lock().unwrap();
391        let transaction = conn.transaction()?;
392        let updated = transaction.execute(
393            "UPDATE psp9_authority_epochs SET epoch = epoch WHERE session_id = ?",
394            [session_id],
395        )?;
396        anyhow::ensure!(updated == 1, "authority epoch is not initialized");
397        let epoch: i64 = transaction.query_row(
398            "SELECT epoch FROM psp9_authority_epochs WHERE session_id = ?",
399            [session_id],
400            |row| row.get(0),
401        )?;
402        anyhow::ensure!(
403            epoch as u64 == expected_epoch,
404            "authority epoch changed before durable effect: expected {expected_epoch}, found {epoch}"
405        );
406        let output = operation()?;
407        transaction.commit()?;
408        Ok(output)
409    }
410
411    pub fn revoke_authority(&self, session_id: &str) -> Result<u64> {
412        let mut conn = self.conn.lock().unwrap();
413        let transaction = conn.transaction()?;
414        transaction.execute(
415            "UPDATE psp9_authority_epochs SET epoch = epoch + 1 WHERE session_id = ?",
416            [session_id],
417        )?;
418        let epoch: i64 = transaction.query_row(
419            "SELECT epoch FROM psp9_authority_epochs WHERE session_id = ?",
420            [session_id],
421            |row| row.get(0),
422        )?;
423        transaction.commit()?;
424        Ok(epoch as u64)
425    }
426
427    pub fn record_grant_policy(
428        &self,
429        session_id: &str,
430        policy_id: &str,
431        policy_json: &str,
432    ) -> Result<()> {
433        self.conn.lock().unwrap().execute(
434            "INSERT INTO psp9_grant_policies (policy_id, session_id, policy_json) \
435             VALUES (?, ?, ?)",
436            duckdb::params![policy_id, session_id, policy_json],
437        )?;
438        Ok(())
439    }
440
441    /// The newest persisted grant-policy JSON for a session, if any.
442    pub fn get_grant_policy(&self, session_id: &str) -> Result<Option<String>> {
443        let conn = self.conn.lock().unwrap();
444        let mut statement = conn.prepare(
445            "SELECT policy_json FROM psp9_grant_policies WHERE session_id = ? \
446             ORDER BY created_at DESC, policy_id DESC LIMIT 1",
447        )?;
448        let mut rows = statement.query([session_id])?;
449        Ok(rows.next()?.map(|row| row.get(0)).transpose()?)
450    }
451
452    /// R1/R5: durably single-assign an external-effect intent before the
453    /// filesystem or network is touched. Redelivery with identical bytes is
454    /// idempotent; key reuse with different content is rejected.
455    pub fn record_external_effect_intent(
456        &self,
457        session_id: &str,
458        idempotency_key: &str,
459        intent_hash: &str,
460        intent_json: &str,
461    ) -> Result<()> {
462        let mut conn = self.conn.lock().unwrap();
463        let transaction = conn.transaction()?;
464        let existing: Option<String> = transaction
465            .query_row(
466                "SELECT intent_hash FROM psp9_external_effects \
467                 WHERE session_id = ? AND idempotency_key = ?",
468                duckdb::params![session_id, idempotency_key],
469                |row| row.get(0),
470            )
471            .ok();
472        match existing {
473            Some(hash) if hash == intent_hash => {}
474            Some(_) => anyhow::bail!("idempotency key reused for different external effect"),
475            None => {
476                transaction.execute(
477                    "INSERT INTO psp9_external_effects \
478                     (session_id, idempotency_key, intent_hash, intent_json, status) \
479                     VALUES (?, ?, ?, ?, 'INTENT_RECORDED')",
480                    duckdb::params![session_id, idempotency_key, intent_hash, intent_json],
481                )?;
482            }
483        }
484        transaction.commit()?;
485        Ok(())
486    }
487
488    pub fn complete_external_effect(
489        &self,
490        session_id: &str,
491        idempotency_key: &str,
492        result_json: &str,
493    ) -> Result<()> {
494        let updated = self.conn.lock().unwrap().execute(
495            "UPDATE psp9_external_effects SET result_json = ?, status = 'COMPLETED', \
496             completed_at = CURRENT_TIMESTAMP \
497             WHERE session_id = ? AND idempotency_key = ? AND status = 'INTENT_RECORDED'",
498            duckdb::params![result_json, session_id, idempotency_key],
499        )?;
500        anyhow::ensure!(
501            updated == 1,
502            "external effect has no pending write-ahead intent"
503        );
504        Ok(())
505    }
506
507    /// Completed promotion intents, newest first — the rollback surface.
508    pub fn completed_external_effects(
509        &self,
510        session_id: &str,
511    ) -> Result<Vec<Psp9ExternalEffectRow>> {
512        let conn = self.conn.lock().unwrap();
513        let mut statement = conn.prepare(
514            "SELECT idempotency_key, intent_hash, intent_json, result_json, status \
515             FROM psp9_external_effects WHERE session_id = ? AND status = 'COMPLETED' \
516             ORDER BY created_at DESC",
517        )?;
518        let rows = statement.query_map([session_id], |row| {
519            Ok(Psp9ExternalEffectRow {
520                idempotency_key: row.get(0)?,
521                intent_hash: row.get(1)?,
522                intent_json: row.get(2)?,
523                result_json: row.get(3)?,
524                status: row.get(4)?,
525            })
526        })?;
527        rows.collect::<std::result::Result<Vec<_>, _>>()
528            .map_err(Into::into)
529    }
530
531    pub fn pending_external_effects(&self, session_id: &str) -> Result<Vec<Psp9ExternalEffectRow>> {
532        let conn = self.conn.lock().unwrap();
533        let mut statement = conn.prepare(
534            "SELECT idempotency_key, intent_hash, intent_json, result_json, status \
535             FROM psp9_external_effects WHERE session_id = ? AND status != 'COMPLETED' \
536             ORDER BY created_at",
537        )?;
538        let rows = statement.query_map([session_id], |row| {
539            Ok(Psp9ExternalEffectRow {
540                idempotency_key: row.get(0)?,
541                intent_hash: row.get(1)?,
542                intent_json: row.get(2)?,
543                result_json: row.get(3)?,
544                status: row.get(4)?,
545            })
546        })?;
547        rows.collect::<std::result::Result<Vec<_>, _>>()
548            .map_err(Into::into)
549    }
550}
551
552#[cfg(test)]
553mod tests {
554    use super::*;
555
556    fn scratch_store() -> (std::path::PathBuf, SessionStore) {
557        let dir = std::env::temp_dir().join(format!("perspt_p9_{}", uuid::Uuid::new_v4()));
558        std::fs::create_dir_all(&dir).unwrap();
559        let store = SessionStore::open(&dir.join("t.db")).unwrap();
560        (dir, store)
561    }
562
563    #[test]
564    fn psp9_events_round_trip_in_order() {
565        let (dir, store) = scratch_store();
566        for sequence in 0..3i64 {
567            store
568                .record_psp9_event(&Psp9LedgerRow {
569                    session_id: "s1".into(),
570                    sequence,
571                    event_json: format!("{{\"seq\":{sequence}}}"),
572                    prev_hash: format!("h{}", sequence.saturating_sub(1)),
573                    hash: format!("h{sequence}"),
574                })
575                .unwrap();
576        }
577        let rows = store.get_psp9_events("s1").unwrap();
578        assert_eq!(rows.len(), 3);
579        assert_eq!(rows[2].hash, "h2");
580        assert!(store.get_psp9_events("other").unwrap().is_empty());
581        let _ = std::fs::remove_dir_all(&dir);
582    }
583
584    #[test]
585    fn external_effects_are_single_assignment() {
586        let (dir, store) = scratch_store();
587        store
588            .record_external_effect_intent("s1", "promote:n1", "h1", "{\"a\":1}")
589            .unwrap();
590        store
591            .record_external_effect_intent("s1", "promote:n1", "h1", "{\"a\":1}")
592            .unwrap();
593        assert!(store
594            .record_external_effect_intent("s1", "promote:n1", "h2", "{\"a\":2}")
595            .is_err());
596        store
597            .complete_external_effect("s1", "promote:n1", "{\"ok\":true}")
598            .unwrap();
599        assert!(store.pending_external_effects("s1").unwrap().is_empty());
600        let _ = std::fs::remove_dir_all(&dir);
601    }
602
603    #[test]
604    fn artifacts_verdicts_and_calibration_rows_persist() {
605        let (dir, store) = scratch_store();
606        let bytes = b"complete compiler output";
607        let hash = "artifact-hash".to_string();
608        store.put_psp9_artifact(&hash, bytes, "text/plain").unwrap();
609        assert_eq!(store.get_psp9_artifact(&hash).unwrap().unwrap(), bytes);
610        store
611            .record_psp9_checkpoint("s1", "root-1", "{\"control\":{}}")
612            .unwrap();
613        store
614            .record_psp9_verdict(&Psp9VerdictRow {
615                session_id: "s1".into(),
616                candidate_id: "c1".into(),
617                validator_id: "cargo-test".into(),
618                stratum: "rust:test:v1".into(),
619                missed: false,
620                unsafe_label: Some(false),
621                evidence_hash: hash.clone(),
622            })
623            .unwrap();
624        store
625            .record_psp9_calibration_epoch(&Psp9CalibrationEpochRow {
626                epoch_id: "e1".into(),
627                stratum: "rust:test:v1".into(),
628                target_rho: 0.1,
629                threshold: Some(0.5),
630                state: "shadow".into(),
631                sample_count: 1,
632            })
633            .unwrap();
634        store
635            .record_psp9_calibration_sample("e1", "sample-1", 0.2, None, true)
636            .unwrap();
637        assert_eq!(
638            store
639                .latest_psp9_calibration_epoch("rust:test:v1")
640                .unwrap()
641                .unwrap()
642                .state,
643            "shadow"
644        );
645        assert_eq!(
646            store.psp9_calibration_samples("e1").unwrap(),
647            vec![("sample-1".into(), 0.2, None, true)]
648        );
649        let _ = std::fs::remove_dir_all(&dir);
650    }
651
652    #[test]
653    fn delayed_audit_labels_are_single_assignment() {
654        let (dir, store) = scratch_store();
655        store
656            .record_psp9_calibration_epoch(&Psp9CalibrationEpochRow {
657                epoch_id: "e1".into(),
658                stratum: "rust:test:v1".into(),
659                target_rho: 0.1,
660                threshold: None,
661                state: "insufficient_samples".into(),
662                sample_count: 0,
663            })
664            .unwrap();
665        store
666            .record_psp9_calibration_sample("e1", "sample-1", 0.2, None, true)
667            .unwrap();
668        assert_eq!(
669            store.pending_psp9_audit_samples(10).unwrap(),
670            vec![("e1".into(), "sample-1".into(), 0.2)]
671        );
672        assert_eq!(
673            store
674                .label_psp9_calibration_sample("sample-1", false)
675                .unwrap(),
676            1
677        );
678        // A second label never overwrites the first.
679        assert_eq!(
680            store
681                .label_psp9_calibration_sample("sample-1", true)
682                .unwrap(),
683            0
684        );
685        assert_eq!(
686            store
687                .labeled_psp9_samples_for_stratum("rust:test:v1")
688                .unwrap(),
689            vec![(0.2, false)]
690        );
691        assert_eq!(
692            store.psp9_epoch_stratum("e1").unwrap().as_deref(),
693            Some("rust:test:v1")
694        );
695        let _ = std::fs::remove_dir_all(&dir);
696    }
697
698    #[test]
699    fn durable_effect_is_guarded_by_the_exact_authority_epoch() {
700        let (_dir, store) = scratch_store();
701        store.initialize_authority_epoch("s1", 3).unwrap();
702        let ran = std::sync::atomic::AtomicBool::new(false);
703        store
704            .with_authority_epoch("s1", 3, || {
705                ran.store(true, std::sync::atomic::Ordering::SeqCst);
706                Ok(())
707            })
708            .unwrap();
709        assert!(ran.load(std::sync::atomic::Ordering::SeqCst));
710
711        store.revoke_authority("s1").unwrap();
712        ran.store(false, std::sync::atomic::Ordering::SeqCst);
713        assert!(store
714            .with_authority_epoch("s1", 3, || {
715                ran.store(true, std::sync::atomic::Ordering::SeqCst);
716                Ok(())
717            })
718            .is_err());
719        assert!(!ran.load(std::sync::atomic::Ordering::SeqCst));
720    }
721}
perspt_store/store/psp9_ledger.rs

perspt_store/store/
psp9_ledger.rs
