agent_graph_mcp/
migrations.rs

1//! Versioned durable-daemon migrations. Kept independent so the store can call it.
2use rusqlite::{params, Connection, OptionalExtension};
3use sha2::{Digest, Sha256};
4
5pub const CURRENT_VERSION: i64 = 4;
6pub const LEGACY_OWNER_UNKNOWN: &str = "legacy_owner_unknown";
7
8#[allow(dead_code)]
9pub trait MigrationStore {
10    fn connection(&self) -> &Connection;
11}
12
13pub fn apply(conn: &mut Connection, binary_digest: &str) -> rusqlite::Result<()> {
14    let tx = conn.transaction()?;
15    tx.execute_batch("CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, migration_digest TEXT NOT NULL);")?;
16    let exists: Option<i64> = tx
17        .query_row(
18            "SELECT version FROM schema_migrations WHERE version = ?1",
19            [CURRENT_VERSION],
20            |r| r.get(0),
21        )
22        .optional()?;
23    if exists.is_none() {
24        tx.execute_batch("CREATE TABLE IF NOT EXISTS daemon_instances (instance_id TEXT PRIMARY KEY, generation INTEGER NOT NULL UNIQUE, pid INTEGER NOT NULL, boot_id TEXT, executable_digest TEXT, started_at TEXT NOT NULL, heartbeat_at TEXT NOT NULL, clean_shutdown_at TEXT); CREATE TABLE IF NOT EXISTS run_publication_state (run_id TEXT PRIMARY KEY, state TEXT NOT NULL, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, reason TEXT); CREATE TABLE IF NOT EXISTS operator_receipts (receipt_id TEXT PRIMARY KEY, request_digest TEXT NOT NULL, action TEXT NOT NULL, resource_kind TEXT NOT NULL, resource_id TEXT NOT NULL, state_digest TEXT NOT NULL, operator_uid INTEGER NOT NULL, daemon_instance_id TEXT NOT NULL, nonce TEXT NOT NULL UNIQUE, issued_at TEXT NOT NULL, expires_at TEXT NOT NULL, consumed_at TEXT); CREATE INDEX IF NOT EXISTS idx_operator_receipts_nonce ON operator_receipts(nonce);")?;
25        let has_owner: Option<i64> = tx
26            .query_row(
27                "SELECT 1 FROM pragma_table_info('executions') WHERE name='owner_instance_id'",
28                [],
29                |r| r.get(0),
30            )
31            .optional()?;
32        if has_owner.is_none() {
33            let has_executions: Option<i64> = tx
34                .query_row(
35                    "SELECT 1 FROM sqlite_master WHERE type='table' AND name='executions'",
36                    [],
37                    |r| r.get(0),
38                )
39                .optional()?;
40            if has_executions.is_some() {
41                tx.execute_batch("ALTER TABLE executions ADD COLUMN owner_instance_id TEXT;")?;
42                tx.execute("UPDATE executions SET owner_instance_id = ?1 WHERE owner_instance_id IS NULL AND status IN ('accepted','running')", [LEGACY_OWNER_UNKNOWN])?;
43            }
44        }
45
46        // v4: deletion governance — phantom remediation + retention lifecycle
47        let has_superseded: Option<i64> = tx
48            .query_row(
49                "SELECT 1 FROM pragma_table_info('executions') WHERE name='superseded_by'",
50                [],
51                |r| r.get(0),
52            )
53            .optional()?;
54        if has_superseded.is_none() {
55            tx.execute_batch("ALTER TABLE executions ADD COLUMN superseded_by TEXT DEFAULT NULL;")?;
56        }
57        tx.execute_batch(
58            "CREATE TABLE IF NOT EXISTS legal_holds (\
59                hold_id TEXT PRIMARY KEY,\
60                graph_name TEXT NOT NULL,\
61                reason TEXT,\
62                issued_at TEXT NOT NULL DEFAULT (datetime('now')),\
63                expires_at TEXT,\
64                issued_by TEXT\
65            );\
66            CREATE TABLE IF NOT EXISTS archive_manifests (\
67                graph_name TEXT PRIMARY KEY,\
68                graph_version TEXT,\
69                spec_digest TEXT,\
70                version_count INTEGER,\
71                last_execution_at TEXT,\
72                execution_count INTEGER,\
73                content_digest TEXT,\
74                created_at TEXT NOT NULL DEFAULT (datetime('now'))\
75            );",
76        )?;
77
78        let digest = migration_digest(binary_digest);
79        tx.execute(
80            "INSERT INTO schema_migrations(version,migration_digest) VALUES (?1,?2)",
81            params![CURRENT_VERSION, digest],
82        )?;
83    }
84    tx.commit()
85}
86
87pub fn migration_digest(binary_digest: &str) -> String {
88    let mut h = Sha256::new();
89    h.update(format!(
90        "agent-graph-mcp:migration:{CURRENT_VERSION}:{binary_digest}"
91    ));
92    format!("{:x}", h.finalize())
93}
94
95#[allow(dead_code)]
96pub fn owner_for_new_run<'a>(owner: &'a str) -> rusqlite::Result<&'a str> {
97    if owner.is_empty() {
98        Err(rusqlite::Error::InvalidParameterName(
99            "owner_instance_id".into(),
100        ))
101    } else {
102        Ok(owner)
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109    #[test]
110    fn legacy_active_rows_are_quarantined() {
111        let mut c = Connection::open_in_memory().unwrap();
112        c.execute_batch("CREATE TABLE executions(run_id TEXT PRIMARY KEY,status TEXT NOT NULL); INSERT INTO executions VALUES ('a','running'),('b','completed');").unwrap();
113        apply(&mut c, "bin").unwrap();
114        let owner: String = c
115            .query_row(
116                "SELECT owner_instance_id FROM executions WHERE run_id='a'",
117                [],
118                |r| r.get(0),
119            )
120            .unwrap();
121        assert_eq!(owner, LEGACY_OWNER_UNKNOWN);
122        apply(&mut c, "bin").unwrap();
123        assert_eq!(
124            c.query_row::<i64, _, _>("SELECT count(*) FROM schema_migrations", [], |r| r.get(0))
125                .unwrap(),
126            1
127        );
128    }
129}
agent_graph_mcp/migrations.rs

agent_graph_mcp/
migrations.rs
