perspt_store/
schema.rs

1//! DuckDB Schema Initialization
2//!
3//! Creates the required tables for SRBN session persistence.
4
5use anyhow::Result;
6use duckdb::Connection;
7
8const CURRENT_SCHEMA_VERSION: i64 = 1;
9const EXPECTED_DUCKDB_VERSION: &str = "v1.5.5";
10
11/// Initialize the store through an idempotent transactional migration.
12pub fn init_schema(conn: &Connection) -> Result<()> {
13    check_dynamic_abi(conn)?;
14    conn.execute_batch(
15        "CREATE TABLE IF NOT EXISTS schema_migrations (\
16         version BIGINT PRIMARY KEY, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)",
17    )?;
18    let applied: bool = conn.query_row(
19        "SELECT count(*) > 0 FROM schema_migrations WHERE version = ?",
20        [CURRENT_SCHEMA_VERSION],
21        |row| row.get(0),
22    )?;
23    if applied {
24        return Ok(());
25    }
26    conn.execute_batch("BEGIN TRANSACTION")?;
27    let migration = apply_schema_current(conn).and_then(|()| {
28        conn.execute(
29            "INSERT INTO schema_migrations (version) VALUES (?)",
30            [CURRENT_SCHEMA_VERSION],
31        )?;
32        Ok(())
33    });
34    match migration {
35        Ok(()) => conn.execute_batch("COMMIT")?,
36        Err(error) => {
37            let _ = conn.execute_batch("ROLLBACK");
38            return Err(error.context("applying store schema migration v1"));
39        }
40    }
41    conn.execute_batch("CHECKPOINT")?;
42    Ok(())
43}
44
45fn check_dynamic_abi(conn: &Connection) -> Result<()> {
46    if cfg!(feature = "bundled") {
47        return Ok(());
48    }
49    let actual: String = conn.query_row("SELECT version()", [], |row| row.get(0))?;
50    anyhow::ensure!(
51        actual.starts_with(EXPECTED_DUCKDB_VERSION),
52        "incompatible dynamically linked DuckDB ABI: expected {}, found {}; \
53         install DuckDB 1.5.5 or build with the `bundled` feature",
54        EXPECTED_DUCKDB_VERSION,
55        actual
56    );
57    Ok(())
58}
59
60/// Current schema body. Every statement remains idempotent so an interrupted
61/// first migration can be retried safely after its transaction rolls back.
62fn apply_schema_current(conn: &Connection) -> Result<()> {
63    // Sessions table - top-level session tracking
64    conn.execute(
65        r#"
66        CREATE TABLE IF NOT EXISTS sessions (
67            session_id VARCHAR PRIMARY KEY,
68            task TEXT NOT NULL,
69            working_dir TEXT NOT NULL,
70            merkle_root BLOB,
71            detected_toolchain VARCHAR,
72            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
73            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
74            status VARCHAR DEFAULT 'active'
75        )
76        "#,
77        [],
78    )?;
79
80    // =========================================================================
81    // Plan Revision, Feature Charter, and Repair Footprint Tables
82    // =========================================================================
83
84    // =========================================================================
85    // PSP-7: SRBN step records and correction attempt telemetry
86    // =========================================================================
87
88    // PSP-9 system 14: the durable canonical event stream. The SDK keeps the
89    // chain semantics; this table keeps the bytes.
90    conn.execute(
91        r#"
92        CREATE TABLE IF NOT EXISTS psp9_ledger_events (
93            session_id VARCHAR NOT NULL,
94            sequence BIGINT NOT NULL,
95            event_json TEXT NOT NULL,
96            prev_hash VARCHAR NOT NULL,
97            hash VARCHAR NOT NULL,
98            PRIMARY KEY (session_id, sequence)
99        )
100        "#,
101        [],
102    )?;
103    conn.execute(
104        r#"
105        CREATE TABLE IF NOT EXISTS psp9_authority_epochs (
106            session_id VARCHAR PRIMARY KEY,
107            epoch BIGINT NOT NULL
108        )
109        "#,
110        [],
111    )?;
112    conn.execute(
113        r#"
114        CREATE TABLE IF NOT EXISTS psp9_grant_policies (
115            policy_id VARCHAR PRIMARY KEY,
116            session_id VARCHAR NOT NULL,
117            policy_json TEXT NOT NULL,
118            revoked BOOLEAN NOT NULL DEFAULT FALSE,
119            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
120        )
121        "#,
122        [],
123    )?;
124    conn.execute(
125        r#"
126        CREATE TABLE IF NOT EXISTS psp9_external_effects (
127            session_id VARCHAR NOT NULL,
128            idempotency_key VARCHAR NOT NULL,
129            intent_hash VARCHAR NOT NULL,
130            intent_json TEXT NOT NULL,
131            result_json TEXT,
132            status VARCHAR NOT NULL,
133            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
134            completed_at TIMESTAMP,
135            PRIMARY KEY (session_id, idempotency_key)
136        )
137        "#,
138        [],
139    )?;
140    conn.execute(
141        r#"
142        CREATE TABLE IF NOT EXISTS psp9_artifacts (
143            content_hash VARCHAR PRIMARY KEY,
144            content BLOB NOT NULL,
145            byte_len BIGINT NOT NULL,
146            media_type VARCHAR NOT NULL,
147            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
148        )
149        "#,
150        [],
151    )?;
152    conn.execute(
153        r#"
154        CREATE TABLE IF NOT EXISTS psp9_context_checkpoints (
155            session_id VARCHAR NOT NULL,
156            covered_event_root VARCHAR NOT NULL,
157            checkpoint_json TEXT NOT NULL,
158            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
159            PRIMARY KEY (session_id, covered_event_root)
160        )
161        "#,
162        [],
163    )?;
164    conn.execute(
165        r#"
166        CREATE TABLE IF NOT EXISTS psp9_verdicts (
167            session_id VARCHAR NOT NULL,
168            candidate_id VARCHAR NOT NULL,
169            validator_id VARCHAR NOT NULL,
170            stratum VARCHAR NOT NULL,
171            missed BOOLEAN NOT NULL,
172            unsafe_label BOOLEAN,
173            evidence_hash VARCHAR NOT NULL,
174            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
175            PRIMARY KEY (session_id, candidate_id, validator_id)
176        )
177        "#,
178        [],
179    )?;
180    conn.execute(
181        r#"
182        CREATE TABLE IF NOT EXISTS psp9_calibration_epochs (
183            epoch_id VARCHAR PRIMARY KEY,
184            stratum VARCHAR NOT NULL,
185            target_rho DOUBLE NOT NULL,
186            threshold DOUBLE,
187            state VARCHAR NOT NULL,
188            sample_count BIGINT NOT NULL,
189            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
190        )
191        "#,
192        [],
193    )?;
194    conn.execute(
195        r#"
196        CREATE TABLE IF NOT EXISTS psp9_calibration_samples (
197            epoch_id VARCHAR NOT NULL,
198            sample_id VARCHAR NOT NULL,
199            score DOUBLE NOT NULL,
200            -- NULL until the delayed audit label arrives; a sample only
201            -- counts toward the conformal floor once labeled.
202            unsafe_label BOOLEAN,
203            audit_selected BOOLEAN NOT NULL,
204            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
205            PRIMARY KEY (epoch_id, sample_id)
206        )
207        "#,
208        [],
209    )?;
210    // Migration for databases created before delayed audit labels existed.
211    // Inspect the catalog first rather than relying on ignored DDL errors.
212    if column_is_nullable(conn, "psp9_calibration_samples", "unsafe_label")? == Some(false) {
213        conn.execute(
214            "ALTER TABLE psp9_calibration_samples ALTER COLUMN unsafe_label DROP NOT NULL",
215            [],
216        )?;
217    }
218
219    log::info!("DuckDB schema initialized successfully");
220    Ok(())
221}
222
223fn column_is_nullable(conn: &Connection, table: &str, column: &str) -> Result<Option<bool>> {
224    let mut statement = conn.prepare(
225        "SELECT is_nullable FROM information_schema.columns \
226         WHERE table_schema = 'main' AND table_name = ? AND column_name = ?",
227    )?;
228    let mut rows = statement.query([table, column])?;
229    let Some(row) = rows.next()? else {
230        return Ok(None);
231    };
232    let nullable: String = row.get(0)?;
233    Ok(Some(nullable.eq_ignore_ascii_case("YES")))
234}
235
236#[cfg(test)]
237mod tests {
238    use super::*;
239
240    #[test]
241    fn schema_migration_is_transactional_and_idempotent() {
242        let connection = Connection::open_in_memory().unwrap();
243        init_schema(&connection).unwrap();
244        init_schema(&connection).unwrap();
245        let count: i64 = connection
246            .query_row("SELECT count(*) FROM schema_migrations", [], |row| {
247                row.get(0)
248            })
249            .unwrap();
250        assert_eq!(count, 1);
251    }
252
253    #[test]
254    fn delayed_label_column_is_nullable() {
255        let connection = Connection::open_in_memory().unwrap();
256        init_schema(&connection).unwrap();
257        assert_eq!(
258            column_is_nullable(&connection, "psp9_calibration_samples", "unsafe_label").unwrap(),
259            Some(true)
260        );
261    }
262}
perspt_store/schema.rs

perspt_store/
schema.rs
