ticket: own typed schema migrations

crates/ticket/src/lib.rs

@@ -20,8 +20,13 @@ use serde_yaml::{Mapping as YamlMapping, Value as YamlValue};
 use thiserror::Error;
 
 pub mod config;
+mod sqlite_schema;
 pub mod tool;
 
+pub use sqlite_schema::{
+    LATEST_SQLITE_TICKET_SCHEMA_VERSION, migrate_sqlite_ticket_schema, verify_sqlite_ticket_schema,
+};
+
 const REQUIRED_FIELDS: [&str; 4] = ["title", "state", "created_at", "updated_at"];
 const MAX_STATE_CHANGE_REASON_BYTES: usize = 1024;
 const MAX_INTAKE_SUMMARY_BODY_BYTES: usize = 16 * 1024;
@@ -2301,7 +2306,7 @@ impl fmt::Debug for SqliteTicketBackend {
 }
 
 impl SqliteTicketBackend {
-    pub fn new(db_path: impl Into<PathBuf>, workspace_id: impl Into<String>) -> Self {
+    fn configured(db_path: impl Into<PathBuf>, workspace_id: impl Into<String>) -> Self {
         Self {
             db_path: db_path.into(),
             workspace_id: workspace_id.into(),
@@ -2311,6 +2316,27 @@ impl SqliteTicketBackend {
         }
     }
 
+    /// Opens a standalone Ticket backend, applying all Ticket-owned migrations once.
+    pub fn open(db_path: impl Into<PathBuf>, workspace_id: impl Into<String>) -> Result<Self> {
+        let backend = Self::configured(db_path, workspace_id);
+        let connection = backend.connect()?;
+        migrate_sqlite_ticket_schema(&connection)?;
+        Ok(backend)
+    }
+
+    /// Connects to a database whose Ticket schema was composed by its startup owner.
+    ///
+    /// This performs verification only and never creates or alters schema objects.
+    pub fn open_verified(
+        db_path: impl Into<PathBuf>,
+        workspace_id: impl Into<String>,
+    ) -> Result<Self> {
+        let backend = Self::configured(db_path, workspace_id);
+        let connection = backend.connect()?;
+        verify_sqlite_ticket_schema(&connection)?;
+        Ok(backend)
+    }
+
     pub fn with_event_attributes(mut self, attributes: BTreeMap<String, String>) -> Self {
         self.event_attributes = attributes;
         self
@@ -2338,7 +2364,6 @@ impl SqliteTicketBackend {
 
     pub fn import_from_local_backend(&self, local: &LocalTicketBackend) -> Result<()> {
         let conn = self.open_connection()?;
-        self.ensure_schema(&conn)?;
         conn.execute_batch("BEGIN IMMEDIATE").map_err(sqlite_err)?;
         let result = (|| {
             for summary in local.list(TicketListQuery::all())? {
@@ -2351,7 +2376,7 @@ impl SqliteTicketBackend {
         finish_sqlite_transaction(&conn, result)
     }
 
-    fn open_connection(&self) -> Result<Connection> {
+    fn connect(&self) -> Result<Connection> {
         if let Some(parent) = self.db_path.parent() {
             fs::create_dir_all(parent).map_err(|error| io_err(parent, error))?;
         }
@@ -2361,115 +2386,20 @@ impl SqliteTicketBackend {
         Ok(conn)
     }
 
-    fn ensure_schema(&self, conn: &Connection) -> Result<()> {
+    fn open_connection(&self) -> Result<Connection> {
-        conn.execute_batch(r#"
+        let connection = self.connect()?;
-CREATE TABLE IF NOT EXISTS typed_tickets (
+        verify_sqlite_ticket_schema(&connection)?;
-    workspace_id TEXT NOT NULL,
+        Ok(connection)
-    ticket_id TEXT NOT NULL,
-    slug TEXT NOT NULL,
-    title TEXT NOT NULL,
-    status TEXT NOT NULL,
-    kind TEXT NOT NULL,
-    priority TEXT NOT NULL,
-    body TEXT NOT NULL,
-    created_at TEXT,
-    updated_at TEXT,
-    assignee TEXT,
-    readiness TEXT,
-    workflow_state TEXT NOT NULL,
-    workflow_state_explicit INTEGER NOT NULL,
-    queued_by TEXT,
-    queued_at TEXT,
-    resolution TEXT,
-    repository_id TEXT,
-    ref_selector TEXT,
-    PRIMARY KEY (workspace_id, ticket_id)
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_labels (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, ordinal INTEGER NOT NULL, label TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, ordinal),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_risk_flags (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, ordinal INTEGER NOT NULL, risk_flag TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, ordinal),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_raw_frontmatter (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, key),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_events (
-    workspace_id TEXT NOT NULL,
-    ticket_id TEXT NOT NULL,
-    event_index INTEGER NOT NULL,
-    kind TEXT NOT NULL,
-    author TEXT,
-    at TEXT,
-    status TEXT,
-    from_state TEXT,
-    to_state TEXT,
-    reason TEXT,
-    state_field TEXT,
-    heading TEXT,
-    body TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, event_index),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_event_references (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, event_index INTEGER NOT NULL, ordinal INTEGER NOT NULL, kind TEXT NOT NULL, target TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, event_index, ordinal),
-    FOREIGN KEY (workspace_id, ticket_id, event_index) REFERENCES typed_ticket_events(workspace_id, ticket_id, event_index) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_event_attributes (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, event_index INTEGER NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, event_index, key),
-    FOREIGN KEY (workspace_id, ticket_id, event_index) REFERENCES typed_ticket_events(workspace_id, ticket_id, event_index) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_relations (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, kind TEXT NOT NULL, target TEXT NOT NULL, note TEXT, author TEXT NOT NULL, at TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, kind, target),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_orchestration_plans (
-    workspace_id TEXT NOT NULL,
-    ticket_id TEXT NOT NULL,
-    record_id TEXT NOT NULL,
-    kind TEXT NOT NULL,
-    related_ticket TEXT,
-    note TEXT,
-    accepted_summary TEXT,
-    accepted_branch TEXT,
-    accepted_worktree TEXT,
-    accepted_role_plan TEXT,
-    author TEXT NOT NULL,
-    at TEXT NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, record_id),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-CREATE TABLE IF NOT EXISTS typed_ticket_artifacts (
-    workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, relative_path TEXT NOT NULL, content BLOB NOT NULL,
-    PRIMARY KEY (workspace_id, ticket_id, relative_path),
-    FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
-);
-"#)
-        .map_err(sqlite_err)?;
-        ensure_sqlite_ticket_column(conn, "repository_id", "TEXT")?;
-        ensure_sqlite_ticket_column(conn, "ref_selector", "TEXT")?;
-        Ok(())
     }
 
     fn with_write<R>(&self, op: impl FnOnce(&Connection) -> Result<R>) -> Result<R> {
         let conn = self.open_connection()?;
-        self.ensure_schema(&conn)?;
         conn.execute_batch("BEGIN IMMEDIATE").map_err(sqlite_err)?;
         finish_sqlite_transaction(&conn, op(&conn))
     }
 
     fn with_read<R>(&self, op: impl FnOnce(&Connection) -> Result<R>) -> Result<R> {
         let conn = self.open_connection()?;
-        self.ensure_schema(&conn)?;
         op(&conn)
     }
 
@@ -2915,30 +2845,6 @@ CREATE TABLE IF NOT EXISTS typed_ticket_artifacts (
     }
 }
 
-fn ensure_sqlite_ticket_column(
-    conn: &rusqlite::Connection,
-    name: &str,
-    sql_type: &str,
-) -> Result<()> {
-    let mut statement = conn
-        .prepare("PRAGMA table_info(typed_tickets)")
-        .map_err(sqlite_err)?;
-    let columns = statement
-        .query_map([], |row| row.get::<_, String>(1))
-        .map_err(sqlite_err)?;
-    for column in columns {
-        if column.map_err(sqlite_err)? == name {
-            return Ok(());
-        }
-    }
-    conn.execute(
-        format!("ALTER TABLE typed_tickets ADD COLUMN {name} {sql_type}").as_str(),
-        [],
-    )
-    .map_err(sqlite_err)?;
-    Ok(())
-}
-
 fn finish_sqlite_transaction<R>(conn: &Connection, result: Result<R>) -> Result<R> {
     match result {
         Ok(output) => {
@@ -6406,28 +6312,11 @@ state: planning
     #[test]
     fn sqlite_backend_persists_and_edits_ticket_target() {
         let tmp = TempDir::new().unwrap();
-        let backend = SqliteTicketBackend::new(tmp.path().join("workspace.db"), "workspace-test");
+        let backend =
+            SqliteTicketBackend::open(tmp.path().join("workspace.db"), "workspace-test").unwrap();
         assert_ticket_target_edit_semantics(&backend);
     }
 
-    #[test]
-    fn sqlite_ticket_target_columns_are_added_to_existing_table() {
-        let tmp = TempDir::new().unwrap();
-        let conn = rusqlite::Connection::open(tmp.path().join("workspace.db")).unwrap();
-        conn.execute_batch("CREATE TABLE typed_tickets (ticket_id TEXT PRIMARY KEY);")
-            .unwrap();
-        ensure_sqlite_ticket_column(&conn, "repository_id", "TEXT").unwrap();
-        ensure_sqlite_ticket_column(&conn, "ref_selector", "TEXT").unwrap();
-        let mut statement = conn.prepare("PRAGMA table_info(typed_tickets)").unwrap();
-        let columns = statement
-            .query_map([], |row| row.get::<_, String>(1))
-            .unwrap()
-            .collect::<std::result::Result<Vec<_>, _>>()
-            .unwrap();
-        assert!(columns.iter().any(|column| column == "repository_id"));
-        assert!(columns.iter().any(|column| column == "ref_selector"));
-    }
-
     #[test]
     fn local_backend_edit_item_supports_partial_body_replacement() {
         let tmp = TempDir::new().unwrap();
@@ -6438,7 +6327,8 @@ state: planning
     #[test]
     fn sqlite_backend_edit_item_supports_partial_body_replacement() {
         let tmp = TempDir::new().unwrap();
-        let backend = SqliteTicketBackend::new(tmp.path().join("workspace.db"), "workspace-test");
+        let backend =
+            SqliteTicketBackend::open(tmp.path().join("workspace.db"), "workspace-test").unwrap();
         assert_partial_body_replacement_semantics(&backend);
     }
 
@@ -6446,7 +6336,7 @@ state: planning
     fn sqlite_mutation_hook_failure_rolls_back_ticket_event() {
         let tmp = TempDir::new().unwrap();
         let db_path = tmp.path().join("workspace.db");
-        let backend = SqliteTicketBackend::new(&db_path, "workspace-test");
+        let backend = SqliteTicketBackend::open(&db_path, "workspace-test").unwrap();
         let created = backend.create(NewTicket::new("Atomic mutation")).unwrap();
         let before = backend
             .show(TicketIdOrSlug::Id(created.id.clone()))
@@ -6480,7 +6370,8 @@ state: planning
     #[test]
     fn sqlite_backend_persists_core_ticket_operations() {
         let tmp = TempDir::new().unwrap();
-        let backend = SqliteTicketBackend::new(tmp.path().join("workspace.db"), "workspace-test");
+        let backend =
+            SqliteTicketBackend::open(tmp.path().join("workspace.db"), "workspace-test").unwrap();
         let created = backend.create(NewTicket::new("SQLite Ticket")).unwrap();
         backend
             .add_event(
@@ -6501,7 +6392,9 @@ state: planning
             )
             .unwrap();
 
-        let reopened = SqliteTicketBackend::new(tmp.path().join("workspace.db"), "workspace-test");
+        let reopened =
+            SqliteTicketBackend::open_verified(tmp.path().join("workspace.db"), "workspace-test")
+                .unwrap();
         let list = reopened.list(TicketListQuery::all()).unwrap();
         assert_eq!(list.len(), 1);
         assert_eq!(list[0].id, created.id);
@@ -6531,7 +6424,8 @@ state: planning
         let tmp = TempDir::new().unwrap();
         let local = backend(&tmp);
         let created = local.create(NewTicket::new("Legacy Ticket")).unwrap();
-        let db = SqliteTicketBackend::new(tmp.path().join("workspace.db"), "workspace-test");
+        let db =
+            SqliteTicketBackend::open(tmp.path().join("workspace.db"), "workspace-test").unwrap();
         db.import_from_local_backend(&local).unwrap();
 
         let ticket = db.show(TicketIdOrSlug::Id(created.id.clone())).unwrap();

crates/workspace-server/src/authority.rs

@@ -131,7 +131,7 @@ impl SqliteWorkspaceAuthority {
         Ok(Self {
             workspace_id: workspace_id.clone(),
             store: SqliteWorkspaceStore::open(&database_path)?,
-            ticket_backend: SqliteTicketBackend::new(database_path, workspace_id),
+            ticket_backend: SqliteTicketBackend::open_verified(database_path, workspace_id)?,
         })
     }
 
@@ -725,7 +725,8 @@ mod tests {
         let dir = tempfile::tempdir().unwrap();
         write_ticket(dir.path(), "00000000001J2", "Read bridge", "ready");
         let db_path = dir.path().join("workspace.db");
-        SqliteTicketBackend::new(&db_path, "workspace-test")
+        SqliteTicketBackend::open(&db_path, "workspace-test")
+            .unwrap()
             .import_from_local_backend(&ticket::LocalTicketBackend::new(
                 dir.path().join(".yoi/tickets"),
             ))

crates/workspace-server/src/server.rs

@@ -2383,10 +2383,10 @@ struct BrowserCloseTicketRequest {
 fn browser_ticket_backend(api: &WorkspaceApi) -> Result<SqliteTicketBackend> {
     let config = ticket::config::TicketConfig::load_workspace(&api.config.workspace_root)
         .map_err(|error| Error::Config(format!("load Ticket workspace settings: {error}")))?;
-    Ok(SqliteTicketBackend::new(
+    Ok(SqliteTicketBackend::open_verified(
         api.config.database_path.clone(),
         api.config.workspace_id.clone(),
-    )
+    )?
     .with_record_language(config.ticket_record_language()))
 }
 
@@ -2553,10 +2553,11 @@ async fn execute_worker_ticket_rest_operation(
     validate_workspace_scope(api, workspace_id)?;
     let config = ticket::config::TicketConfig::load_workspace(&api.config.workspace_root)
         .map_err(|error| Error::Config(format!("load Ticket workspace settings: {error}")))?;
-    let mut backend = SqliteTicketBackend::new(
+    let mut backend = SqliteTicketBackend::open_verified(
         api.config.database_path.clone(),
         api.config.workspace_id.clone(),
     )
+    .map_err(Error::from)?
     .with_record_language(config.ticket_record_language());
     let operation_kind = ticket_mutation_operation_kind(&operation);
     let is_mutation = operation_kind != "read";
@@ -15225,7 +15226,7 @@ mod tests {
     ) {
         use ticket::TicketBackend as _;
 
-        let backend = ticket::SqliteTicketBackend::new(database_path, workspace_id);
+        let backend = ticket::SqliteTicketBackend::open(database_path, workspace_id).unwrap();
         let mut input = ticket::NewTicket::new(title);
         input.workflow_state = Some(state);
         backend.create(input).unwrap();

crates/workspace-server/src/store.rs

@@ -754,6 +754,7 @@ impl SqliteWorkspaceStore {
     pub fn from_connection(conn: Connection) -> Result<Self> {
         configure_sqlite(&conn)?;
         apply_migrations(&conn)?;
+        ticket::migrate_sqlite_ticket_schema(&conn)?;
         Ok(Self {
             conn: Arc::new(Mutex::new(conn)),
         })
@@ -4526,6 +4527,45 @@ mod tests {
     use super::*;
     use std::collections::BTreeSet;
 
+    #[test]
+    fn startup_composes_ticket_migrations_when_control_plane_is_current() {
+        let conn = Connection::open_in_memory().unwrap();
+        configure_sqlite(&conn).unwrap();
+        apply_migrations(&conn).unwrap();
+        assert!(!table_exists(&conn, "ticket_schema_migrations").unwrap());
+
+        let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
+        store
+            .with_conn(|conn| {
+                ticket::verify_sqlite_ticket_schema(conn)?;
+                let latest = conn.query_row(
+                    "SELECT MAX(version) FROM ticket_schema_migrations",
+                    [],
+                    |row| row.get::<_, i64>(0),
+                )?;
+                assert_eq!(latest, ticket::LATEST_SQLITE_TICKET_SCHEMA_VERSION);
+                Ok(())
+            })
+            .unwrap();
+    }
+
+    #[test]
+    fn startup_fails_closed_when_current_ticket_schema_has_drifted() {
+        let conn = Connection::open_in_memory().unwrap();
+        configure_sqlite(&conn).unwrap();
+        apply_migrations(&conn).unwrap();
+        ticket::migrate_sqlite_ticket_schema(&conn).unwrap();
+        conn.execute_batch("DROP TABLE typed_ticket_artifacts")
+            .unwrap();
+
+        let result = SqliteWorkspaceStore::from_connection(conn);
+        let error = match result {
+            Ok(_) => panic!("schema drift unexpectedly passed startup verification"),
+            Err(error) => error,
+        };
+        assert!(error.to_string().contains("typed_ticket_artifacts"));
+    }
+
     #[test]
     fn removes_unused_control_plane_ticket_tables() {
         let conn = Connection::open_in_memory().unwrap();

crates/yoi/src/ticket_cli.rs

@@ -370,7 +370,7 @@ fn backend_for_workspace(workspace: &Path) -> Result<Box<dyn TicketBackend>, Tic
     let workspace_id = workspace_id_for_workspace(workspace)?;
     let db_path = server_database_path(workspace)?;
     Ok(Box::new(
-        SqliteTicketBackend::new(db_path, workspace_id)
+        SqliteTicketBackend::open(db_path, workspace_id)?
             .with_record_language(config.ticket_record_language()),
     ))
 }
@@ -381,7 +381,7 @@ fn import_local(workspace: &Path) -> Result<TicketCliOutput, TicketCliError> {
         .with_record_language(config.ticket_record_language());
     let workspace_id = workspace_id_for_workspace(workspace)?;
     let db_path = server_database_path(workspace)?;
-    let sqlite = SqliteTicketBackend::new(db_path.clone(), workspace_id)
+    let sqlite = SqliteTicketBackend::open(db_path.clone(), workspace_id)?
         .with_record_language(config.ticket_record_language());
     sqlite.import_from_local_backend(&local)?;
     Ok(success(format!(