zebflow 0.9.1

//! SQLite-backed platform catalog adapter (WAL mode, bundled SQLite 3.47).
//!
//! Uses a single `catalog.db` file under `{data_root}/platform/` with proper
//! WAL journaling. Safe across K8s restarts — no unsafe mmap code.

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use rusqlite::{Connection, OptionalExtension, Row, Transaction, TransactionBehavior, params};
use serde_json::Value;

use crate::infra::cluster::registry::WorkerRegistryRecord;
use crate::infra::execution::placement::{
    ProjectRuntimeMode, ProjectRuntimePlacement, ProjectRuntimePlacementTarget,
};
use crate::infra::secrets::keyring::{
    CREDENTIAL_KEY_REL, CREDENTIAL_KEY_VAR, CredentialKeyring, WrappedDataKey,
};
use crate::platform::adapters::data::DataAdapter;
use crate::platform::error::PlatformError;
use crate::platform::model::{
    CREDENTIAL_STATE_CHOSEN, CredentialKeyGeneration, CredentialKeyringReport, ErrorGroupBounds, PipelineErrorGroup,
    CredentialSweepReport, HubAccessGrant, HubAssetPackage, HubAssetVersion, HubAuthority,
    HubPublisher, HubToken, LOCAL_AUTHORITY_EVENT_BREAK_GLASS, McpSession, PipelineInvocationEntry,
    PipelineInvocationLogPipelineStats, PipelineInvocationLogStats, PipelineMeta,
    PlatformHubRepository, PlatformOffice, PlatformOfficeIdentityWrite, PlatformOfficeJoinToken,
    PlatformOfficeLocalAuthorityEvent, PlatformOfficeNode, PlatformOfficeVouchRedemption,
    PlatformProject, PlatformServiceInstance, PlatformUser, PlatformUserLocalAuth,
    ProjectAccessRolePreset, ProjectCapability, ProjectCredential, ProjectDbConnection,
    ProjectHubRepository, ProjectInvite, ProjectInviteStatus, ProjectMember, ProjectOperationKind,
    ProjectOperationRecord, ProjectOperationStatus, ProjectPolicy, ProjectPolicyBinding,
    ProjectSubjectKind, StoredUser, now_ts, slug_segment,
};

const SCHEMA_SQL: &str = "
CREATE TABLE IF NOT EXISTS users (
    owner         TEXT PRIMARY KEY,
    user_id       TEXT NOT NULL UNIQUE DEFAULT '',
    role          TEXT NOT NULL DEFAULT 'owner',
    git_name      TEXT NOT NULL DEFAULT '',
    git_email     TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS user_local_auth (
    user_id              TEXT PRIMARY KEY,
    password_hash        TEXT NOT NULL DEFAULT '',
    password_alg         TEXT NOT NULL DEFAULT 'sha256',
    password_updated_at  INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS projects (
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    project_id    TEXT NOT NULL UNIQUE DEFAULT '',
    owner_user_id TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project),
    FOREIGN KEY (owner_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS project_credentials (
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    credential_id TEXT NOT NULL,
    title         TEXT NOT NULL DEFAULT '',
    kind          TEXT NOT NULL DEFAULT '',
    secret_json   TEXT NOT NULL DEFAULT 'null',
    notes         TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, credential_id)
);
CREATE TABLE IF NOT EXISTS project_db_connections (
    owner            TEXT NOT NULL,
    project          TEXT NOT NULL,
    connection_id    TEXT NOT NULL,
    connection_slug  TEXT NOT NULL DEFAULT '',
    connection_label TEXT NOT NULL DEFAULT '',
    database_kind    TEXT NOT NULL DEFAULT '',
    credential_id    TEXT,
    config_json      TEXT NOT NULL DEFAULT 'null',
    created_at       INTEGER NOT NULL DEFAULT 0,
    updated_at       INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, connection_id)
);
CREATE TABLE IF NOT EXISTS project_hub_repositories (
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    repository_id TEXT NOT NULL,
    title         TEXT NOT NULL DEFAULT '',
    base_url      TEXT NOT NULL DEFAULT '',
    remote_owner  TEXT NOT NULL DEFAULT '',
    remote_project TEXT NOT NULL DEFAULT '',
    read_token    TEXT NOT NULL DEFAULT '',
    visibility    TEXT NOT NULL DEFAULT 'public',
    enabled       INTEGER NOT NULL DEFAULT 1,
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, repository_id)
);
CREATE TABLE IF NOT EXISTS platform_hub_repositories (
    source_id     TEXT NOT NULL UNIQUE DEFAULT '',
    owner_user_id TEXT NOT NULL DEFAULT '',
    owner         TEXT NOT NULL,
    repository_id TEXT NOT NULL,
    title         TEXT NOT NULL DEFAULT '',
    base_url      TEXT NOT NULL DEFAULT '',
    remote_owner  TEXT NOT NULL DEFAULT '',
    remote_project TEXT NOT NULL DEFAULT '',
    read_token    TEXT NOT NULL DEFAULT '',
    kind          TEXT NOT NULL DEFAULT 'api',
    priority      INTEGER NOT NULL DEFAULT 100,
    visibility    TEXT NOT NULL DEFAULT 'public',
    enabled       INTEGER NOT NULL DEFAULT 1,
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, repository_id),
    FOREIGN KEY (owner_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS hub_access_grants (
    grant_id       TEXT PRIMARY KEY,
    source_owner   TEXT NOT NULL DEFAULT '',
    source_id      TEXT NOT NULL DEFAULT '',
    repository_id  TEXT NOT NULL DEFAULT '',
    grant_scope    TEXT NOT NULL DEFAULT 'selected_project',
    target_owner   TEXT NOT NULL DEFAULT '',
    target_project TEXT NOT NULL DEFAULT '',
    can_read       INTEGER NOT NULL DEFAULT 1,
    can_publish    INTEGER NOT NULL DEFAULT 0,
    can_manage     INTEGER NOT NULL DEFAULT 0,
    enabled        INTEGER NOT NULL DEFAULT 1,
    created_at     INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS hub_authorities (
    authority_id     TEXT PRIMARY KEY,
    host_project_id  TEXT NOT NULL UNIQUE DEFAULT '',
    owner            TEXT NOT NULL DEFAULT '',
    project          TEXT NOT NULL DEFAULT '',
    enabled          INTEGER NOT NULL DEFAULT 0,
    public_base_url  TEXT NOT NULL DEFAULT '',
    created_at       INTEGER NOT NULL DEFAULT 0,
    updated_at       INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (host_project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS hub_publishers (
    authority_id    TEXT NOT NULL DEFAULT '',
    publisher_pk    TEXT NOT NULL UNIQUE DEFAULT '',
    owner          TEXT NOT NULL,
    project        TEXT NOT NULL,
    publisher_id   TEXT NOT NULL,
    display_name   TEXT NOT NULL DEFAULT '',
    publisher_url  TEXT NOT NULL DEFAULT '',
    email          TEXT NOT NULL DEFAULT '',
    description    TEXT NOT NULL DEFAULT '',
    icon_url       TEXT NOT NULL DEFAULT '',
    website_url    TEXT NOT NULL DEFAULT '',
    enabled        INTEGER NOT NULL DEFAULT 1,
    can_read       INTEGER NOT NULL DEFAULT 1,
    can_publish    INTEGER NOT NULL DEFAULT 1,
    can_manage     INTEGER NOT NULL DEFAULT 0,
    max_packages      INTEGER NOT NULL DEFAULT 0,
    max_package_bytes INTEGER NOT NULL DEFAULT 0,
    max_media_files   INTEGER NOT NULL DEFAULT 0,
    max_image_bytes   INTEGER NOT NULL DEFAULT 0,
    created_at     INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, publisher_id),
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS hub_asset_packages (
    package_pk       TEXT NOT NULL UNIQUE DEFAULT '',
    authority_id     TEXT NOT NULL DEFAULT '',
    publisher_pk     TEXT NOT NULL DEFAULT '',
    package_id       TEXT PRIMARY KEY,
    authority_owner  TEXT NOT NULL DEFAULT '',
    authority_project TEXT NOT NULL DEFAULT '',
    publisher_owner  TEXT NOT NULL DEFAULT '',
    publisher_id     TEXT NOT NULL DEFAULT '',
    publisher_display_name TEXT NOT NULL DEFAULT '',
    publisher_url    TEXT NOT NULL DEFAULT '',
    publisher_email  TEXT NOT NULL DEFAULT '',
    asset_kind       TEXT NOT NULL DEFAULT '',
    title            TEXT NOT NULL DEFAULT '',
    description      TEXT NOT NULL DEFAULT '',
    summary          TEXT NOT NULL DEFAULT '',
    description_md   TEXT NOT NULL DEFAULT '',
    image_url        TEXT NOT NULL DEFAULT '',
    media_json       TEXT NOT NULL DEFAULT '[]',
    gallery_json     TEXT NOT NULL DEFAULT '{}',
    visibility       TEXT NOT NULL DEFAULT 'private',
    tags_json        TEXT NOT NULL DEFAULT '[]',
    retracted_at     INTEGER,
    retracted_reason TEXT NOT NULL DEFAULT '',
    created_at       INTEGER NOT NULL DEFAULT 0,
    updated_at       INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (publisher_pk) REFERENCES hub_publishers(publisher_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS hub_asset_versions (
    package_pk        TEXT NOT NULL DEFAULT '',
    package_id        TEXT NOT NULL,
    version           TEXT NOT NULL,
    authority_owner   TEXT NOT NULL DEFAULT '',
    authority_project TEXT NOT NULL DEFAULT '',
    publisher_owner   TEXT NOT NULL DEFAULT '',
    publisher_id      TEXT NOT NULL DEFAULT '',
    source_owner      TEXT NOT NULL DEFAULT '',
    source_project    TEXT NOT NULL DEFAULT '',
    source_kind       TEXT NOT NULL DEFAULT '',
    source_ref        TEXT NOT NULL DEFAULT '',
    artifact_rel_path TEXT NOT NULL DEFAULT '',
    artifact_sha256   TEXT NOT NULL DEFAULT '',
    manifest_json     TEXT NOT NULL DEFAULT 'null',
    retracted_at      INTEGER,
    retracted_reason  TEXT NOT NULL DEFAULT '',
    created_at        INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (package_id, version),
    FOREIGN KEY (package_pk) REFERENCES hub_asset_packages(package_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS hub_tokens (
    token_id       TEXT PRIMARY KEY,
    authority_id   TEXT NOT NULL DEFAULT '',
    publisher_pk   TEXT NOT NULL DEFAULT '',
    owner          TEXT NOT NULL DEFAULT '',
    project        TEXT NOT NULL DEFAULT '',
    publisher_id   TEXT NOT NULL DEFAULT '',
    publisher_display_name TEXT NOT NULL DEFAULT '',
    publisher_url  TEXT NOT NULL DEFAULT '',
    publisher_email TEXT NOT NULL DEFAULT '',
    title          TEXT NOT NULL DEFAULT '',
    secret_hash    TEXT NOT NULL DEFAULT '',
    scopes_json    TEXT NOT NULL DEFAULT '[]',
    expires_at     INTEGER,
    last_used_at   INTEGER,
    revoked_at     INTEGER,
    created_at     INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (publisher_pk) REFERENCES hub_publishers(publisher_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS pipeline_meta (
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    file_rel_path TEXT NOT NULL,
    name          TEXT NOT NULL DEFAULT '',
    title         TEXT NOT NULL DEFAULT '',
    virtual_path  TEXT NOT NULL DEFAULT '',
    description   TEXT NOT NULL DEFAULT '',
    trigger_kind  TEXT NOT NULL DEFAULT '',
    hash          TEXT NOT NULL DEFAULT '',
    active_hash   TEXT,
    activated_at  INTEGER,
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, file_rel_path)
);
CREATE TABLE IF NOT EXISTS project_policies (
    project_id        TEXT NOT NULL DEFAULT '',
    owner             TEXT NOT NULL,
    project           TEXT NOT NULL,
    policy_id         TEXT NOT NULL,
    title             TEXT NOT NULL DEFAULT '',
    capabilities_json TEXT NOT NULL DEFAULT '[]',
    managed           INTEGER NOT NULL DEFAULT 0,
    created_at        INTEGER NOT NULL DEFAULT 0,
    updated_at        INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, policy_id),
    UNIQUE (project_id, policy_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS project_policy_bindings (
    project_id   TEXT NOT NULL DEFAULT '',
    owner        TEXT NOT NULL,
    project      TEXT NOT NULL,
    subject_kind TEXT NOT NULL,
    subject_id   TEXT NOT NULL,
    policy_id    TEXT NOT NULL,
    created_at   INTEGER NOT NULL DEFAULT 0,
    updated_at   INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, subject_id, policy_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (project_id, policy_id) REFERENCES project_policies(project_id, policy_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS project_members (
    project_id              TEXT NOT NULL DEFAULT '',
    owner                  TEXT NOT NULL,
    project                TEXT NOT NULL,
    user_id                TEXT NOT NULL,
    member_user_id         TEXT NOT NULL DEFAULT '',
    role_preset            TEXT NOT NULL DEFAULT 'reporter',
    custom_policy_ids_json TEXT NOT NULL DEFAULT '[]',
    mcp_capabilities_json  TEXT NOT NULL DEFAULT '[]',
    created_by             TEXT NOT NULL DEFAULT '',
    created_by_user_id     TEXT NOT NULL DEFAULT '',
    created_at             INTEGER NOT NULL DEFAULT 0,
    updated_at             INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, user_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (member_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS project_invites (
    owner                  TEXT NOT NULL,
    project                TEXT NOT NULL,
    invite_id              TEXT NOT NULL,
    target_user            TEXT NOT NULL DEFAULT '',
    role_preset            TEXT NOT NULL DEFAULT 'reporter',
    custom_policy_ids_json TEXT NOT NULL DEFAULT '[]',
    mcp_capabilities_json  TEXT NOT NULL DEFAULT '[]',
    note                   TEXT NOT NULL DEFAULT '',
    invited_by             TEXT NOT NULL DEFAULT '',
    status                 TEXT NOT NULL DEFAULT 'pending',
    expires_at             INTEGER,
    created_at             INTEGER NOT NULL DEFAULT 0,
    updated_at             INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, invite_id)
);
CREATE TABLE IF NOT EXISTS offices (
    office_id     TEXT PRIMARY KEY,
    office_slug   TEXT NOT NULL UNIQUE DEFAULT '',
    label         TEXT NOT NULL DEFAULT '',
    office_kind   TEXT NOT NULL DEFAULT 'office',
    base_url      TEXT NOT NULL DEFAULT '',
    status        TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS office_join_tokens (
    office_id     TEXT PRIMARY KEY,
    secret_digest TEXT NOT NULL DEFAULT '',
    status        TEXT NOT NULL DEFAULT 'active',
    note          TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    last_used_at  INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS platform_service_instances (
    service_instance_id TEXT PRIMARY KEY,
    service_kind        TEXT NOT NULL DEFAULT '',
    display_label       TEXT NOT NULL DEFAULT '',
    host_office_id      TEXT NOT NULL DEFAULT '',
    state_office_id     TEXT NOT NULL DEFAULT '',
    public_base_url     TEXT NOT NULL DEFAULT '',
    enabled             INTEGER NOT NULL DEFAULT 0,
    status              TEXT NOT NULL DEFAULT '',
    placement_generation INTEGER NOT NULL DEFAULT 0,
    created_at          INTEGER NOT NULL DEFAULT 0,
    updated_at          INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS office_nodes (
    node_id             TEXT PRIMARY KEY,
    office_id           TEXT NOT NULL DEFAULT '',
    label               TEXT NOT NULL DEFAULT '',
    base_url            TEXT NOT NULL DEFAULT '',
    status              TEXT NOT NULL DEFAULT '',
    capabilities_json   TEXT NOT NULL DEFAULT '{}',
    registered_at       INTEGER NOT NULL DEFAULT 0,
    last_heartbeat_at   INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS worker_registry (
    office_id           TEXT NOT NULL DEFAULT '',
    office_slug         TEXT NOT NULL DEFAULT '',
    node_id             TEXT PRIMARY KEY,
    label               TEXT NOT NULL DEFAULT '',
    base_url            TEXT NOT NULL DEFAULT '',
    status              TEXT NOT NULL DEFAULT '',
    capabilities_json   TEXT NOT NULL DEFAULT '{}',
    registered_at       INTEGER NOT NULL DEFAULT 0,
    last_heartbeat_at   INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS project_runtime_placements (
    project_id           TEXT NOT NULL DEFAULT '',
    owner               TEXT NOT NULL,
    project             TEXT NOT NULL,
    mode                TEXT NOT NULL DEFAULT 'shared',
    target              TEXT NOT NULL DEFAULT 'local',
    target_office_id    TEXT,
    target_node_id      TEXT,
    worker_id           TEXT,
    resource_profile    TEXT NOT NULL DEFAULT '',
    desired_replicas    INTEGER NOT NULL DEFAULT 1,
    effective_state     TEXT NOT NULL DEFAULT '',
    created_at          INTEGER NOT NULL DEFAULT 0,
    updated_at          INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (target_office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (target_node_id) REFERENCES office_nodes(node_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS project_operations (
    project_id          TEXT NOT NULL DEFAULT '',
    owner               TEXT NOT NULL,
    project             TEXT NOT NULL,
    operation_id        TEXT NOT NULL,
    kind                TEXT NOT NULL DEFAULT '',
    status              TEXT NOT NULL DEFAULT 'pending',
    current_step        TEXT NOT NULL DEFAULT '',
    source_office_id    TEXT,
    target_office_id    TEXT,
    artifact_rel_path   TEXT,
    artifact_sha256     TEXT,
    artifact_bytes      INTEGER,
    error_message       TEXT,
    retry_count         INTEGER NOT NULL DEFAULT 0,
    created_at          INTEGER NOT NULL DEFAULT 0,
    updated_at          INTEGER NOT NULL DEFAULT 0,
    completed_at        INTEGER,
    PRIMARY KEY (owner, project, operation_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (source_office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (target_office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
);
CREATE INDEX IF NOT EXISTS idx_project_operations_project
    ON project_operations (owner, project, updated_at DESC, operation_id ASC);
CREATE TABLE IF NOT EXISTS mcp_sessions (
    token              TEXT PRIMARY KEY,
    owner              TEXT NOT NULL DEFAULT '',
    project            TEXT NOT NULL DEFAULT '',
    capabilities_json  TEXT NOT NULL DEFAULT '[]',
    created_at         INTEGER NOT NULL DEFAULT 0,
    auto_reset_seconds INTEGER,
    enabled            INTEGER NOT NULL DEFAULT 1
);
CREATE TABLE IF NOT EXISTS pipeline_invocations (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    file_rel_path TEXT NOT NULL,
    run_id        TEXT NOT NULL DEFAULT '',
    at            INTEGER NOT NULL DEFAULT 0,
    duration_ms   INTEGER NOT NULL DEFAULT 0,
    status        TEXT NOT NULL DEFAULT '',
    trigger       TEXT NOT NULL DEFAULT '',
    error         TEXT,
    trace_json    TEXT NOT NULL DEFAULT '[]'
);
CREATE INDEX IF NOT EXISTS idx_pipeline_invocations_pipeline
    ON pipeline_invocations (owner, project, file_rel_path, at DESC);
";

const SCHEMA_MIGRATIONS_SQL: &str = "
CREATE TABLE IF NOT EXISTS schema_migrations (
    version     INTEGER PRIMARY KEY,
    name        TEXT NOT NULL,
    applied_at  INTEGER NOT NULL DEFAULT 0
);
";

#[derive(Clone, Copy)]
struct MigrationDef {
    version: i64,
    name: &'static str,
    apply: fn(&Transaction<'_>) -> Result<(), PlatformError>,
}

/// Platform catalog adapter backed by a single WAL-mode SQLite file.
pub struct SqliteDataAdapter {
    conn: Arc<Mutex<Connection>>,
    data_root: Option<PathBuf>,
    /// This instance's credential encryption keyring.
    ///
    /// `Some` for every adapter that owns a data root, which is every adapter a
    /// server mode or an offline command builds. `None` only for
    /// [`Self::new_at_db_path`], the office-hosted hub store, which holds no
    /// credentials and refuses the operations that would need one — the
    /// contract's "reading a credential when the key is absent" rejection,
    /// enforced by the storage boundary rather than by a caller remembering.
    keyring: Option<Arc<CredentialKeyring>>,
}

fn hub_access_grant_from_row(row: &Row<'_>) -> rusqlite::Result<HubAccessGrant> {
    Ok(HubAccessGrant {
        grant_id: row.get(0)?,
        source_owner: row.get(1)?,
        source_id: row.get(2)?,
        repository_id: row.get(3)?,
        grant_scope: row.get(4)?,
        target_owner: row.get(5)?,
        target_project: row.get(6)?,
        can_read: row.get::<_, i64>(7)? != 0,
        can_publish: row.get::<_, i64>(8)? != 0,
        can_manage: row.get::<_, i64>(9)? != 0,
        enabled: row.get::<_, i64>(10)? != 0,
        created_at: row.get(11)?,
        updated_at: row.get(12)?,
    })
}

/// The platform error code for one keyring failure.
///
/// Distinct codes rather than one, because an operator's next move differs:
/// a missing key is "find the file", a wrong key is "you have the wrong file",
/// and a mismatch is "decide which of the two you meant".
fn credential_key_error_code(err: &crate::infra::secrets::CredentialKeyError) -> &'static str {
    use crate::infra::secrets::CredentialKeyError as E;
    match err {
        E::Missing { .. } => "PLATFORM_CREDENTIAL_KEY_MISSING",
        E::Unusable { .. } | E::NoRandomness => "PLATFORM_CREDENTIAL_KEY_UNUSABLE",
        E::Mismatch { .. } => "PLATFORM_CREDENTIAL_KEY_MISMATCH",
        E::Io { .. } => "PLATFORM_CREDENTIAL_KEY_IO",
        E::UnknownGeneration { .. } => "PLATFORM_CREDENTIAL_KEY_UNKNOWN_GENERATION",
        E::RekeyNotOwned => "PLATFORM_CREDENTIAL_KEY_NOT_OWNED",
    }
}

impl SqliteDataAdapter {
    /// Opens or creates `{data_root}/platform/catalog.db` with WAL mode and
    /// applies ordered platform schema migrations.
    pub fn new(data_root: &Path) -> Result<Self, PlatformError> {
        std::fs::create_dir_all(data_root.join("platform"))?;
        let path = data_root.join("platform").join("catalog.db");
        Self::new_at_db_path_with_foreign_keys(&path, true, Some(data_root.to_path_buf()))
    }

    /// Opens or creates an exact SQLite DB path with the same catalog schema.
    ///
    /// Platform services use this to keep service-owned operational state under
    /// their own mounted service root instead of the platform control DB.
    pub fn new_at_db_path(path: &Path) -> Result<Self, PlatformError> {
        Self::new_at_db_path_with_foreign_keys(path, false, None)
    }

    fn new_at_db_path_with_foreign_keys(
        path: &Path,
        foreign_keys: bool,
        data_root: Option<PathBuf>,
    ) -> Result<Self, PlatformError> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let mut conn = Connection::open(&path)
            .map_err(|e| PlatformError::new("PLATFORM_SQLITE_OPEN", e.to_string()))?;
        let foreign_keys_pragma = if foreign_keys { "ON" } else { "OFF" };
        conn.execute_batch(&format!(
            "PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA foreign_keys={foreign_keys_pragma};"
        ))
            .map_err(|e| PlatformError::new("PLATFORM_SQLITE_PRAGMA", e.to_string()))?;
        Self::run_migrations(&mut conn)?;
        conn.execute_batch(&format!("PRAGMA foreign_keys={foreign_keys_pragma};"))
            .map_err(|e| PlatformError::new("PLATFORM_SQLITE_PRAGMA", e.to_string()))?;
        let mut adapter = Self {
            conn: Arc::new(Mutex::new(conn)),
            data_root,
            keyring: None,
        };
        adapter.open_credential_keyring()?;
        Ok(adapter)
    }

    /// Resolve this instance's credential keyring, or refuse to exist.
    ///
    /// This runs inside the constructor deliberately. "Refuse to start on a
    /// missing or wrong key" is a rule about the storage, so it belongs where
    /// the storage is opened: every server mode (`zeb`, `zeb controller`,
    /// `zeb office`, `zeb run`) and every offline command (`zeb project …`,
    /// `zeb admin …`) reaches a data root through this one function, and none
    /// of them can be given a working catalog and a broken keyring.
    fn open_credential_keyring(&mut self) -> Result<(), PlatformError> {
        let Some(data_root) = self.data_root.clone() else {
            return Ok(());
        };
        let stored = self.load_credential_keys()?;
        // Blank is unset, the rule `interface.md` §3a applies to every other
        // variable: an exported-but-empty value is a misconfiguration, not a
        // request to fall back to the file.
        let supplied = std::env::var(CREDENTIAL_KEY_VAR)
            .ok()
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty());
        let boot = CredentialKeyring::open(
            &data_root.join(CREDENTIAL_KEY_REL),
            supplied.as_deref(),
            &stored,
            now_ts(),
        )
        .map_err(|err| PlatformError::new(credential_key_error_code(&err), err.to_string()))?;
        if let Some(row) = boot.insert {
            self.insert_credential_key(&row)?;
        }
        self.keyring = Some(Arc::new(boot.keyring));
        Ok(())
    }

    /// The keyring, or the rejection the contract names for its absence.
    fn keyring(&self) -> Result<&CredentialKeyring, PlatformError> {
        self.keyring.as_deref().ok_or_else(|| {
            PlatformError::new(
                "PLATFORM_CREDENTIAL_KEY_MISSING",
                "this catalog was opened without a data root, so it has no credential encryption \
                 keyring and will not read or write a credential",
            )
        })
    }

    fn load_credential_keys(&self) -> Result<Vec<WrappedDataKey>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT key_id, wrapped_key, created_at FROM credential_keys ORDER BY key_id ASC",
            )
            .map_err(Self::qe)?;
        let rows = stmt
            .query_map([], |row| {
                Ok(WrappedDataKey {
                    key_id: row.get::<_, i64>(0)? as u32,
                    wrapped: row.get::<_, String>(1)?,
                    created_at: row.get::<_, i64>(2)?,
                })
            })
            .map_err(Self::qe)?
            .collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Self::qe)?;
        Ok(rows)
    }

    fn insert_credential_key(&self, row: &WrappedDataKey) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO credential_keys (key_id, wrapped_key, created_at) VALUES (?1, ?2, ?3)",
            params![row.key_id as i64, &row.wrapped, row.created_at],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    /// The secret half as it goes into `secret_json`: always a `zfc1` envelope.
    ///
    /// The whole secret half, whatever a credential type declared about any
    /// field inside it — "a declaration mistake then costs display safety,
    /// never storage safety".
    fn seal_secret_json(&self, secret: &Value) -> Result<String, PlatformError> {
        let plaintext = serde_json::to_string(secret)?;
        let envelope = self
            .keyring()?
            .seal(&plaintext)
            .map_err(|err| PlatformError::new(credential_key_error_code(&err), err.to_string()))?;
        Ok(serde_json::to_string(&Value::String(envelope))?)
    }

    /// The secret half as it comes out of `secret_json`.
    ///
    /// A value that is not an envelope is a row written before this feature
    /// existed. It is accepted and returned as it is, the way `ZebFsAcl`
    /// accepts its own older shape: a credential cannot be regenerated, so
    /// refusing one would destroy something the operator cannot replace. Every
    /// write re-emits it as ciphertext, and
    /// `POST /api/platform/credentials/reencrypt` converts the rest in one call.
    fn open_secret_json(&self, stored: &str) -> Result<Value, PlatformError> {
        let parsed = serde_json::from_str::<Value>(stored).unwrap_or(Value::Null);
        let Some(text) = parsed.as_str() else {
            return Ok(parsed);
        };
        if !crate::infra::secrets::is_secret_envelope(text) {
            return Ok(parsed);
        }
        let plaintext = self
            .keyring()?
            .open_envelope(text)
            .map_err(|err| PlatformError::new(credential_key_error_code(&err), err.to_string()))?;
        Ok(serde_json::from_str::<Value>(&plaintext).unwrap_or(Value::Null))
    }

    fn qe(e: rusqlite::Error) -> PlatformError {
        PlatformError::new("PLATFORM_SQLITE", e.to_string())
    }

    fn json_error(e: serde_json::Error) -> PlatformError {
        PlatformError::new("PLATFORM_SQLITE_JSON", e.to_string())
    }

    fn project_invocations_db_path(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Option<PathBuf>, PlatformError> {
        let Some(data_root) = &self.data_root else {
            return Ok(None);
        };
        let owner = slug_segment(owner);
        let project = slug_segment(project);
        if owner.is_empty() || project.is_empty() {
            return Err(PlatformError::new(
                "PLATFORM_INVOCATION_LOG_PROJECT",
                "owner/project must not be empty for project invocation logs",
            ));
        }
        Ok(Some(
            data_root
                .join("users")
                .join(owner)
                .join(project)
                .join("data")
                .join("logs")
                .join("invocations.db"),
        ))
    }

    fn open_project_invocations_conn(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Option<Connection>, PlatformError> {
        let Some(path) = self.project_invocations_db_path(owner, project)? else {
            return Ok(None);
        };
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let conn = Connection::open(&path)
            .map_err(|e| PlatformError::new("PLATFORM_INVOCATION_LOG_OPEN", e.to_string()))?;
        conn.execute_batch(
            "
            PRAGMA journal_mode=WAL;
            PRAGMA synchronous=NORMAL;
            CREATE TABLE IF NOT EXISTS pipeline_invocations (
                id            INTEGER PRIMARY KEY AUTOINCREMENT,
                file_rel_path TEXT NOT NULL,
                run_id        TEXT NOT NULL DEFAULT '',
                at            INTEGER NOT NULL,
                duration_ms   INTEGER NOT NULL,
                status        TEXT NOT NULL,
                trigger       TEXT NOT NULL DEFAULT '',
                error         TEXT,
                trace_json    TEXT NOT NULL DEFAULT '[]'
            );
            CREATE INDEX IF NOT EXISTS idx_pipeline_invocations_project_pipeline
                ON pipeline_invocations (file_rel_path, at DESC);
            CREATE TABLE IF NOT EXISTS pipeline_error_groups (
                file_rel_path    TEXT NOT NULL,
                signature        TEXT NOT NULL,
                node_id          TEXT NOT NULL DEFAULT '',
                code             TEXT NOT NULL DEFAULT '',
                message_pattern  TEXT NOT NULL DEFAULT '',
                count            INTEGER NOT NULL DEFAULT 0,
                first_seen       INTEGER NOT NULL,
                last_seen        INTEGER NOT NULL,
                reopened_at      INTEGER,
                first_json       TEXT NOT NULL,
                latest_json      TEXT NOT NULL,
                occurrences_json TEXT NOT NULL DEFAULT '[]',
                capture          TEXT NOT NULL DEFAULT 'first-latest',
                captured_json    TEXT NOT NULL DEFAULT '[]',
                PRIMARY KEY (file_rel_path, signature)
            );
            CREATE INDEX IF NOT EXISTS idx_pipeline_error_groups_seen
                ON pipeline_error_groups (last_seen DESC);
            ",
        )
        .map_err(|e| PlatformError::new("PLATFORM_INVOCATION_LOG_SCHEMA", e.to_string()))?;
        Ok(Some(conn))
    }

    fn error_group_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<PipelineErrorGroup> {
        let first: String = row.get("first_json")?;
        let latest: String = row.get("latest_json")?;
        let occurrences: String = row.get("occurrences_json")?;
        let captured: String = row.get("captured_json")?;
        Ok(PipelineErrorGroup {
            file_rel_path: row.get("file_rel_path")?,
            signature: row.get("signature")?,
            node_id: row.get("node_id")?,
            code: row.get("code")?,
            message_pattern: row.get("message_pattern")?,
            count: row.get::<_, i64>("count")? as u64,
            first_seen: row.get("first_seen")?,
            last_seen: row.get("last_seen")?,
            reopened_at: row.get("reopened_at")?,
            first: serde_json::from_str(&first).unwrap_or_else(|_| PipelineInvocationEntry::default()),
            latest: serde_json::from_str(&latest).unwrap_or_else(|_| PipelineInvocationEntry::default()),
            occurrences: serde_json::from_str(&occurrences).unwrap_or_default(),
            capture: row.get("capture")?,
            captured: serde_json::from_str(&captured).unwrap_or_default(),
        })
    }

    fn sha256_hex(input: &str) -> String {
        use sha2::{Digest, Sha256};

        let mut hasher = Sha256::new();
        hasher.update(input.as_bytes());
        let digest = hasher.finalize();
        digest.iter().map(|b| format!("{b:02x}")).collect()
    }

    fn hub_token_scopes(scope_read: bool, scope_publish: bool, scope_manage: bool) -> Vec<String> {
        let mut scopes = Vec::new();
        if scope_read {
            scopes.push("hub:read".to_string());
        }
        if scope_publish {
            scopes.push("hub:publish".to_string());
        }
        if scope_manage {
            scopes.push("hub:manage".to_string());
        }
        scopes
    }

    fn table_has_column<C>(conn: &C, table: &str, column: &str) -> Result<bool, PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        let pragma_sql = format!("PRAGMA table_info({table})");
        let mut stmt = conn.prepare(&pragma_sql).map_err(Self::qe)?;
        let exists = stmt
            .query_map([], |row| row.get::<_, String>(1))
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .any(|name| name == column);
        Ok(exists)
    }

    fn migrations() -> [MigrationDef; 21] {
        [
            MigrationDef {
                version: 1,
                name: "initial_catalog",
                apply: Self::apply_migration_0001_initial_catalog,
            },
            MigrationDef {
                version: 2,
                name: "pipeline_invocation_run_id",
                apply: Self::apply_migration_0002_pipeline_invocation_run_id,
            },
            MigrationDef {
                version: 3,
                name: "hub_columns_and_publishers",
                apply: Self::apply_migration_0003_hub_columns_and_publishers,
            },
            MigrationDef {
                version: 4,
                name: "stable_ids_and_user_local_auth",
                apply: Self::apply_migration_0004_stable_ids_and_user_local_auth,
            },
            MigrationDef {
                version: 5,
                name: "authorities_offices_and_runtime_normalization",
                apply: Self::apply_migration_0005_authorities_offices_and_runtime_normalization,
            },
            MigrationDef {
                version: 6,
                name: "hub_and_operations_internal_ids",
                apply: Self::apply_migration_0006_hub_and_operations_internal_ids,
            },
            MigrationDef {
                version: 7,
                name: "ownership_and_access_internal_ids",
                apply: Self::apply_migration_0007_ownership_and_access_internal_ids,
            },
            MigrationDef {
                version: 8,
                name: "constraint_and_uniqueness_hardening",
                apply: Self::apply_migration_0008_constraint_and_uniqueness_hardening,
            },
            MigrationDef {
                version: 9,
                name: "core_foreign_key_enforcement",
                apply: Self::apply_migration_0009_core_foreign_key_enforcement,
            },
            MigrationDef {
                version: 10,
                name: "project_access_normalization",
                apply: Self::apply_migration_0010_project_access_normalization,
            },
            MigrationDef {
                version: 11,
                name: "hub_token_scope_flags",
                apply: Self::apply_migration_0011_hub_token_scope_flags,
            },
            MigrationDef {
                version: 12,
                name: "platform_service_instances",
                apply: Self::apply_migration_0012_platform_service_instances,
            },
            MigrationDef {
                version: 13,
                name: "hub_publisher_limits",
                apply: Self::apply_migration_0013_hub_publisher_limits,
            },
            MigrationDef {
                version: 14,
                name: "reserved_post_stable_hub_schema",
                apply: Self::apply_migration_0014_reserved_post_stable_hub_schema,
            },
            MigrationDef {
                version: 15,
                name: "hub_access_grants",
                apply: Self::apply_migration_0015_hub_access_grants,
            },
            MigrationDef {
                version: 16,
                name: "user_credential_state",
                apply: Self::apply_migration_0016_user_credential_state,
            },
            MigrationDef {
                version: 17,
                name: "office_join_tokens",
                apply: Self::apply_migration_0017_office_join_tokens,
            },
            MigrationDef {
                version: 18,
                name: "office_vouch_redemptions_and_identity_writes",
                apply: Self::apply_migration_0018_office_vouches_and_identity_writes,
            },
            MigrationDef {
                version: 19,
                name: "office_local_authority",
                apply: Self::apply_migration_0019_office_local_authority,
            },
            MigrationDef {
                version: 20,
                name: "credential_keys",
                apply: Self::apply_migration_0020_credential_keys,
            },
            MigrationDef {
                version: 21,
                name: "member_provenance_is_history_not_reference",
                apply: Self::apply_migration_0021_member_provenance_is_history_not_reference,
            },
        ]
    }

    fn run_migrations(conn: &mut Connection) -> Result<(), PlatformError> {
        conn.execute_batch(SCHEMA_MIGRATIONS_SQL)
            .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))?;

        let applied_versions = {
            let mut stmt = conn
                .prepare("SELECT version FROM schema_migrations ORDER BY version ASC")
                .map_err(Self::qe)?;
            stmt.query_map([], |row| row.get::<_, i64>(0))
                .map_err(Self::qe)?
                .collect::<Result<BTreeSet<_>, _>>()
                .map_err(Self::qe)?
        };

        for migration in Self::migrations() {
            if applied_versions.contains(&migration.version) {
                continue;
            }
            let tx = conn.transaction().map_err(Self::qe)?;
            (migration.apply)(&tx)?;
            tx.execute(
                "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?1, ?2, strftime('%s','now'))",
                params![migration.version, migration.name],
            )
            .map_err(Self::qe)?;
            tx.commit().map_err(Self::qe)?;
        }
        Ok(())
    }

    fn apply_migration_0001_initial_catalog(tx: &Transaction<'_>) -> Result<(), PlatformError> {
        tx.execute_batch(SCHEMA_SQL)
            .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))
    }

    fn apply_migration_0002_pipeline_invocation_run_id(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_table_column(
            tx,
            "pipeline_invocations",
            "run_id",
            "TEXT NOT NULL DEFAULT ''",
        )
    }

    fn apply_migration_0003_hub_columns_and_publishers(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_hub_schema(tx)
    }

    fn apply_migration_0004_stable_ids_and_user_local_auth(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch(
            "
CREATE TABLE IF NOT EXISTS user_local_auth (
    user_id              TEXT PRIMARY KEY,
    password_hash        TEXT NOT NULL DEFAULT '',
    password_alg         TEXT NOT NULL DEFAULT 'sha256',
    password_updated_at  INTEGER NOT NULL DEFAULT 0
);
",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))?;

        Self::ensure_table_column(tx, "users", "user_id", "TEXT NOT NULL DEFAULT ''")?;
        Self::ensure_table_column(tx, "projects", "project_id", "TEXT NOT NULL DEFAULT ''")?;

        if Self::table_has_column(tx, "users", "password")? {
            let mut stmt = tx
                .prepare(
                    "SELECT owner, created_at FROM users WHERE COALESCE(user_id, '') = '' ORDER BY owner ASC",
                )
                .map_err(Self::qe)?;
            let items = stmt
                .query_map([], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
                })
                .map_err(Self::qe)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(Self::qe)?;
            drop(stmt);
            for (owner, created_at) in items {
                let user_id = format!("usr_{}", uuid::Uuid::new_v4().simple());
                tx.execute(
                    "UPDATE users SET user_id = ?1 WHERE owner = ?2",
                    params![user_id, owner],
                )
                .map_err(Self::qe)?;
                let password = tx
                    .query_row(
                        "SELECT password FROM users WHERE owner = ?1",
                        params![owner],
                        |row| row.get::<_, String>(0),
                    )
                    .map_err(Self::qe)?;
                tx.execute(
                    "INSERT OR IGNORE INTO user_local_auth (user_id, password_hash, password_alg, password_updated_at)
                     VALUES (?1, ?2, 'sha256', ?3)",
                    params![user_id, Self::sha256_hex(&password), created_at],
                )
                .map_err(Self::qe)?;
            }
            let mut stmt = tx
                .prepare(
                    "SELECT user_id, owner, password, created_at FROM users
                     WHERE COALESCE(user_id, '') <> '' ORDER BY owner ASC",
                )
                .map_err(Self::qe)?;
            let items = stmt
                .query_map([], |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, i64>(3)?,
                    ))
                })
                .map_err(Self::qe)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(Self::qe)?;
            drop(stmt);
            for (user_id, _owner, password, created_at) in items {
                tx.execute(
                    "INSERT OR IGNORE INTO user_local_auth (user_id, password_hash, password_alg, password_updated_at)
                     VALUES (?1, ?2, 'sha256', ?3)",
                    params![user_id, Self::sha256_hex(&password), created_at],
                )
                .map_err(Self::qe)?;
            }
        }

        {
            let mut stmt = tx
                .prepare(
                    "SELECT owner, project FROM projects WHERE COALESCE(project_id, '') = '' ORDER BY owner ASC, project ASC",
                )
                .map_err(Self::qe)?;
            let items = stmt
                .query_map([], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
                })
                .map_err(Self::qe)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(Self::qe)?;
            drop(stmt);
            for (owner, project) in items {
                let project_id = format!("prj_{}", uuid::Uuid::new_v4().simple());
                tx.execute(
                    "UPDATE projects SET project_id = ?1 WHERE owner = ?2 AND project = ?3",
                    params![project_id, owner, project],
                )
                .map_err(Self::qe)?;
            }
        }

        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_users_user_id ON users(user_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_projects_project_id ON projects(project_id)",
            [],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn apply_migration_0005_authorities_offices_and_runtime_normalization(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch(
            "
CREATE TABLE IF NOT EXISTS hub_authorities (
    authority_id     TEXT PRIMARY KEY,
    host_project_id  TEXT NOT NULL UNIQUE DEFAULT '',
    owner            TEXT NOT NULL DEFAULT '',
    project          TEXT NOT NULL DEFAULT '',
    enabled          INTEGER NOT NULL DEFAULT 0,
    public_base_url  TEXT NOT NULL DEFAULT '',
    created_at       INTEGER NOT NULL DEFAULT 0,
    updated_at       INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS offices (
    office_id     TEXT PRIMARY KEY,
    office_slug   TEXT NOT NULL UNIQUE DEFAULT '',
    label         TEXT NOT NULL DEFAULT '',
    office_kind   TEXT NOT NULL DEFAULT 'office',
    base_url      TEXT NOT NULL DEFAULT '',
    status        TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS platform_service_instances (
    service_instance_id TEXT PRIMARY KEY,
    service_kind        TEXT NOT NULL DEFAULT '',
    display_label       TEXT NOT NULL DEFAULT '',
    host_office_id      TEXT NOT NULL DEFAULT '',
    state_office_id     TEXT NOT NULL DEFAULT '',
    public_base_url     TEXT NOT NULL DEFAULT '',
    enabled             INTEGER NOT NULL DEFAULT 0,
    status              TEXT NOT NULL DEFAULT '',
    placement_generation INTEGER NOT NULL DEFAULT 0,
    created_at          INTEGER NOT NULL DEFAULT 0,
    updated_at          INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS office_nodes (
    node_id             TEXT PRIMARY KEY,
    office_id           TEXT NOT NULL DEFAULT '',
    label               TEXT NOT NULL DEFAULT '',
    base_url            TEXT NOT NULL DEFAULT '',
    status              TEXT NOT NULL DEFAULT '',
    capabilities_json   TEXT NOT NULL DEFAULT '{}',
    registered_at       INTEGER NOT NULL DEFAULT 0,
    last_heartbeat_at   INTEGER NOT NULL DEFAULT 0
);
",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))?;

        Self::ensure_table_column(
            tx,
            "worker_registry",
            "office_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "worker_registry",
            "office_slug",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_runtime_placements",
            "project_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(tx, "project_runtime_placements", "target_office_id", "TEXT")?;
        Self::ensure_table_column(tx, "project_runtime_placements", "target_node_id", "TEXT")?;
        Self::ensure_table_column(
            tx,
            "project_runtime_placements",
            "resource_profile",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_runtime_placements",
            "desired_replicas",
            "INTEGER NOT NULL DEFAULT 1",
        )?;
        Self::ensure_table_column(
            tx,
            "project_runtime_placements",
            "effective_state",
            "TEXT NOT NULL DEFAULT ''",
        )?;

        tx.execute(
            "UPDATE worker_registry
             SET office_id = node_id
             WHERE COALESCE(office_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE worker_registry
             SET office_slug = office_id
             WHERE COALESCE(office_slug, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "INSERT OR REPLACE INTO offices
             (office_id, office_slug, label, office_kind, base_url, status, created_at, updated_at)
             SELECT office_id,
                    CASE WHEN COALESCE(office_slug, '') = '' THEN office_id ELSE office_slug END,
                    label,
                    'office',
                    base_url,
                    status,
                    registered_at,
                    last_heartbeat_at
             FROM worker_registry",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "INSERT OR REPLACE INTO office_nodes
             (node_id, office_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at)
             SELECT node_id, office_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at
             FROM worker_registry",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_runtime_placements
             SET project_id = (
                 SELECT projects.project_id
                 FROM projects
                 WHERE projects.owner = project_runtime_placements.owner
                   AND projects.project = project_runtime_placements.project
             )
             WHERE COALESCE(project_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_runtime_placements
             SET target_node_id = worker_id
             WHERE COALESCE(target_node_id, '') = '' AND COALESCE(worker_id, '') <> ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_runtime_placements
             SET target_office_id = target_node_id
             WHERE COALESCE(target_office_id, '') = '' AND COALESCE(target_node_id, '') <> ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_runtime_placements
             SET effective_state = CASE
                 WHEN target = 'worker' AND COALESCE(target_node_id, '') <> '' THEN 'assigned'
                 ELSE 'local'
             END
             WHERE COALESCE(effective_state, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "INSERT OR IGNORE INTO hub_authorities
             (authority_id, host_project_id, owner, project, enabled, public_base_url, created_at, updated_at)
             SELECT 'mka_' || project_id,
                    project_id,
                    owner,
                    project,
                    0,
                    '',
                    created_at,
                    updated_at
             FROM projects
             WHERE COALESCE(project_id, '') <> ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_authorities_host_project_id
             ON hub_authorities(host_project_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_office_nodes_office_id
             ON office_nodes(office_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_runtime_placements_project_id
             ON project_runtime_placements(project_id)",
            [],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn apply_migration_0006_hub_and_operations_internal_ids(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "authority_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "publisher_pk",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(tx, "hub_tokens", "authority_id", "TEXT NOT NULL DEFAULT ''")?;
        Self::ensure_table_column(tx, "hub_tokens", "publisher_pk", "TEXT NOT NULL DEFAULT ''")?;
        Self::ensure_table_column(
            tx,
            "hub_asset_packages",
            "package_pk",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_asset_packages",
            "authority_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_asset_packages",
            "publisher_pk",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_asset_versions",
            "package_pk",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_operations",
            "project_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;

        tx.execute(
            "UPDATE hub_publishers
             SET authority_id = (
                 SELECT authority_id
                 FROM hub_authorities
                 WHERE hub_authorities.owner = hub_publishers.owner
                   AND hub_authorities.project = hub_publishers.project
             )
             WHERE COALESCE(authority_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;

        {
            let mut stmt = tx
                .prepare(
                    "SELECT owner, project, publisher_id
                     FROM hub_publishers
                     WHERE COALESCE(publisher_pk, '') = ''
                     ORDER BY owner ASC, project ASC, publisher_id ASC",
                )
                .map_err(Self::qe)?;
            let items = stmt
                .query_map([], |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                })
                .map_err(Self::qe)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(Self::qe)?;
            drop(stmt);
            for (owner, project, publisher_id) in items {
                let publisher_pk = format!("mpub_{}", uuid::Uuid::new_v4().simple());
                tx.execute(
                    "UPDATE hub_publishers
                     SET publisher_pk = ?1
                     WHERE owner = ?2 AND project = ?3 AND publisher_id = ?4",
                    params![publisher_pk, owner, project, publisher_id],
                )
                .map_err(Self::qe)?;
            }
        }

        tx.execute(
            "UPDATE hub_tokens
             SET authority_id = (
                 SELECT authority_id
                 FROM hub_authorities
                 WHERE hub_authorities.owner = hub_tokens.owner
                   AND hub_authorities.project = hub_tokens.project
             )
             WHERE COALESCE(authority_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE hub_tokens
             SET publisher_pk = (
                 SELECT publisher_pk
                 FROM hub_publishers
                 WHERE hub_publishers.owner = hub_tokens.owner
                   AND hub_publishers.project = hub_tokens.project
                   AND hub_publishers.publisher_id = hub_tokens.publisher_id
             )
             WHERE COALESCE(publisher_pk, '') = ''",
            [],
        )
        .map_err(Self::qe)?;

        {
            let mut stmt = tx
                .prepare(
                    "SELECT package_id
                     FROM hub_asset_packages
                     WHERE COALESCE(package_pk, '') = ''
                     ORDER BY package_id ASC",
                )
                .map_err(Self::qe)?;
            let items = stmt
                .query_map([], |row| row.get::<_, String>(0))
                .map_err(Self::qe)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(Self::qe)?;
            drop(stmt);
            for package_id in items {
                let package_pk = format!("mpkg_{}", uuid::Uuid::new_v4().simple());
                tx.execute(
                    "UPDATE hub_asset_packages SET package_pk = ?1 WHERE package_id = ?2",
                    params![package_pk, package_id],
                )
                .map_err(Self::qe)?;
            }
        }

        tx.execute(
            "UPDATE hub_asset_packages
             SET authority_id = (
                 SELECT authority_id
                 FROM hub_authorities
                 WHERE hub_authorities.owner = hub_asset_packages.authority_owner
                   AND hub_authorities.project = hub_asset_packages.authority_project
             )
             WHERE COALESCE(authority_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE hub_asset_packages
             SET publisher_pk = (
                 SELECT publisher_pk
                 FROM hub_publishers
                 WHERE hub_publishers.owner = hub_asset_packages.authority_owner
                   AND hub_publishers.project = hub_asset_packages.authority_project
                   AND hub_publishers.publisher_id = hub_asset_packages.publisher_id
             )
             WHERE COALESCE(publisher_pk, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE hub_asset_versions
             SET package_pk = (
                 SELECT package_pk
                 FROM hub_asset_packages
                 WHERE hub_asset_packages.package_id = hub_asset_versions.package_id
             )
             WHERE COALESCE(package_pk, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_operations
             SET project_id = (
                 SELECT projects.project_id
                 FROM projects
                 WHERE projects.owner = project_operations.owner
                   AND projects.project = project_operations.project
             )
             WHERE COALESCE(project_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;

        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_publishers_publisher_pk
             ON hub_publishers(publisher_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_asset_packages_package_pk
             ON hub_asset_packages(package_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_tokens_publisher_pk
             ON hub_tokens(publisher_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_asset_versions_package_pk
             ON hub_asset_versions(package_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_operations_project_id
             ON project_operations(project_id, updated_at DESC, operation_id ASC)",
            [],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn apply_migration_0007_ownership_and_access_internal_ids(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_table_column(tx, "projects", "owner_user_id", "TEXT NOT NULL DEFAULT ''")?;
        Self::ensure_table_column(
            tx,
            "platform_hub_repositories",
            "source_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "platform_hub_repositories",
            "owner_user_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_policies",
            "project_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_policy_bindings",
            "project_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_members",
            "project_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            tx,
            "project_members",
            "created_by_user_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;

        tx.execute(
            "UPDATE projects
             SET owner_user_id = (
                 SELECT users.user_id FROM users WHERE users.owner = projects.owner
             )
             WHERE COALESCE(owner_user_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;

        {
            let mut stmt = tx
                .prepare(
                    "SELECT owner, repository_id
                     FROM platform_hub_repositories
                     WHERE COALESCE(source_id, '') = ''
                     ORDER BY owner ASC, repository_id ASC",
                )
                .map_err(Self::qe)?;
            let items = stmt
                .query_map([], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
                })
                .map_err(Self::qe)?
                .collect::<Result<Vec<_>, _>>()
                .map_err(Self::qe)?;
            drop(stmt);
            for (owner, repository_id) in items {
                let source_id = format!("pmr_{}", uuid::Uuid::new_v4().simple());
                tx.execute(
                    "UPDATE platform_hub_repositories
                     SET source_id = ?1
                     WHERE owner = ?2 AND repository_id = ?3",
                    params![source_id, owner, repository_id],
                )
                .map_err(Self::qe)?;
            }
        }

        tx.execute(
            "UPDATE platform_hub_repositories
             SET owner_user_id = (
                 SELECT users.user_id FROM users WHERE users.owner = platform_hub_repositories.owner
             )
             WHERE COALESCE(owner_user_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_policies
             SET project_id = (
                 SELECT projects.project_id
                 FROM projects
                 WHERE projects.owner = project_policies.owner
                   AND projects.project = project_policies.project
             )
             WHERE COALESCE(project_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_policy_bindings
             SET project_id = (
                 SELECT projects.project_id
                 FROM projects
                 WHERE projects.owner = project_policy_bindings.owner
                   AND projects.project = project_policy_bindings.project
             )
             WHERE COALESCE(project_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_members
             SET project_id = (
                 SELECT projects.project_id
                 FROM projects
                 WHERE projects.owner = project_members.owner
                   AND projects.project = project_members.project
             )
             WHERE COALESCE(project_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "UPDATE project_members
             SET created_by_user_id = (
                 SELECT users.user_id
                 FROM users
                 WHERE users.owner = project_members.created_by
             )
             WHERE COALESCE(created_by_user_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;

        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_platform_hub_repositories_source_id
             ON platform_hub_repositories(source_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_projects_owner_user_id
             ON projects(owner_user_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_members_project_id
             ON project_members(project_id, user_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_policies_project_id
             ON project_policies(project_id, policy_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_policy_bindings_project_id
             ON project_policy_bindings(project_id, subject_id, policy_id)",
            [],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn apply_migration_0008_constraint_and_uniqueness_hardening(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_db_connections_slug
             ON project_db_connections(owner, project, connection_slug)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_members_project_user
             ON project_members(project_id, user_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_policies_project_policy
             ON project_policies(project_id, policy_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_policy_bindings_project_subject_policy
             ON project_policy_bindings(project_id, subject_kind, subject_id, policy_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_publishers_authority_publisher
             ON hub_publishers(authority_id, publisher_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_publishers_authority_url
             ON hub_publishers(authority_id, publisher_url)
             WHERE publisher_url <> ''",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_asset_packages_authority_package
             ON hub_asset_packages(authority_id, package_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_asset_versions_packagepk_version
             ON hub_asset_versions(package_pk, version)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_platform_hub_repositories_owner_user
             ON platform_hub_repositories(owner_user_id, enabled, title)",
            [],
        )
        .map_err(Self::qe)?;
        Self::ensure_table_column(
            tx,
            "platform_hub_repositories",
            "visibility",
            "TEXT NOT NULL DEFAULT 'public'",
        )?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_pipeline_meta_project_file
             ON pipeline_meta(owner, project, file_rel_path)",
            [],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn apply_migration_0009_core_foreign_key_enforcement(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch("PRAGMA defer_foreign_keys = ON;")
            .map_err(Self::qe)?;

        Self::assert_non_empty_column(tx, "users", "user_id")?;
        Self::assert_non_empty_column(tx, "projects", "project_id")?;
        Self::assert_non_empty_column(tx, "projects", "owner_user_id")?;
        Self::assert_non_empty_column(tx, "platform_hub_repositories", "source_id")?;
        Self::assert_non_empty_column(tx, "platform_hub_repositories", "owner_user_id")?;
        Self::assert_non_empty_column(tx, "hub_authorities", "authority_id")?;
        Self::assert_non_empty_column(tx, "hub_authorities", "host_project_id")?;
        Self::assert_non_empty_column(tx, "hub_publishers", "authority_id")?;
        Self::assert_non_empty_column(tx, "hub_publishers", "publisher_pk")?;
        Self::assert_non_empty_column(tx, "hub_tokens", "authority_id")?;
        Self::assert_non_empty_column(tx, "hub_tokens", "publisher_pk")?;
        Self::assert_non_empty_column(tx, "hub_asset_packages", "package_pk")?;
        Self::assert_non_empty_column(tx, "hub_asset_packages", "authority_id")?;
        Self::assert_non_empty_column(tx, "hub_asset_packages", "publisher_pk")?;
        Self::assert_non_empty_column(tx, "hub_asset_versions", "package_pk")?;
        Self::assert_non_empty_column(tx, "office_nodes", "office_id")?;
        Self::assert_non_empty_column(tx, "project_runtime_placements", "project_id")?;
        Self::assert_non_empty_column(tx, "project_operations", "project_id")?;

        Self::rebuild_table(
            tx,
            "user_local_auth",
            "
CREATE TABLE user_local_auth (
    user_id              TEXT PRIMARY KEY,
    password_hash        TEXT NOT NULL DEFAULT '',
    password_alg         TEXT NOT NULL DEFAULT 'sha256',
    password_updated_at  INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE CASCADE
)",
            &[
                "user_id",
                "password_hash",
                "password_alg",
                "password_updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "projects",
            "
CREATE TABLE projects (
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    project_id    TEXT NOT NULL UNIQUE DEFAULT '',
    owner_user_id TEXT NOT NULL DEFAULT '',
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project),
    FOREIGN KEY (owner_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "owner",
                "project",
                "project_id",
                "owner_user_id",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "platform_hub_repositories",
            "
CREATE TABLE platform_hub_repositories (
    source_id      TEXT NOT NULL UNIQUE DEFAULT '',
    owner_user_id  TEXT NOT NULL DEFAULT '',
    owner          TEXT NOT NULL,
    repository_id  TEXT NOT NULL,
    title          TEXT NOT NULL DEFAULT '',
    base_url       TEXT NOT NULL DEFAULT '',
    remote_owner   TEXT NOT NULL DEFAULT '',
    remote_project TEXT NOT NULL DEFAULT '',
    read_token     TEXT NOT NULL DEFAULT '',
    kind           TEXT NOT NULL DEFAULT 'api',
    priority       INTEGER NOT NULL DEFAULT 100,
    visibility     TEXT NOT NULL DEFAULT 'public',
    enabled        INTEGER NOT NULL DEFAULT 1,
    created_at     INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, repository_id),
    FOREIGN KEY (owner_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "source_id",
                "owner_user_id",
                "owner",
                "repository_id",
                "title",
                "base_url",
                "remote_owner",
                "remote_project",
                "read_token",
                // `kind` and `priority` are deliberately absent from the copy
                // list: a table rebuilt from a schema that predates them has
                // no such columns to read, and the new table's defaults --
                // `api` and 100 -- are exactly what a row written before
                // static repositories existed means.
                "visibility",
                "enabled",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "hub_authorities",
            "
CREATE TABLE hub_authorities (
    authority_id     TEXT PRIMARY KEY,
    host_project_id  TEXT NOT NULL UNIQUE DEFAULT '',
    owner            TEXT NOT NULL DEFAULT '',
    project          TEXT NOT NULL DEFAULT '',
    enabled          INTEGER NOT NULL DEFAULT 0,
    public_base_url  TEXT NOT NULL DEFAULT '',
    created_at       INTEGER NOT NULL DEFAULT 0,
    updated_at       INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (host_project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "authority_id",
                "host_project_id",
                "owner",
                "project",
                "enabled",
                "public_base_url",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "hub_publishers",
            "
CREATE TABLE hub_publishers (
    authority_id    TEXT NOT NULL DEFAULT '',
    publisher_pk    TEXT NOT NULL UNIQUE DEFAULT '',
    owner           TEXT NOT NULL,
    project         TEXT NOT NULL,
    publisher_id    TEXT NOT NULL,
    display_name    TEXT NOT NULL DEFAULT '',
    publisher_url   TEXT NOT NULL DEFAULT '',
    email           TEXT NOT NULL DEFAULT '',
    description     TEXT NOT NULL DEFAULT '',
    icon_url        TEXT NOT NULL DEFAULT '',
    website_url     TEXT NOT NULL DEFAULT '',
    enabled         INTEGER NOT NULL DEFAULT 1,
    can_read        INTEGER NOT NULL DEFAULT 1,
    can_publish     INTEGER NOT NULL DEFAULT 1,
    can_manage      INTEGER NOT NULL DEFAULT 0,
    max_packages       INTEGER NOT NULL DEFAULT 0,
    max_package_bytes  INTEGER NOT NULL DEFAULT 0,
    max_media_files    INTEGER NOT NULL DEFAULT 0,
    max_image_bytes    INTEGER NOT NULL DEFAULT 0,
    created_at      INTEGER NOT NULL DEFAULT 0,
    updated_at      INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, publisher_id),
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "authority_id",
                "publisher_pk",
                "owner",
                "project",
                "publisher_id",
                "display_name",
                "publisher_url",
                "email",
                "description",
                "icon_url",
                "website_url",
                "enabled",
                "can_read",
                "can_publish",
                "can_manage",
                "max_packages",
                "max_package_bytes",
                "max_media_files",
                "max_image_bytes",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "hub_tokens",
            "
CREATE TABLE hub_tokens (
    token_id                TEXT PRIMARY KEY,
    authority_id            TEXT NOT NULL DEFAULT '',
    publisher_pk            TEXT NOT NULL DEFAULT '',
    owner                   TEXT NOT NULL DEFAULT '',
    project                 TEXT NOT NULL DEFAULT '',
    publisher_id            TEXT NOT NULL DEFAULT '',
    publisher_display_name  TEXT NOT NULL DEFAULT '',
    publisher_url           TEXT NOT NULL DEFAULT '',
    publisher_email         TEXT NOT NULL DEFAULT '',
    title                   TEXT NOT NULL DEFAULT '',
    secret_hash             TEXT NOT NULL DEFAULT '',
    scopes_json             TEXT NOT NULL DEFAULT '[]',
    expires_at              INTEGER,
    last_used_at            INTEGER,
    revoked_at              INTEGER,
    created_at              INTEGER NOT NULL DEFAULT 0,
    updated_at              INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (publisher_pk) REFERENCES hub_publishers(publisher_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "token_id",
                "authority_id",
                "publisher_pk",
                "owner",
                "project",
                "publisher_id",
                "publisher_display_name",
                "publisher_url",
                "publisher_email",
                "title",
                "secret_hash",
                "scopes_json",
                "expires_at",
                "last_used_at",
                "revoked_at",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "hub_asset_packages",
            "
CREATE TABLE hub_asset_packages (
    package_pk              TEXT NOT NULL UNIQUE DEFAULT '',
    authority_id            TEXT NOT NULL DEFAULT '',
    publisher_pk            TEXT NOT NULL DEFAULT '',
    package_id              TEXT PRIMARY KEY,
    authority_owner         TEXT NOT NULL DEFAULT '',
    authority_project       TEXT NOT NULL DEFAULT '',
    publisher_owner         TEXT NOT NULL DEFAULT '',
    publisher_id            TEXT NOT NULL DEFAULT '',
    publisher_display_name  TEXT NOT NULL DEFAULT '',
    publisher_url           TEXT NOT NULL DEFAULT '',
    publisher_email         TEXT NOT NULL DEFAULT '',
    asset_kind              TEXT NOT NULL DEFAULT '',
    title                   TEXT NOT NULL DEFAULT '',
    description             TEXT NOT NULL DEFAULT '',
    summary                 TEXT NOT NULL DEFAULT '',
    description_md          TEXT NOT NULL DEFAULT '',
    image_url               TEXT NOT NULL DEFAULT '',
    media_json              TEXT NOT NULL DEFAULT '[]',
    gallery_json            TEXT NOT NULL DEFAULT '{}',
    visibility              TEXT NOT NULL DEFAULT 'private',
    tags_json               TEXT NOT NULL DEFAULT '[]',
    created_at              INTEGER NOT NULL DEFAULT 0,
    updated_at              INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (publisher_pk) REFERENCES hub_publishers(publisher_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "package_pk",
                "authority_id",
                "publisher_pk",
                "package_id",
                "authority_owner",
                "authority_project",
                "publisher_owner",
                "publisher_id",
                "publisher_display_name",
                "publisher_url",
                "publisher_email",
                "asset_kind",
                "title",
                "description",
                "image_url",
                "visibility",
                "tags_json",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "hub_asset_versions",
            "
CREATE TABLE hub_asset_versions (
    package_pk         TEXT NOT NULL DEFAULT '',
    package_id         TEXT NOT NULL,
    version            TEXT NOT NULL,
    authority_owner    TEXT NOT NULL DEFAULT '',
    authority_project  TEXT NOT NULL DEFAULT '',
    publisher_owner    TEXT NOT NULL DEFAULT '',
    publisher_id       TEXT NOT NULL DEFAULT '',
    source_owner       TEXT NOT NULL DEFAULT '',
    source_project     TEXT NOT NULL DEFAULT '',
    source_kind        TEXT NOT NULL DEFAULT '',
    source_ref         TEXT NOT NULL DEFAULT '',
    artifact_rel_path  TEXT NOT NULL DEFAULT '',
    artifact_sha256    TEXT NOT NULL DEFAULT '',
    manifest_json      TEXT NOT NULL DEFAULT 'null',
    created_at         INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (package_id, version),
    FOREIGN KEY (package_pk) REFERENCES hub_asset_packages(package_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "package_pk",
                "package_id",
                "version",
                "authority_owner",
                "authority_project",
                "publisher_owner",
                "publisher_id",
                "source_owner",
                "source_project",
                "source_kind",
                "source_ref",
                "artifact_rel_path",
                "artifact_sha256",
                "manifest_json",
                "created_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "office_nodes",
            "
CREATE TABLE office_nodes (
    node_id             TEXT PRIMARY KEY,
    office_id           TEXT NOT NULL DEFAULT '',
    label               TEXT NOT NULL DEFAULT '',
    base_url            TEXT NOT NULL DEFAULT '',
    status              TEXT NOT NULL DEFAULT '',
    capabilities_json   TEXT NOT NULL DEFAULT '{}',
    registered_at       INTEGER NOT NULL DEFAULT 0,
    last_heartbeat_at   INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "node_id",
                "office_id",
                "label",
                "base_url",
                "status",
                "capabilities_json",
                "registered_at",
                "last_heartbeat_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "project_runtime_placements",
            "
CREATE TABLE project_runtime_placements (
    project_id         TEXT NOT NULL DEFAULT '',
    owner              TEXT NOT NULL,
    project            TEXT NOT NULL,
    mode               TEXT NOT NULL DEFAULT 'shared',
    target             TEXT NOT NULL DEFAULT 'local',
    target_office_id   TEXT,
    target_node_id     TEXT,
    worker_id          TEXT,
    resource_profile   TEXT NOT NULL DEFAULT '',
    desired_replicas   INTEGER NOT NULL DEFAULT 1,
    effective_state    TEXT NOT NULL DEFAULT '',
    created_at         INTEGER NOT NULL DEFAULT 0,
    updated_at         INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (target_office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (target_node_id) REFERENCES office_nodes(node_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "project_id",
                "owner",
                "project",
                "mode",
                "target",
                "target_office_id",
                "target_node_id",
                "worker_id",
                "resource_profile",
                "desired_replicas",
                "effective_state",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "project_operations",
            "
CREATE TABLE project_operations (
    project_id          TEXT NOT NULL DEFAULT '',
    owner               TEXT NOT NULL,
    project             TEXT NOT NULL,
    operation_id        TEXT NOT NULL,
    kind                TEXT NOT NULL DEFAULT '',
    status              TEXT NOT NULL DEFAULT 'pending',
    current_step        TEXT NOT NULL DEFAULT '',
    source_office_id    TEXT,
    target_office_id    TEXT,
    artifact_rel_path   TEXT,
    artifact_sha256     TEXT,
    artifact_bytes      INTEGER,
    error_message       TEXT,
    retry_count         INTEGER NOT NULL DEFAULT 0,
    created_at          INTEGER NOT NULL DEFAULT 0,
    updated_at          INTEGER NOT NULL DEFAULT 0,
    completed_at        INTEGER,
    PRIMARY KEY (owner, project, operation_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (source_office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (target_office_id) REFERENCES offices(office_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "project_id",
                "owner",
                "project",
                "operation_id",
                "kind",
                "status",
                "current_step",
                "source_office_id",
                "target_office_id",
                "artifact_rel_path",
                "artifact_sha256",
                "artifact_bytes",
                "error_message",
                "retry_count",
                "created_at",
                "updated_at",
                "completed_at",
            ],
        )?;

        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_office_nodes_office_id
             ON office_nodes(office_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_runtime_placements_project_id
             ON project_runtime_placements(project_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_operations_project
             ON project_operations(owner, project, updated_at DESC, operation_id ASC)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_operations_project_id
             ON project_operations(project_id, updated_at DESC, operation_id ASC)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_publishers_publisher_pk
             ON hub_publishers(publisher_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_asset_packages_package_pk
             ON hub_asset_packages(package_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_tokens_publisher_pk
             ON hub_tokens(publisher_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_asset_versions_package_pk
             ON hub_asset_versions(package_pk)",
            [],
        )
        .map_err(Self::qe)?;
        Self::apply_migration_0008_constraint_and_uniqueness_hardening(tx)?;
        let violations = {
            let mut stmt = tx.prepare("PRAGMA foreign_key_check").map_err(Self::qe)?;
            stmt.query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, i64>(3)?,
                ))
            })
            .map_err(Self::qe)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(Self::qe)?
        };
        if let Some((table, rowid, parent, fk)) = violations.into_iter().next() {
            return Err(PlatformError::new(
                "PLATFORM_SQLITE_FK_CHECK_FAILED",
                format!(
                    "foreign key violation after rebuild: table={table} rowid={rowid} parent={parent} fk={fk}"
                ),
            ));
        }
        Ok(())
    }

    fn apply_migration_0010_project_access_normalization(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch("PRAGMA defer_foreign_keys = ON;")
            .map_err(Self::qe)?;
        Self::ensure_table_column(
            tx,
            "project_members",
            "member_user_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;

        tx.execute(
            "UPDATE project_members
             SET member_user_id = (
                 SELECT users.user_id
                 FROM users
                 WHERE users.owner = project_members.user_id
             )
             WHERE COALESCE(member_user_id, '') = ''",
            [],
        )
        .map_err(Self::qe)?;

        Self::assert_non_empty_column(tx, "project_policies", "project_id")?;
        Self::assert_non_empty_column(tx, "project_policy_bindings", "project_id")?;
        Self::assert_non_empty_column(tx, "project_members", "project_id")?;
        Self::assert_non_empty_column(tx, "project_members", "member_user_id")?;
        Self::assert_non_empty_column(tx, "project_members", "created_by_user_id")?;

        Self::rebuild_table(
            tx,
            "project_policies",
            "
CREATE TABLE project_policies (
    project_id        TEXT NOT NULL DEFAULT '',
    owner             TEXT NOT NULL,
    project           TEXT NOT NULL,
    policy_id         TEXT NOT NULL,
    title             TEXT NOT NULL DEFAULT '',
    capabilities_json TEXT NOT NULL DEFAULT '[]',
    managed           INTEGER NOT NULL DEFAULT 0,
    created_at        INTEGER NOT NULL DEFAULT 0,
    updated_at        INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, policy_id),
    UNIQUE (project_id, policy_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "project_id",
                "owner",
                "project",
                "policy_id",
                "title",
                "capabilities_json",
                "managed",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "project_policy_bindings",
            "
CREATE TABLE project_policy_bindings (
    project_id    TEXT NOT NULL DEFAULT '',
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    subject_kind  TEXT NOT NULL,
    subject_id    TEXT NOT NULL,
    policy_id     TEXT NOT NULL,
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, subject_id, policy_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (project_id, policy_id) REFERENCES project_policies(project_id, policy_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "project_id",
                "owner",
                "project",
                "subject_kind",
                "subject_id",
                "policy_id",
                "created_at",
                "updated_at",
            ],
        )?;
        Self::rebuild_table(
            tx,
            "project_members",
            "
CREATE TABLE project_members (
    project_id              TEXT NOT NULL DEFAULT '',
    owner                   TEXT NOT NULL,
    project                 TEXT NOT NULL,
    user_id                 TEXT NOT NULL,
    member_user_id          TEXT NOT NULL DEFAULT '',
    role_preset             TEXT NOT NULL DEFAULT 'reporter',
    custom_policy_ids_json  TEXT NOT NULL DEFAULT '[]',
    mcp_capabilities_json   TEXT NOT NULL DEFAULT '[]',
    created_by              TEXT NOT NULL DEFAULT '',
    created_by_user_id      TEXT NOT NULL DEFAULT '',
    created_at              INTEGER NOT NULL DEFAULT 0,
    updated_at              INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, user_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (member_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (created_by_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "project_id",
                "owner",
                "project",
                "user_id",
                "member_user_id",
                "role_preset",
                "custom_policy_ids_json",
                "mcp_capabilities_json",
                "created_by",
                "created_by_user_id",
                "created_at",
                "updated_at",
            ],
        )?;

        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_policies_project_policy
             ON project_policies(project_id, policy_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_policy_bindings_project_subject_policy
             ON project_policy_bindings(project_id, subject_kind, subject_id, policy_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_members_project_user
             ON project_members(project_id, user_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_members_project_member_user_id
             ON project_members(project_id, member_user_id)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_project_members_member_user_id
             ON project_members(member_user_id)",
            [],
        )
        .map_err(Self::qe)?;

        let violations = {
            let mut stmt = tx.prepare("PRAGMA foreign_key_check").map_err(Self::qe)?;
            stmt.query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, i64>(3)?,
                ))
            })
            .map_err(Self::qe)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(Self::qe)?
        };
        if let Some((table, rowid, parent, fk)) = violations.into_iter().next() {
            return Err(PlatformError::new(
                "PLATFORM_SQLITE_FK_CHECK_FAILED",
                format!(
                    "foreign key violation after access rebuild: table={table} rowid={rowid} parent={parent} fk={fk}"
                ),
            ));
        }
        Ok(())
    }

    fn apply_migration_0011_hub_token_scope_flags(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch("PRAGMA defer_foreign_keys = ON;")
            .map_err(Self::qe)?;
        Self::ensure_table_column(tx, "hub_tokens", "scope_read", "INTEGER NOT NULL DEFAULT 0")?;
        Self::ensure_table_column(
            tx,
            "hub_tokens",
            "scope_publish",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_tokens",
            "scope_manage",
            "INTEGER NOT NULL DEFAULT 0",
        )?;

        if Self::table_has_column(tx, "hub_tokens", "scopes_json")? {
            tx.execute(
                "UPDATE hub_tokens
                 SET scope_read = CASE
                        WHEN COALESCE(scopes_json, '[]') LIKE '%\"hub:read\"%' THEN 1
                        ELSE 0
                     END,
                     scope_publish = CASE
                        WHEN COALESCE(scopes_json, '[]') LIKE '%\"hub:publish\"%' THEN 1
                        ELSE 0
                     END,
                     scope_manage = CASE
                        WHEN COALESCE(scopes_json, '[]') LIKE '%\"hub:manage\"%' THEN 1
                        ELSE 0
                     END",
                [],
            )
            .map_err(Self::qe)?;
        }

        let old_table = "hub_tokens__old";
        tx.execute_batch(&format!("ALTER TABLE hub_tokens RENAME TO {old_table};"))
            .map_err(Self::qe)?;
        tx.execute_batch(
            "
CREATE TABLE hub_tokens (
    token_id                TEXT PRIMARY KEY,
    authority_id            TEXT NOT NULL DEFAULT '',
    publisher_pk            TEXT NOT NULL DEFAULT '',
    owner                   TEXT NOT NULL DEFAULT '',
    project                 TEXT NOT NULL DEFAULT '',
    publisher_id            TEXT NOT NULL DEFAULT '',
    publisher_display_name  TEXT NOT NULL DEFAULT '',
    publisher_url           TEXT NOT NULL DEFAULT '',
    publisher_email         TEXT NOT NULL DEFAULT '',
    title                   TEXT NOT NULL DEFAULT '',
    secret_hash             TEXT NOT NULL DEFAULT '',
    scope_read              INTEGER NOT NULL DEFAULT 0,
    scope_publish           INTEGER NOT NULL DEFAULT 0,
    scope_manage            INTEGER NOT NULL DEFAULT 0,
    expires_at              INTEGER,
    last_used_at            INTEGER,
    revoked_at              INTEGER,
    created_at              INTEGER NOT NULL DEFAULT 0,
    updated_at              INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (authority_id) REFERENCES hub_authorities(authority_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (publisher_pk) REFERENCES hub_publishers(publisher_pk)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
        )
        .map_err(Self::qe)?;
        let old_has_scope_columns = Self::table_has_column(tx, old_table, "scope_read")?;
        let old_has_scopes_json = Self::table_has_column(tx, old_table, "scopes_json")?;
        let insert_sql = if old_has_scope_columns {
            format!(
                "INSERT INTO hub_tokens
                 (token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at)
                 SELECT token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at
                 FROM {old_table}"
            )
        } else if old_has_scopes_json {
            format!(
                "INSERT INTO hub_tokens
                 (token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at)
                 SELECT token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash,
                        CASE WHEN COALESCE(scopes_json, '[]') LIKE '%\"hub:read\"%' THEN 1 ELSE 0 END,
                        CASE WHEN COALESCE(scopes_json, '[]') LIKE '%\"hub:publish\"%' THEN 1 ELSE 0 END,
                        CASE WHEN COALESCE(scopes_json, '[]') LIKE '%\"hub:manage\"%' THEN 1 ELSE 0 END,
                        expires_at, last_used_at, revoked_at, created_at, updated_at
                 FROM {old_table}"
            )
        } else {
            return Err(PlatformError::new(
                "PLATFORM_SQLITE_SCOPE_MIGRATION",
                "hub_tokens has neither scope flags nor scopes_json",
            ));
        };
        tx.execute(&insert_sql, []).map_err(Self::qe)?;
        tx.execute_batch(&format!("DROP TABLE {old_table};"))
            .map_err(Self::qe)?;

        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_tokens_publisher_pk
             ON hub_tokens(publisher_pk)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_tokens_scope_publish
             ON hub_tokens(scope_publish, revoked_at, expires_at)",
            [],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "CREATE INDEX IF NOT EXISTS idx_hub_tokens_scope_read
             ON hub_tokens(scope_read, revoked_at, expires_at)",
            [],
        )
        .map_err(Self::qe)?;

        let violations = {
            let mut stmt = tx.prepare("PRAGMA foreign_key_check").map_err(Self::qe)?;
            stmt.query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, i64>(3)?,
                ))
            })
            .map_err(Self::qe)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(Self::qe)?
        };
        if let Some((table, rowid, parent, fk)) = violations.into_iter().next() {
            return Err(PlatformError::new(
                "PLATFORM_SQLITE_FK_CHECK_FAILED",
                format!(
                    "foreign key violation after token scope rebuild: table={table} rowid={rowid} parent={parent} fk={fk}"
                ),
            ));
        }
        Ok(())
    }

    fn apply_migration_0012_platform_service_instances(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch(
            "
CREATE TABLE IF NOT EXISTS platform_service_instances (
    service_instance_id TEXT PRIMARY KEY,
    service_kind        TEXT NOT NULL DEFAULT '',
    display_label       TEXT NOT NULL DEFAULT '',
    host_office_id      TEXT NOT NULL DEFAULT '',
    state_office_id     TEXT NOT NULL DEFAULT '',
    public_base_url     TEXT NOT NULL DEFAULT '',
    enabled             INTEGER NOT NULL DEFAULT 0,
    status              TEXT NOT NULL DEFAULT '',
    placement_generation INTEGER NOT NULL DEFAULT 0,
    created_at          INTEGER NOT NULL DEFAULT 0,
    updated_at          INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_platform_service_instances_kind
    ON platform_service_instances(service_kind);
CREATE INDEX IF NOT EXISTS idx_platform_service_instances_host
    ON platform_service_instances(host_office_id);
",
        )
        .map_err(Self::qe)
    }

    fn apply_migration_0013_hub_publisher_limits(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "can_read",
            "INTEGER NOT NULL DEFAULT 1",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "can_publish",
            "INTEGER NOT NULL DEFAULT 1",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "can_manage",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "max_packages",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "max_package_bytes",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "max_media_files",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            tx,
            "hub_publishers",
            "max_image_bytes",
            "INTEGER NOT NULL DEFAULT 0",
        )
    }

    fn apply_migration_0014_reserved_post_stable_hub_schema(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_hub_repository_schema(tx)?;
        Ok(())
    }

    fn apply_migration_0015_hub_access_grants(tx: &Transaction<'_>) -> Result<(), PlatformError> {
        Self::ensure_hub_access_grant_schema(tx)?;
        Ok(())
    }

    /// Adds `user_local_auth.credential_state` — `generated` while the
    /// password is a platform-generated one, `chosen` once a person set it.
    ///
    /// Existing rows default to `chosen`: their operators have been running
    /// with those passwords, and retroactively forcing every account through
    /// the change screen is not what this column is for. First boot marks a
    /// freshly generated superadmin `generated` explicitly.
    fn apply_migration_0016_user_credential_state(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::ensure_table_column(
            tx,
            "user_local_auth",
            "credential_state",
            "TEXT NOT NULL DEFAULT 'chosen'",
        )
    }

    /// Per-office join tokens (`offices.md` §8).
    ///
    /// The record belongs in the platform catalog beside `offices` because it
    /// is one office's membership state, and because the foreign key is the
    /// traceability the contract asks for: a token cannot exist without the
    /// office record that says who holds it, and deleting an office takes its
    /// token with it.
    ///
    /// Pre-release: nothing migrates. The shared environment secret was never
    /// a row, so there is nothing to carry forward.
    fn apply_migration_0017_office_join_tokens(tx: &Transaction<'_>) -> Result<(), PlatformError> {
        tx.execute_batch(
            "CREATE TABLE IF NOT EXISTS office_join_tokens (
                office_id     TEXT PRIMARY KEY,
                secret_digest TEXT NOT NULL DEFAULT '',
                status        TEXT NOT NULL DEFAULT 'active',
                note          TEXT NOT NULL DEFAULT '',
                created_at    INTEGER NOT NULL DEFAULT 0,
                last_used_at  INTEGER NOT NULL DEFAULT 0,
                FOREIGN KEY (office_id) REFERENCES offices(office_id)
                    ON UPDATE CASCADE
                    ON DELETE CASCADE
            );",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))
    }

    fn apply_migration_0018_office_vouches_and_identity_writes(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        // Two tables on the *office* side of a join, both in the catalog
        // `instance-directory.md` puts in the STORE tier.
        //
        // `office_vouch_redemptions` is what makes a vouch spendable once. The
        // nonce is the primary key, so the second redemption is refused by the
        // store rather than by a check that could race with itself. There is
        // no foreign key to `offices`: an office does not hold a row about
        // itself, and the office id here is a label on the record, not a
        // reference into a table this instance owns.
        //
        // `office_identity_writes` is `offices.md` §8's log. It is append-only
        // by construction — nothing in the codebase updates or deletes a row —
        // because a log the writer can edit answers no question worth asking.
        tx.execute_batch(
            "CREATE TABLE IF NOT EXISTS office_vouch_redemptions (
                nonce       TEXT PRIMARY KEY,
                office_id   TEXT NOT NULL DEFAULT '',
                identity    TEXT NOT NULL DEFAULT '',
                expires_at  INTEGER NOT NULL DEFAULT 0,
                redeemed_at INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_office_vouch_redemptions_expires_at
                ON office_vouch_redemptions (expires_at);
            CREATE TABLE IF NOT EXISTS office_identity_writes (
                write_id   TEXT PRIMARY KEY,
                office_id  TEXT NOT NULL DEFAULT '',
                owner      TEXT NOT NULL DEFAULT '',
                action     TEXT NOT NULL DEFAULT '',
                role       TEXT NOT NULL DEFAULT '',
                source     TEXT NOT NULL DEFAULT '',
                detail     TEXT NOT NULL DEFAULT '',
                written_at INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_office_identity_writes_written_at
                ON office_identity_writes (written_at);",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))
    }

    /// `offices.md` §6's break-glass record and §7's detach record.
    ///
    /// One table, because they are two acts of the same kind: performed on the
    /// host, changing whether local login works, and read back by the office's
    /// own operator. It also *is* the state — the rule "local login is disabled
    /// while joined" is derived from these rows rather than from a second flag
    /// that could disagree with them.
    ///
    /// No foreign key to `offices`. On an office, `office_id` is the label the
    /// office wears for its controller, not a row this instance owns; on a
    /// controller, the reported copy arrives before nothing and after nothing
    /// that guarantees an office row exists.
    fn apply_migration_0019_office_local_authority(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        tx.execute_batch(
            "CREATE TABLE IF NOT EXISTS office_local_authority (
                event_id         TEXT PRIMARY KEY,
                office_id        TEXT NOT NULL DEFAULT '',
                event            TEXT NOT NULL DEFAULT '',
                join_fingerprint TEXT NOT NULL DEFAULT '',
                owner            TEXT NOT NULL DEFAULT '',
                detail           TEXT NOT NULL DEFAULT '',
                acted_at         INTEGER NOT NULL DEFAULT 0,
                reported_at      INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_office_local_authority_acted_at
                ON office_local_authority (acted_at);",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))
    }

    /// The keyring: one row per data key generation, wrapped under the
    /// instance key that never lives here.
    ///
    /// `key_id` is the number every `zfc1:` envelope names, and the largest is
    /// the one new writes use. Rows are never deleted by rotation — "older keys
    /// stay for reads" is the whole reason the column exists.
    fn apply_migration_0020_credential_keys(tx: &Transaction<'_>) -> Result<(), PlatformError> {
        tx.execute_batch(
            "CREATE TABLE IF NOT EXISTS credential_keys (
                key_id      INTEGER PRIMARY KEY,
                wrapped_key TEXT NOT NULL,
                created_at  INTEGER NOT NULL DEFAULT 0
            );",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))
    }

    /// `created_by_user_id` loses its foreign key onto `users`.
    ///
    /// The column records who created a membership — provenance, not a live
    /// reference. With `ON DELETE RESTRICT` it chained the creator to
    /// existence: a maintainer who invited one person into someone else's
    /// project could never be deleted, blocked by a row about a member who is
    /// not them. `member_user_id` keeps its RESTRICT — that one names the
    /// person the row is *about*, and it is the safety net that makes
    /// `delete_user` fail loudly if a membership survives the sweep.
    fn apply_migration_0021_member_provenance_is_history_not_reference(
        tx: &Transaction<'_>,
    ) -> Result<(), PlatformError> {
        Self::rebuild_table(
            tx,
            "project_members",
            "
CREATE TABLE project_members (
    project_id              TEXT NOT NULL DEFAULT '',
    owner                   TEXT NOT NULL,
    project                 TEXT NOT NULL,
    user_id                 TEXT NOT NULL,
    member_user_id          TEXT NOT NULL DEFAULT '',
    role_preset             TEXT NOT NULL DEFAULT 'reporter',
    custom_policy_ids_json  TEXT NOT NULL DEFAULT '[]',
    mcp_capabilities_json   TEXT NOT NULL DEFAULT '[]',
    created_by              TEXT NOT NULL DEFAULT '',
    created_by_user_id      TEXT NOT NULL DEFAULT '',
    created_at              INTEGER NOT NULL DEFAULT 0,
    updated_at              INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, user_id),
    FOREIGN KEY (project_id) REFERENCES projects(project_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT,
    FOREIGN KEY (member_user_id) REFERENCES users(user_id)
        ON UPDATE CASCADE
        ON DELETE RESTRICT
)",
            &[
                "project_id",
                "owner",
                "project",
                "user_id",
                "member_user_id",
                "role_preset",
                "custom_policy_ids_json",
                "mcp_capabilities_json",
                "created_by",
                "created_by_user_id",
                "created_at",
                "updated_at",
            ],
        )?;
        // The rename-rebuild drops every index on the table; put back the four
        // that earlier migrations created.
        tx.execute_batch(
            "CREATE INDEX IF NOT EXISTS idx_project_members_project_id
                 ON project_members(project_id, user_id);
             CREATE UNIQUE INDEX IF NOT EXISTS idx_project_members_project_user
                 ON project_members(project_id, user_id);
             CREATE UNIQUE INDEX IF NOT EXISTS idx_project_members_project_member_user_id
                 ON project_members(project_id, member_user_id);
             CREATE INDEX IF NOT EXISTS idx_project_members_member_user_id
                 ON project_members(member_user_id);",
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn ensure_table_column<C>(
        conn: &C,
        table: &str,
        column: &str,
        definition: &str,
    ) -> Result<(), PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        let exists = Self::table_has_column(conn, table, column)?;
        if exists {
            return Ok(());
        }
        conn.execute(
            &format!("ALTER TABLE {table} ADD COLUMN {column} {definition}"),
            [],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    /// Adds the presentation columns a package row carries beside its releases.
    ///
    /// They arrived after `hub_asset_packages` existed, so a database written
    /// by an older binary is brought forward here rather than rebuilt.
    fn ensure_hub_asset_package_columns<C>(conn: &C) -> Result<(), PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        for (column, definition) in [
            ("summary", "TEXT NOT NULL DEFAULT ''"),
            ("description_md", "TEXT NOT NULL DEFAULT ''"),
            ("image_url", "TEXT NOT NULL DEFAULT ''"),
            ("media_json", "TEXT NOT NULL DEFAULT '[]'"),
            ("gallery_json", "TEXT NOT NULL DEFAULT '{}'"),
            ("retracted_at", "INTEGER"),
            ("retracted_reason", "TEXT NOT NULL DEFAULT ''"),
        ] {
            Self::ensure_table_column(conn, "hub_asset_packages", column, definition)?;
        }
        Ok(())
    }

    fn ensure_hub_asset_version_columns<C>(conn: &C) -> Result<(), PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        for (column, definition) in [
            ("retracted_at", "INTEGER"),
            ("retracted_reason", "TEXT NOT NULL DEFAULT ''"),
        ] {
            Self::ensure_table_column(conn, "hub_asset_versions", column, definition)?;
        }
        Ok(())
    }

    fn assert_non_empty_column(
        tx: &Transaction<'_>,
        table: &str,
        column: &str,
    ) -> Result<(), PlatformError> {
        let sql = format!("SELECT COUNT(1) FROM {table} WHERE COALESCE(TRIM({column}), '') = ''");
        let count: i64 = tx.query_row(&sql, [], |row| row.get(0)).map_err(Self::qe)?;
        if count > 0 {
            return Err(PlatformError::new(
                "PLATFORM_SQLITE_FK_REBUILD_BLOCKED",
                format!("{table}.{column} has {count} blank rows"),
            ));
        }
        Ok(())
    }

    fn rebuild_table(
        tx: &Transaction<'_>,
        table: &str,
        create_sql: &str,
        columns: &[&str],
    ) -> Result<(), PlatformError> {
        let old_table = format!("{table}__old");
        tx.execute_batch(&format!("ALTER TABLE {table} RENAME TO {old_table};"))
            .map_err(Self::qe)?;
        tx.execute_batch(create_sql).map_err(Self::qe)?;
        let cols = columns.join(", ");
        tx.execute(
            &format!("INSERT INTO {table} ({cols}) SELECT {cols} FROM {old_table}"),
            [],
        )
        .map_err(Self::qe)?;
        tx.execute_batch(&format!("DROP TABLE {old_table};"))
            .map_err(Self::qe)?;
        Ok(())
    }

    fn ensure_hub_schema<C>(conn: &C) -> Result<(), PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        conn.execute_batch(
            "
CREATE TABLE IF NOT EXISTS hub_publishers (
    owner          TEXT NOT NULL,
    project        TEXT NOT NULL,
    publisher_id   TEXT NOT NULL,
    display_name   TEXT NOT NULL DEFAULT '',
    publisher_url  TEXT NOT NULL DEFAULT '',
    email          TEXT NOT NULL DEFAULT '',
    description    TEXT NOT NULL DEFAULT '',
    icon_url       TEXT NOT NULL DEFAULT '',
    website_url    TEXT NOT NULL DEFAULT '',
    enabled        INTEGER NOT NULL DEFAULT 1,
    can_read       INTEGER NOT NULL DEFAULT 1,
    can_publish    INTEGER NOT NULL DEFAULT 1,
    can_manage     INTEGER NOT NULL DEFAULT 0,
    max_packages      INTEGER NOT NULL DEFAULT 0,
    max_package_bytes INTEGER NOT NULL DEFAULT 0,
    max_media_files   INTEGER NOT NULL DEFAULT 0,
    max_image_bytes   INTEGER NOT NULL DEFAULT 0,
    created_at     INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, publisher_id)
);
",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "can_read",
            "INTEGER NOT NULL DEFAULT 1",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "can_publish",
            "INTEGER NOT NULL DEFAULT 1",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "can_manage",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "max_packages",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "max_package_bytes",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "max_media_files",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_publishers",
            "max_image_bytes",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "authority_owner",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "authority_project",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "publisher_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "publisher_display_name",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "publisher_url",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "publisher_email",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_packages",
            "image_url",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_versions",
            "authority_owner",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_versions",
            "authority_project",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_asset_versions",
            "publisher_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(conn, "hub_tokens", "project", "TEXT NOT NULL DEFAULT ''")?;
        Self::ensure_table_column(
            conn,
            "hub_tokens",
            "publisher_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_tokens",
            "publisher_display_name",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_tokens",
            "publisher_url",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "hub_tokens",
            "publisher_email",
            "TEXT NOT NULL DEFAULT ''",
        )
    }

    fn ensure_hub_repository_schema<C>(conn: &C) -> Result<(), PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        conn.execute_batch(
            "
CREATE TABLE IF NOT EXISTS project_hub_repositories (
    owner         TEXT NOT NULL,
    project       TEXT NOT NULL,
    repository_id TEXT NOT NULL,
    title         TEXT NOT NULL DEFAULT '',
    base_url      TEXT NOT NULL DEFAULT '',
    remote_owner  TEXT NOT NULL DEFAULT '',
    remote_project TEXT NOT NULL DEFAULT '',
    read_token    TEXT NOT NULL DEFAULT '',
    visibility    TEXT NOT NULL DEFAULT 'public',
    enabled       INTEGER NOT NULL DEFAULT 1,
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, project, repository_id)
);
CREATE TABLE IF NOT EXISTS platform_hub_repositories (
    source_id     TEXT NOT NULL UNIQUE DEFAULT '',
    owner_user_id TEXT NOT NULL DEFAULT '',
    owner         TEXT NOT NULL,
    repository_id TEXT NOT NULL,
    title         TEXT NOT NULL DEFAULT '',
    base_url      TEXT NOT NULL DEFAULT '',
    remote_owner  TEXT NOT NULL DEFAULT '',
    remote_project TEXT NOT NULL DEFAULT '',
    read_token    TEXT NOT NULL DEFAULT '',
    kind          TEXT NOT NULL DEFAULT 'api',
    priority      INTEGER NOT NULL DEFAULT 100,
    visibility    TEXT NOT NULL DEFAULT 'public',
    enabled       INTEGER NOT NULL DEFAULT 1,
    created_at    INTEGER NOT NULL DEFAULT 0,
    updated_at    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (owner, repository_id)
);
",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))?;
        Self::ensure_table_column(
            conn,
            "project_hub_repositories",
            "visibility",
            "TEXT NOT NULL DEFAULT 'public'",
        )?;
        Self::ensure_table_column(
            conn,
            "platform_hub_repositories",
            "visibility",
            "TEXT NOT NULL DEFAULT 'public'",
        )?;
        Self::ensure_table_column(
            conn,
            "platform_hub_repositories",
            "source_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        Self::ensure_table_column(
            conn,
            "platform_hub_repositories",
            "owner_user_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        // Rows written before static repositories existed are API hubs, and
        // sit behind the official sources until something says otherwise.
        Self::ensure_table_column(
            conn,
            "platform_hub_repositories",
            "kind",
            "TEXT NOT NULL DEFAULT 'api'",
        )?;
        Self::ensure_table_column(
            conn,
            "platform_hub_repositories",
            "priority",
            "INTEGER NOT NULL DEFAULT 100",
        )?;
        Ok(())
    }

    fn ensure_hub_access_grant_schema<C>(conn: &C) -> Result<(), PlatformError>
    where
        C: std::ops::Deref<Target = Connection>,
    {
        conn.execute_batch(
            "
CREATE TABLE IF NOT EXISTS hub_access_grants (
    grant_id       TEXT PRIMARY KEY,
    source_owner   TEXT NOT NULL DEFAULT '',
    source_id      TEXT NOT NULL DEFAULT '',
    repository_id  TEXT NOT NULL DEFAULT '',
    grant_scope    TEXT NOT NULL DEFAULT 'selected_project',
    target_owner   TEXT NOT NULL DEFAULT '',
    target_project TEXT NOT NULL DEFAULT '',
    can_read       INTEGER NOT NULL DEFAULT 1,
    can_publish    INTEGER NOT NULL DEFAULT 0,
    can_manage     INTEGER NOT NULL DEFAULT 0,
    enabled        INTEGER NOT NULL DEFAULT 1,
    created_at     INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_hub_access_grants_unique_target
    ON hub_access_grants(source_id, grant_scope, target_owner, target_project);
CREATE INDEX IF NOT EXISTS idx_hub_access_grants_source
    ON hub_access_grants(source_owner, repository_id, enabled);
CREATE INDEX IF NOT EXISTS idx_hub_access_grants_project
    ON hub_access_grants(target_owner, target_project, enabled);
",
        )
        .map_err(|e| PlatformError::new("PLATFORM_SQLITE_SCHEMA", e.to_string()))?;
        Ok(())
    }

    fn decode_project_operation_kind(raw: &str) -> Result<ProjectOperationKind, PlatformError> {
        match raw {
            "export_bundle" => Ok(ProjectOperationKind::ExportBundle),
            "export_files" => Ok(ProjectOperationKind::ExportFiles),
            "import_bundle" => Ok(ProjectOperationKind::ImportBundle),
            "import_files" => Ok(ProjectOperationKind::ImportFiles),
            "export_full" => Ok(ProjectOperationKind::ExportFull),
            "import_full" => Ok(ProjectOperationKind::ImportFull),
            "platform_import" => Ok(ProjectOperationKind::PlatformImport),
            "rollback_import" => Ok(ProjectOperationKind::RollbackImport),
            other => Err(PlatformError::new(
                "PLATFORM_SQLITE_DECODE",
                format!("unknown project operation kind '{other}'"),
            )),
        }
    }

    fn decode_project_operation_status(raw: &str) -> Result<ProjectOperationStatus, PlatformError> {
        match raw {
            "pending" => Ok(ProjectOperationStatus::Pending),
            "running" => Ok(ProjectOperationStatus::Running),
            "failed" => Ok(ProjectOperationStatus::Failed),
            "completed" => Ok(ProjectOperationStatus::Completed),
            other => Err(PlatformError::new(
                "PLATFORM_SQLITE_DECODE",
                format!("unknown project operation status '{other}'"),
            )),
        }
    }
}

impl DataAdapter for SqliteDataAdapter {
    fn id(&self) -> &'static str {
        "data.sqlite"
    }

    // ──────────────────────── Users ────────────────────────

    fn get_user_auth(&self, owner: &str) -> Result<Option<StoredUser>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT u.user_id, u.owner, u.role, u.git_name, u.git_email, u.created_at, u.updated_at,
                    a.password_hash, a.password_alg, a.password_updated_at, a.credential_state
             FROM users u
             LEFT JOIN user_local_auth a ON a.user_id = u.user_id
             WHERE u.owner = ?1",
            params![owner],
            |row| {
                Ok(StoredUser {
                    profile: PlatformUser {
                        user_id: row.get(0)?,
                        owner: row.get(1)?,
                        role: row.get(2)?,
                        git_name: row.get(3)?,
                        git_email: row.get(4)?,
                        created_at: row.get(5)?,
                        updated_at: row.get(6)?,
                    },
                    auth: PlatformUserLocalAuth {
                        user_id: row.get(0)?,
                        password_hash: row.get::<_, Option<String>>(7)?.unwrap_or_default(),
                        password_alg: row
                            .get::<_, Option<String>>(8)?
                            .unwrap_or_else(|| "sha256".to_string()),
                        password_updated_at: row.get::<_, Option<i64>>(9)?.unwrap_or_default(),
                        credential_state: row
                            .get::<_, Option<String>>(10)?
                            .unwrap_or_else(|| CREDENTIAL_STATE_CHOSEN.to_string()),
                    },
                })
            },
        );
        match result {
            Ok(u) => Ok(Some(u)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn put_user(&self, user: &StoredUser) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO users
             (owner, user_id, role, git_name, git_email, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
             ON CONFLICT(owner) DO UPDATE SET
                 user_id = excluded.user_id,
                 role = excluded.role,
                 git_name = excluded.git_name,
                 git_email = excluded.git_email,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &user.profile.owner,
                &user.profile.user_id,
                &user.profile.role,
                &user.profile.git_name,
                &user.profile.git_email,
                user.profile.created_at,
                user.profile.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        conn.execute(
            "INSERT INTO user_local_auth
             (user_id, password_hash, password_alg, password_updated_at, credential_state)
             VALUES (?1, ?2, ?3, ?4, ?5)
             ON CONFLICT(user_id) DO UPDATE SET
                 password_hash = excluded.password_hash,
                 password_alg = excluded.password_alg,
                 password_updated_at = excluded.password_updated_at,
                 credential_state = excluded.credential_state",
            params![
                &user.auth.user_id,
                &user.auth.password_hash,
                &user.auth.password_alg,
                user.auth.password_updated_at,
                &user.auth.credential_state,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_users(&self) -> Result<Vec<PlatformUser>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT user_id, owner, role, git_name, git_email, created_at, updated_at
                 FROM users ORDER BY owner ASC",
            )
            .map_err(Self::qe)?;
        let users = stmt
            .query_map([], |row| {
                Ok(PlatformUser {
                    user_id: row.get(0)?,
                    owner: row.get(1)?,
                    role: row.get(2)?,
                    git_name: row.get(3)?,
                    git_email: row.get(4)?,
                    created_at: row.get(5)?,
                    updated_at: row.get(6)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .collect();
        Ok(users)
    }

    fn delete_user(&self, owner: &str, user_id: &str) -> Result<(), PlatformError> {
        let mut conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let tx = conn
            .transaction_with_behavior(TransactionBehavior::Exclusive)
            .map_err(Self::qe)?;

        // Rows about the person, not about one of their projects. Their own
        // projects' rows are already gone — the caller transferred or deleted
        // every project first — so what is left is their footprint in other
        // people's space and their platform-level belongings.
        tx.execute(
            "DELETE FROM project_members WHERE member_user_id = ?1",
            params![user_id],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "DELETE FROM project_invites WHERE target_user = ?1",
            params![owner],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "DELETE FROM hub_access_grants WHERE source_owner = ?1 OR target_owner = ?1",
            params![owner],
        )
        .map_err(Self::qe)?;
        tx.execute(
            "DELETE FROM platform_hub_repositories WHERE owner = ?1",
            params![owner],
        )
        .map_err(Self::qe)?;
        tx.execute("DELETE FROM mcp_sessions WHERE owner = ?1", params![owner])
            .map_err(Self::qe)?;
        tx.execute(
            "DELETE FROM user_local_auth WHERE user_id = ?1",
            params![user_id],
        )
        .map_err(Self::qe)?;

        // The users row goes last, so every remaining foreign key onto
        // users(user_id) verifies the sweep: a surviving project or
        // membership makes this DELETE fail loudly instead of orphaning rows.
        let deleted = tx
            .execute("DELETE FROM users WHERE owner = ?1", params![owner])
            .map_err(Self::qe)?;
        if deleted == 0 {
            return Err(PlatformError::new(
                "PLATFORM_USER_NOT_FOUND",
                "no such user",
            ));
        }
        tx.commit().map_err(Self::qe)
    }

    // ──────────────────────── Projects ────────────────────────

    fn get_project(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Option<PlatformProject>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT owner, project, project_id, owner_user_id, created_at, updated_at
             FROM projects WHERE owner = ?1 AND project = ?2",
            params![owner, project],
            |row| {
                Ok(PlatformProject {
                    owner: row.get(0)?,
                    project: row.get(1)?,
                    project_id: row.get(2)?,
                    owner_user_id: row.get(3)?,
                    title: String::new(), // populated from zebflow.yaml by ProjectService
                    created_at: row.get(4)?,
                    updated_at: row.get(5)?,
                })
            },
        );
        match result {
            Ok(p) => Ok(Some(p)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn put_project(&self, project: &PlatformProject) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO projects (owner, project, project_id, owner_user_id, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)
             ON CONFLICT(owner, project) DO UPDATE SET
                 project_id = excluded.project_id,
                 owner_user_id = excluded.owner_user_id,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &project.owner,
                &project.project,
                &project.project_id,
                &project.owner_user_id,
                project.created_at,
                project.updated_at
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_projects(&self, owner: &str) -> Result<Vec<PlatformProject>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT owner, project, project_id, owner_user_id, created_at, updated_at
                 FROM projects WHERE owner = ?1 ORDER BY project ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner], |row| {
                Ok(PlatformProject {
                    owner: row.get(0)?,
                    project: row.get(1)?,
                    project_id: row.get(2)?,
                    owner_user_id: row.get(3)?,
                    title: String::new(),
                    created_at: row.get(4)?,
                    updated_at: row.get(5)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .collect();
        Ok(items)
    }

    /// Deletes a project and every row this database holds *about* that
    /// project.
    ///
    /// Six tables reference `projects(project_id)` with `ON DELETE RESTRICT`,
    /// so deleting the row alone failed with a foreign key error for any
    /// project that had ever been opened — which is every project, since
    /// creation writes policies and a membership. The rest are keyed by
    /// `(owner, project)` with no constraint to catch them, and leaving them
    /// behind is worse than an error: a project recreated under the same slug
    /// would inherit the previous one's credentials and members.
    ///
    /// Every project carries a `hub_authorities` row — a migration backfills one
    /// per project, disabled and empty — so the row alone means nothing and is
    /// deleted with the project. An authority that has *published* something is
    /// different: it is a publisher identity other instances trust, and
    /// dropping its publishers, packages, and tokens as a side effect of
    /// removing a project would delete a catalogue without saying so. That case
    /// is refused, naming what stands in the way.
    fn delete_project(&self, owner: &str, project: &str) -> Result<(), PlatformError> {
        let mut conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let tx = conn
            .transaction_with_behavior(TransactionBehavior::Exclusive)
            .map_err(Self::qe)?;

        let authority_id: Option<String> = tx
            .query_row(
                "SELECT authority_id FROM hub_authorities WHERE owner = ?1 AND project = ?2",
                params![owner, project],
                |row| row.get(0),
            )
            .optional()
            .map_err(Self::qe)?;
        if let Some(authority_id) = authority_id.as_deref() {
            let mut published = Vec::new();
            for (label, sql) in [
                (
                    "publisher",
                    "SELECT COUNT(*) FROM hub_publishers WHERE authority_id = ?1",
                ),
                (
                    "package",
                    "SELECT COUNT(*) FROM hub_asset_packages WHERE authority_id = ?1",
                ),
                (
                    "token",
                    "SELECT COUNT(*) FROM hub_tokens WHERE authority_id = ?1",
                ),
            ] {
                let count: i64 = tx
                    .query_row(sql, params![authority_id], |row| row.get(0))
                    .map_err(Self::qe)?;
                if count > 0 {
                    published.push(format!("{count} {label}(s)"));
                }
            }
            if !published.is_empty() {
                return Err(PlatformError::new(
                    "PLATFORM_PROJECT_HOSTS_HUB_AUTHORITY",
                    format!(
                        "{owner}/{project} hosts a hub authority with {}; remove them before the \
                         project so a published catalogue is not deleted as a side effect",
                        published.join(", ")
                    ),
                ));
            }
            tx.execute(
                "DELETE FROM hub_authorities WHERE authority_id = ?1",
                params![authority_id],
            )
            .map_err(Self::qe)?;
        }

        // Bindings before policies: they carry a composite key into them.
        for table in [
            "project_policy_bindings",
            "project_policies",
            "project_members",
            "project_invites",
            "project_runtime_placements",
            "project_operations",
            "project_credentials",
            "project_db_connections",
            "project_hub_repositories",
            "pipeline_invocations",
            "pipeline_meta",
            "mcp_sessions",
        ] {
            tx.execute(
                &format!("DELETE FROM {table} WHERE owner = ?1 AND project = ?2"),
                params![owner, project],
            )
            .map_err(Self::qe)?;
        }

        tx.execute(
            "DELETE FROM projects WHERE owner = ?1 AND project = ?2",
            params![owner, project],
        )
        .map_err(Self::qe)?;
        tx.commit().map_err(Self::qe)?;
        Ok(())
    }

    // ──────────────────────── Credentials ────────────────────────

    fn get_project_credential(
        &self,
        owner: &str,
        project: &str,
        credential_id: &str,
    ) -> Result<Option<ProjectCredential>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT owner, project, credential_id, title, kind, secret_json, notes, created_at, updated_at
             FROM project_credentials WHERE owner = ?1 AND project = ?2 AND credential_id = ?3",
            params![owner, project, credential_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, i64>(7)?,
                    row.get::<_, i64>(8)?,
                ))
            },
        );
        match result {
            Ok((
                owner,
                project,
                credential_id,
                title,
                kind,
                secret_json,
                notes,
                created_at,
                updated_at,
            )) => {
                let secret = self.open_secret_json(&secret_json)?;
                Ok(Some(ProjectCredential {
                    owner,
                    project,
                    credential_id,
                    title,
                    kind,
                    secret,
                    notes,
                    created_at,
                    updated_at,
                }))
            }
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn put_project_credential(&self, cred: &ProjectCredential) -> Result<(), PlatformError> {
        let secret_json = self.seal_secret_json(&cred.secret)?;
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT OR REPLACE INTO project_credentials
             (owner, project, credential_id, title, kind, secret_json, notes, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                &cred.owner,
                &cred.project,
                &cred.credential_id,
                &cred.title,
                &cred.kind,
                &secret_json,
                &cred.notes,
                cred.created_at,
                cred.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_credentials(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectCredential>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT owner, project, credential_id, title, kind, secret_json, notes, created_at, updated_at
                 FROM project_credentials WHERE owner = ?1 AND project = ?2
                 ORDER BY credential_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, i64>(7)?,
                    row.get::<_, i64>(8)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .map(
                |(
                    owner,
                    project,
                    credential_id,
                    title,
                    kind,
                    secret_json,
                    notes,
                    created_at,
                    updated_at,
                )| {
                    Ok(ProjectCredential {
                        owner,
                        project,
                        credential_id,
                        title,
                        kind,
                        secret: self.open_secret_json(&secret_json)?,
                        notes,
                        created_at,
                        updated_at,
                    })
                },
            )
            .collect::<Result<Vec<_>, PlatformError>>()?;
        Ok(items)
    }

    fn delete_project_credential(
        &self,
        owner: &str,
        project: &str,
        credential_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_credentials WHERE owner = ?1 AND project = ?2 AND credential_id = ?3",
            params![owner, project, credential_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    // ──────────────────── Credential encryption keyring ────────────────────

    fn credential_keyring_report(&self) -> Result<CredentialKeyringReport, PlatformError> {
        let keyring = self.keyring()?;
        let created = self
            .load_credential_keys()?
            .into_iter()
            .map(|row| (row.key_id, row.created_at))
            .collect::<std::collections::HashMap<_, _>>();
        let current_key_id = keyring.current_key_id();
        Ok(CredentialKeyringReport {
            source: match keyring.source() {
                crate::infra::secrets::KeySource::File => "file".to_string(),
                crate::infra::secrets::KeySource::Environment => "environment".to_string(),
            },
            key_file: self
                .data_root
                .as_ref()
                .map(|root| root.join(CREDENTIAL_KEY_REL).display().to_string())
                .unwrap_or_default(),
            current_key_id,
            generations: keyring
                .key_ids()
                .into_iter()
                .map(|key_id| CredentialKeyGeneration {
                    key_id,
                    created_at: created.get(&key_id).copied().unwrap_or_default(),
                    current: key_id == current_key_id,
                })
                .collect(),
        })
    }

    /// The row is written before the key is used, never after.
    ///
    /// A data key held in memory and missing from the catalog would seal
    /// credentials that nothing could ever open again — the one ordering
    /// mistake in this feature that is not recoverable.
    fn rotate_credential_key(&self) -> Result<CredentialKeyringReport, PlatformError> {
        let keyring = self.keyring()?;
        let (row, key) = keyring
            .prepare_rotation(now_ts())
            .map_err(|err| PlatformError::new(credential_key_error_code(&err), err.to_string()))?;
        self.insert_credential_key(&row)?;
        keyring.adopt(row.key_id, key);
        self.credential_keyring_report()
    }

    /// Three steps, in the order that survives being interrupted between any
    /// two of them: the file gains the new key beside the old, the rows move
    /// to the new key in one transaction, the file drops the old key.
    fn rekey_credential_keyring(&self) -> Result<CredentialKeyringReport, PlatformError> {
        let keyring = self.keyring()?;
        let rewrapped = keyring
            .begin_rekey()
            .map_err(|err| PlatformError::new(credential_key_error_code(&err), err.to_string()))?;
        {
            let mut conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
            let tx = conn
                .transaction_with_behavior(TransactionBehavior::Exclusive)
                .map_err(Self::qe)?;
            for row in &rewrapped {
                tx.execute(
                    "UPDATE credential_keys SET wrapped_key = ?1 WHERE key_id = ?2",
                    params![&row.wrapped, row.key_id as i64],
                )
                .map_err(Self::qe)?;
            }
            tx.commit().map_err(Self::qe)?;
        }
        keyring
            .finish_rekey()
            .map_err(|err| PlatformError::new(credential_key_error_code(&err), err.to_string()))?;
        self.credential_keyring_report()
    }

    fn reencrypt_project_credentials(&self) -> Result<CredentialSweepReport, PlatformError> {
        let key_id = self.keyring()?.current_key_id();
        let stored: Vec<(String, String, String, String)> = {
            let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
            let mut stmt = conn
                .prepare(
                    "SELECT owner, project, credential_id, secret_json FROM project_credentials",
                )
                .map_err(Self::qe)?;
            let rows = stmt
                .query_map([], |row| {
                    Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
                })
                .map_err(Self::qe)?
                .collect::<rusqlite::Result<Vec<_>>>()
                .map_err(Self::qe)?;
            rows
        };

        let mut report = CredentialSweepReport {
            key_id,
            examined: stored.len(),
            rewritten: 0,
            migrated_from_plaintext: 0,
            compacted: false,
        };
        for (owner, project, credential_id, secret_json) in stored {
            // Which generation, if any, this row is already sealed under.
            let generation = serde_json::from_str::<Value>(&secret_json)
                .ok()
                .and_then(|value| value.as_str().map(ToString::to_string))
                .and_then(|text| {
                    crate::infra::secrets::parse_secret_envelope(&text).map(|(id, _)| id)
                });
            if generation == Some(key_id) {
                continue;
            }
            let was_plaintext = generation.is_none();
            let secret = self.open_secret_json(&secret_json)?;
            let sealed = self.seal_secret_json(&secret)?;
            let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
            conn.execute(
                "UPDATE project_credentials SET secret_json = ?1
                 WHERE owner = ?2 AND project = ?3 AND credential_id = ?4",
                params![&sealed, &owner, &project, &credential_id],
            )
            .map_err(Self::qe)?;
            report.rewritten += 1;
            if was_plaintext {
                report.migrated_from_plaintext += 1;
            }
        }

        // The superseded bytes are still in the file until the pages holding
        // them are reused, and a plaintext row migrated by this sweep is
        // exactly the case where that matters. Checkpoint the WAL back into the
        // database, then rebuild it. Best effort — a rebuild that cannot take
        // the lock does not undo work that already committed.
        if report.rewritten > 0 {
            let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
            report.compacted = conn
                .execute_batch("PRAGMA wal_checkpoint(TRUNCATE); VACUUM;")
                .is_ok();
        }
        Ok(report)
    }

    // ──────────────────────── DB Connections ────────────────────────

    fn get_project_db_connection(
        &self,
        owner: &str,
        project: &str,
        connection_slug: &str,
    ) -> Result<Option<ProjectDbConnection>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT owner, project, connection_id, connection_slug, connection_label,
                    database_kind, credential_id, config_json, created_at, updated_at
             FROM project_db_connections
             WHERE owner = ?1 AND project = ?2 AND connection_slug = ?3",
            params![owner, project, connection_slug],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, Option<String>>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, i64>(8)?,
                    row.get::<_, i64>(9)?,
                ))
            },
        );
        match result {
            Ok((
                owner,
                project,
                connection_id,
                connection_slug,
                connection_label,
                database_kind,
                credential_id,
                config_json,
                created_at,
                updated_at,
            )) => {
                let config = serde_json::from_str::<Value>(&config_json).unwrap_or(Value::Null);
                Ok(Some(ProjectDbConnection {
                    owner,
                    project,
                    connection_id,
                    connection_slug,
                    connection_label,
                    database_kind,
                    credential_id,
                    config,
                    created_at,
                    updated_at,
                }))
            }
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn put_project_db_connection(
        &self,
        conn_rec: &ProjectDbConnection,
    ) -> Result<(), PlatformError> {
        let config_json = serde_json::to_string(&conn_rec.config)?;
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT OR REPLACE INTO project_db_connections
             (owner, project, connection_id, connection_slug, connection_label,
              database_kind, credential_id, config_json, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
            params![
                &conn_rec.owner,
                &conn_rec.project,
                &conn_rec.connection_id,
                &conn_rec.connection_slug,
                &conn_rec.connection_label,
                &conn_rec.database_kind,
                conn_rec.credential_id.as_deref(),
                &config_json,
                conn_rec.created_at,
                conn_rec.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_db_connections(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectDbConnection>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT owner, project, connection_id, connection_slug, connection_label,
                        database_kind, credential_id, config_json, created_at, updated_at
                 FROM project_db_connections WHERE owner = ?1 AND project = ?2
                 ORDER BY connection_slug ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, Option<String>>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, i64>(8)?,
                    row.get::<_, i64>(9)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .map(
                |(
                    owner,
                    project,
                    connection_id,
                    connection_slug,
                    connection_label,
                    database_kind,
                    credential_id,
                    config_json,
                    created_at,
                    updated_at,
                )| {
                    let config = serde_json::from_str::<Value>(&config_json).unwrap_or(Value::Null);
                    ProjectDbConnection {
                        owner,
                        project,
                        connection_id,
                        connection_slug,
                        connection_label,
                        database_kind,
                        credential_id,
                        config,
                        created_at,
                        updated_at,
                    }
                },
            )
            .collect();
        Ok(items)
    }

    fn delete_project_db_connection(
        &self,
        owner: &str,
        project: &str,
        connection_slug: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_db_connections
             WHERE owner = ?1 AND project = ?2 AND connection_slug = ?3",
            params![owner, project, connection_slug],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_project_hub_repository(
        &self,
        repository: &ProjectHubRepository,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_repository_schema(&conn)?;
        conn.execute(
            "INSERT OR REPLACE INTO project_hub_repositories
             (owner, project, repository_id, title, base_url, remote_owner, remote_project, read_token, enabled, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
            params![
                &repository.owner,
                &repository.project,
                &repository.repository_id,
                &repository.title,
                &repository.base_url,
                &repository.remote_owner,
                &repository.remote_project,
                &repository.read_token,
                if repository.enabled { 1 } else { 0 },
                repository.created_at,
                repository.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_hub_repositories(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectHubRepository>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_repository_schema(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT owner, project, repository_id, title, base_url, remote_owner, remote_project, read_token, enabled, created_at, updated_at
                 FROM project_hub_repositories WHERE owner = ?1 AND project = ?2
                 ORDER BY title ASC, repository_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok(ProjectHubRepository {
                    owner: row.get(0)?,
                    project: row.get(1)?,
                    repository_id: row.get(2)?,
                    title: row.get(3)?,
                    base_url: row.get(4)?,
                    remote_owner: row.get(5)?,
                    remote_project: row.get(6)?,
                    read_token: row.get(7)?,
                    enabled: row.get::<_, i64>(8)? != 0,
                    created_at: row.get(9)?,
                    updated_at: row.get(10)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(Result::ok)
            .collect();
        Ok(items)
    }

    fn delete_project_hub_repository(
        &self,
        owner: &str,
        project: &str,
        repository_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_repository_schema(&conn)?;
        conn.execute(
            "DELETE FROM project_hub_repositories
             WHERE owner = ?1 AND project = ?2 AND repository_id = ?3",
            params![owner, project, repository_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_platform_hub_repository(
        &self,
        repository: &PlatformHubRepository,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_repository_schema(&conn)?;
        conn.execute(
            "INSERT INTO platform_hub_repositories
             (source_id, owner_user_id, owner, repository_id, title, base_url, remote_owner, remote_project, read_token, kind, priority, visibility, enabled, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)
             ON CONFLICT(owner, repository_id) DO UPDATE SET
                 source_id = excluded.source_id,
                 owner_user_id = excluded.owner_user_id,
                 title = excluded.title,
                 base_url = excluded.base_url,
                 remote_owner = excluded.remote_owner,
                 remote_project = excluded.remote_project,
                 read_token = excluded.read_token,
                 kind = excluded.kind,
                 priority = excluded.priority,
                 visibility = excluded.visibility,
                 enabled = excluded.enabled,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &repository.source_id,
                &repository.owner_user_id,
                &repository.owner,
                &repository.repository_id,
                &repository.title,
                &repository.base_url,
                &repository.remote_owner,
                &repository.remote_project,
                &repository.read_token,
                &repository.kind,
                repository.priority,
                &repository.visibility,
                if repository.enabled { 1 } else { 0 },
                repository.created_at,
                repository.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_platform_hub_repositories(
        &self,
        owner: &str,
    ) -> Result<Vec<PlatformHubRepository>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_repository_schema(&conn)?;
        let mut stmt = conn
            .prepare(
                // Resolution order is this ORDER BY. `distribution.md` says the
                // configured sources are searched in a stated, stable order and
                // that the first hit wins, so the list a caller walks is sorted
                // by the field that states the order rather than by a display
                // title anyone can rename.
                "SELECT source_id, owner_user_id, owner, repository_id, title, base_url, remote_owner, remote_project, read_token, kind, priority, visibility, enabled, created_at, updated_at
                 FROM platform_hub_repositories WHERE owner = ?1
                 ORDER BY priority ASC, repository_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner], |row| {
                Ok(PlatformHubRepository {
                    source_id: row.get(0)?,
                    owner_user_id: row.get(1)?,
                    owner: row.get(2)?,
                    repository_id: row.get(3)?,
                    title: row.get(4)?,
                    base_url: row.get(5)?,
                    remote_owner: row.get(6)?,
                    remote_project: row.get(7)?,
                    read_token: row.get(8)?,
                    kind: row.get(9)?,
                    priority: row.get(10)?,
                    visibility: row.get(11)?,
                    enabled: row.get::<_, i64>(12)? != 0,
                    created_at: row.get(13)?,
                    updated_at: row.get(14)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(Result::ok)
            .collect();
        Ok(items)
    }

    fn delete_platform_hub_repository(
        &self,
        owner: &str,
        repository_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_repository_schema(&conn)?;
        conn.execute(
            "DELETE FROM platform_hub_repositories
             WHERE owner = ?1 AND repository_id = ?2",
            params![owner, repository_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_hub_access_grant(&self, grant: &HubAccessGrant) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_access_grant_schema(&conn)?;
        conn.execute(
            "INSERT INTO hub_access_grants
             (grant_id, source_owner, source_id, repository_id, grant_scope, target_owner, target_project, can_read, can_publish, can_manage, enabled, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
             ON CONFLICT(grant_id) DO UPDATE SET
                 source_owner = excluded.source_owner,
                 source_id = excluded.source_id,
                 repository_id = excluded.repository_id,
                 grant_scope = excluded.grant_scope,
                 target_owner = excluded.target_owner,
                 target_project = excluded.target_project,
                 can_read = excluded.can_read,
                 can_publish = excluded.can_publish,
                 can_manage = excluded.can_manage,
                 enabled = excluded.enabled,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &grant.grant_id,
                &grant.source_owner,
                &grant.source_id,
                &grant.repository_id,
                &grant.grant_scope,
                &grant.target_owner,
                &grant.target_project,
                if grant.can_read { 1 } else { 0 },
                if grant.can_publish { 1 } else { 0 },
                if grant.can_manage { 1 } else { 0 },
                if grant.enabled { 1 } else { 0 },
                grant.created_at,
                grant.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_hub_access_grants(
        &self,
        source_owner: &str,
    ) -> Result<Vec<HubAccessGrant>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_access_grant_schema(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT grant_id, source_owner, source_id, repository_id, grant_scope, target_owner, target_project, can_read, can_publish, can_manage, enabled, created_at, updated_at
                 FROM hub_access_grants
                 WHERE source_owner = ?1
                 ORDER BY repository_id ASC, grant_scope ASC, target_owner ASC, target_project ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![source_owner], hub_access_grant_from_row)
            .map_err(Self::qe)?
            .filter_map(Result::ok)
            .collect();
        Ok(items)
    }

    fn list_effective_hub_access_grants(
        &self,
        target_owner: &str,
        target_project: &str,
    ) -> Result<Vec<HubAccessGrant>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_access_grant_schema(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT grant_id, source_owner, source_id, repository_id, grant_scope, target_owner, target_project, can_read, can_publish, can_manage, enabled, created_at, updated_at
                 FROM hub_access_grants
                 WHERE enabled = 1
                   AND (
                       grant_scope = 'all_projects'
                       OR (grant_scope = 'selected_project' AND target_owner = ?1 AND target_project = ?2)
                   )
                 ORDER BY repository_id ASC, grant_scope ASC, grant_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(
                params![target_owner, target_project],
                hub_access_grant_from_row,
            )
            .map_err(Self::qe)?
            .filter_map(Result::ok)
            .collect();
        Ok(items)
    }

    fn delete_hub_access_grant(&self, grant_id: &str) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_access_grant_schema(&conn)?;
        conn.execute(
            "DELETE FROM hub_access_grants WHERE grant_id = ?1",
            params![grant_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_hub_publisher(&self, publisher: &HubPublisher) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO hub_publishers
             (authority_id, publisher_pk, owner, project, publisher_id, display_name, publisher_url, email, description, icon_url, website_url, enabled, can_read, can_publish, can_manage, max_packages, max_package_bytes, max_media_files, max_image_bytes, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21)
             ON CONFLICT(owner, project, publisher_id) DO UPDATE SET
                 authority_id = excluded.authority_id,
                 publisher_pk = excluded.publisher_pk,
                 display_name = excluded.display_name,
                 publisher_url = excluded.publisher_url,
                 email = excluded.email,
                 description = excluded.description,
                 icon_url = excluded.icon_url,
                 website_url = excluded.website_url,
                 enabled = excluded.enabled,
                 can_read = excluded.can_read,
                 can_publish = excluded.can_publish,
                 can_manage = excluded.can_manage,
                 max_packages = excluded.max_packages,
                 max_package_bytes = excluded.max_package_bytes,
                 max_media_files = excluded.max_media_files,
                 max_image_bytes = excluded.max_image_bytes,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &publisher.authority_id,
                &publisher.publisher_pk,
                &publisher.owner,
                &publisher.project,
                &publisher.publisher_id,
                &publisher.display_name,
                &publisher.publisher_url,
                &publisher.email,
                &publisher.description,
                &publisher.icon_url,
                &publisher.website_url,
                if publisher.enabled { 1 } else { 0 },
                if publisher.can_read { 1 } else { 0 },
                if publisher.can_publish { 1 } else { 0 },
                if publisher.can_manage { 1 } else { 0 },
                publisher.max_packages,
                publisher.max_package_bytes,
                publisher.max_media_files,
                publisher.max_image_bytes,
                publisher.created_at,
                publisher.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_hub_publishers(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<HubPublisher>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT authority_id, publisher_pk, owner, project, publisher_id, display_name, publisher_url, email, description, icon_url, website_url, enabled, can_read, can_publish, can_manage, max_packages, max_package_bytes, max_media_files, max_image_bytes, created_at, updated_at
                 FROM hub_publishers WHERE owner = ?1 AND project = ?2
                 ORDER BY display_name ASC, publisher_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok(HubPublisher {
                    authority_id: row.get(0)?,
                    publisher_pk: row.get(1)?,
                    owner: row.get(2)?,
                    project: row.get(3)?,
                    publisher_id: row.get(4)?,
                    display_name: row.get(5)?,
                    publisher_url: row.get(6)?,
                    email: row.get(7)?,
                    description: row.get(8)?,
                    icon_url: row.get(9)?,
                    website_url: row.get(10)?,
                    enabled: row.get::<_, i64>(11)? != 0,
                    can_read: row.get::<_, i64>(12)? != 0,
                    can_publish: row.get::<_, i64>(13)? != 0,
                    can_manage: row.get::<_, i64>(14)? != 0,
                    max_packages: row.get(15)?,
                    max_package_bytes: row.get(16)?,
                    max_media_files: row.get(17)?,
                    max_image_bytes: row.get(18)?,
                    created_at: row.get(19)?,
                    updated_at: row.get(20)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(Result::ok)
            .collect();
        Ok(items)
    }

    fn get_hub_publisher(
        &self,
        owner: &str,
        project: &str,
        publisher_id: &str,
    ) -> Result<Option<HubPublisher>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT authority_id, publisher_pk, owner, project, publisher_id, display_name, publisher_url, email, description, icon_url, website_url, enabled, can_read, can_publish, can_manage, max_packages, max_package_bytes, max_media_files, max_image_bytes, created_at, updated_at
                 FROM hub_publishers WHERE owner = ?1 AND project = ?2 AND publisher_id = ?3",
            )
            .map_err(Self::qe)?;
        match stmt.query_row(params![owner, project, publisher_id], |row| {
            Ok(HubPublisher {
                authority_id: row.get(0)?,
                publisher_pk: row.get(1)?,
                owner: row.get(2)?,
                project: row.get(3)?,
                publisher_id: row.get(4)?,
                display_name: row.get(5)?,
                publisher_url: row.get(6)?,
                email: row.get(7)?,
                description: row.get(8)?,
                icon_url: row.get(9)?,
                website_url: row.get(10)?,
                enabled: row.get::<_, i64>(11)? != 0,
                can_read: row.get::<_, i64>(12)? != 0,
                can_publish: row.get::<_, i64>(13)? != 0,
                can_manage: row.get::<_, i64>(14)? != 0,
                max_packages: row.get(15)?,
                max_package_bytes: row.get(16)?,
                max_media_files: row.get(17)?,
                max_image_bytes: row.get(18)?,
                created_at: row.get(19)?,
                updated_at: row.get(20)?,
            })
        }) {
            Ok(item) => Ok(Some(item)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn delete_hub_publisher(
        &self,
        owner: &str,
        project: &str,
        publisher_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM hub_publishers WHERE owner = ?1 AND project = ?2 AND publisher_id = ?3",
            params![owner, project, publisher_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_hub_asset_package(&self, package: &HubAssetPackage) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_package_columns(&conn)?;
        let tags_json = serde_json::to_string(&package.tags).map_err(Self::json_error)?;
        let media_json = serde_json::to_string(&package.media).map_err(Self::json_error)?;
        let gallery_json = serde_json::to_string(&package.gallery).map_err(Self::json_error)?;
        conn.execute(
            "INSERT INTO hub_asset_packages
             (package_pk, authority_id, publisher_pk, package_id, authority_owner, authority_project, publisher_owner, publisher_id, publisher_display_name, publisher_url, publisher_email, asset_kind, title, description, summary, description_md, image_url, media_json, gallery_json, visibility, tags_json, retracted_at, retracted_reason, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)
             ON CONFLICT(package_id) DO UPDATE SET
                 package_pk = excluded.package_pk,
                 authority_id = excluded.authority_id,
                 publisher_pk = excluded.publisher_pk,
                 authority_owner = excluded.authority_owner,
                 authority_project = excluded.authority_project,
                 publisher_owner = excluded.publisher_owner,
                 publisher_id = excluded.publisher_id,
                 publisher_display_name = excluded.publisher_display_name,
                 publisher_url = excluded.publisher_url,
                 publisher_email = excluded.publisher_email,
                 asset_kind = excluded.asset_kind,
                 title = excluded.title,
                 description = excluded.description,
                 summary = excluded.summary,
                 description_md = excluded.description_md,
                 image_url = excluded.image_url,
                 media_json = excluded.media_json,
                 gallery_json = excluded.gallery_json,
                 visibility = excluded.visibility,
                 tags_json = excluded.tags_json,
                 retracted_at = excluded.retracted_at,
                 retracted_reason = excluded.retracted_reason,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &package.package_pk,
                &package.authority_id,
                &package.publisher_pk,
                &package.package_id,
                &package.authority_owner,
                &package.authority_project,
                &package.publisher_owner,
                &package.publisher_id,
                &package.publisher_display_name,
                &package.publisher_url,
                &package.publisher_email,
                &package.asset_kind,
                &package.title,
                &package.description,
                &package.summary,
                &package.description_md,
                &package.image_url,
                &media_json,
                &gallery_json,
                &package.visibility,
                &tags_json,
                package.retracted_at,
                &package.retracted_reason,
                package.created_at,
                package.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_hub_asset_packages(&self) -> Result<Vec<HubAssetPackage>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_package_columns(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT package_pk, authority_id, publisher_pk, package_id, authority_owner, authority_project, publisher_owner, publisher_id, publisher_display_name, publisher_url, publisher_email, asset_kind, title, description, summary, description_md, image_url, media_json, gallery_json, visibility, tags_json, retracted_at, retracted_reason, created_at, updated_at
                 FROM hub_asset_packages
                 ORDER BY updated_at DESC, package_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                Ok(HubAssetPackage {
                    package_pk: row.get(0)?,
                    authority_id: row.get(1)?,
                    publisher_pk: row.get(2)?,
                    package_id: row.get(3)?,
                    authority_owner: row.get(4)?,
                    authority_project: row.get(5)?,
                    publisher_owner: row.get(6)?,
                    publisher_id: row.get(7)?,
                    publisher_display_name: row.get(8)?,
                    publisher_url: row.get(9)?,
                    publisher_email: row.get(10)?,
                    asset_kind: row.get(11)?,
                    title: row.get(12)?,
                    description: row.get(13)?,
                    summary: row.get(14)?,
                    description_md: row.get(15)?,
                    image_url: row.get(16)?,
                    media: serde_json::from_str(&row.get::<_, String>(17)?).unwrap_or_default(),
                    gallery: serde_json::from_str(&row.get::<_, String>(18)?).unwrap_or_default(),
                    visibility: row.get(19)?,
                    tags: serde_json::from_str::<Vec<String>>(&row.get::<_, String>(20)?)
                        .unwrap_or_default(),
                    retracted_at: row.get(21)?,
                    retracted_reason: row.get(22)?,
                    created_at: row.get(23)?,
                    updated_at: row.get(24)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(Result::ok)
            .collect();
        Ok(items)
    }

    fn get_hub_asset_package(
        &self,
        package_id: &str,
    ) -> Result<Option<HubAssetPackage>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_package_columns(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT package_pk, authority_id, publisher_pk, package_id, authority_owner, authority_project, publisher_owner, publisher_id, publisher_display_name, publisher_url, publisher_email, asset_kind, title, description, summary, description_md, image_url, media_json, gallery_json, visibility, tags_json, retracted_at, retracted_reason, created_at, updated_at
                 FROM hub_asset_packages WHERE package_id = ?1",
            )
            .map_err(Self::qe)?;
        match stmt.query_row(params![package_id], |row| {
            Ok(HubAssetPackage {
                package_pk: row.get(0)?,
                authority_id: row.get(1)?,
                publisher_pk: row.get(2)?,
                package_id: row.get(3)?,
                authority_owner: row.get(4)?,
                authority_project: row.get(5)?,
                publisher_owner: row.get(6)?,
                publisher_id: row.get(7)?,
                publisher_display_name: row.get(8)?,
                publisher_url: row.get(9)?,
                publisher_email: row.get(10)?,
                asset_kind: row.get(11)?,
                title: row.get(12)?,
                description: row.get(13)?,
                summary: row.get(14)?,
                description_md: row.get(15)?,
                image_url: row.get(16)?,
                media: serde_json::from_str(&row.get::<_, String>(17)?).unwrap_or_default(),
                gallery: serde_json::from_str(&row.get::<_, String>(18)?).unwrap_or_default(),
                visibility: row.get(19)?,
                tags: serde_json::from_str::<Vec<String>>(&row.get::<_, String>(20)?)
                    .unwrap_or_default(),
                retracted_at: row.get(21)?,
                retracted_reason: row.get(22)?,
                created_at: row.get(23)?,
                updated_at: row.get(24)?,
            })
        }) {
            Ok(item) => Ok(Some(item)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    /// Upsert one hub asset version.
    ///
    /// `created_at` is deliberately left out of the conflict update: it records
    /// when the release was first published and must survive for the life of
    /// the version. The hub service already refuses a republish outright
    /// (`HUB_VERSION_EXISTS`); this keeps the column honest regardless.
    ///
    /// The retraction columns are left out for the same reason, and one more:
    /// retraction is one-way, so no write through this path may clear a marker
    /// and hand a withdrawn coordinate back.
    fn put_hub_asset_version(&self, version: &HubAssetVersion) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_version_columns(&conn)?;
        let manifest_json = serde_json::to_string(&version.manifest).map_err(Self::json_error)?;
        conn.execute(
            "INSERT INTO hub_asset_versions
             (package_pk, package_id, version, authority_owner, authority_project, publisher_owner, publisher_id, source_owner, source_project, source_kind, source_ref, artifact_rel_path, artifact_sha256, manifest_json, retracted_at, retracted_reason, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)
             ON CONFLICT(package_id, version) DO UPDATE SET
                 package_pk = excluded.package_pk,
                 authority_owner = excluded.authority_owner,
                 authority_project = excluded.authority_project,
                 publisher_owner = excluded.publisher_owner,
                 publisher_id = excluded.publisher_id,
                 source_owner = excluded.source_owner,
                 source_project = excluded.source_project,
                 source_kind = excluded.source_kind,
                 source_ref = excluded.source_ref,
                 artifact_rel_path = excluded.artifact_rel_path,
                 artifact_sha256 = excluded.artifact_sha256,
                 manifest_json = excluded.manifest_json",
            params![
                &version.package_pk,
                &version.package_id,
                &version.version,
                &version.authority_owner,
                &version.authority_project,
                &version.publisher_owner,
                &version.publisher_id,
                &version.source_owner,
                &version.source_project,
                &version.source_kind,
                &version.source_ref,
                &version.artifact_rel_path,
                &version.artifact_sha256,
                &manifest_json,
                version.retracted_at,
                &version.retracted_reason,
                version.created_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_hub_asset_versions(
        &self,
        package_id: &str,
    ) -> Result<Vec<HubAssetVersion>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_version_columns(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT package_pk, package_id, version, authority_owner, authority_project, publisher_owner, publisher_id, source_owner, source_project, source_kind, source_ref, artifact_rel_path, artifact_sha256, manifest_json, retracted_at, retracted_reason, created_at
                 FROM hub_asset_versions WHERE package_id = ?1
                 ORDER BY created_at DESC, version DESC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![package_id], |row| {
                Ok(HubAssetVersion {
                    package_pk: row.get(0)?,
                    package_id: row.get(1)?,
                    version: row.get(2)?,
                    authority_owner: row.get(3)?,
                    authority_project: row.get(4)?,
                    publisher_owner: row.get(5)?,
                    publisher_id: row.get(6)?,
                    source_owner: row.get(7)?,
                    source_project: row.get(8)?,
                    source_kind: row.get(9)?,
                    source_ref: row.get(10)?,
                    artifact_rel_path: row.get(11)?,
                    artifact_sha256: row.get(12)?,
                    manifest: serde_json::from_str::<Value>(&row.get::<_, String>(13)?)
                        .unwrap_or(Value::Null),
                    retracted_at: row.get(14)?,
                    retracted_reason: row.get(15)?,
                    created_at: row.get(16)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .collect();
        Ok(items)
    }

    fn get_hub_asset_version(
        &self,
        package_id: &str,
        version: &str,
    ) -> Result<Option<HubAssetVersion>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_version_columns(&conn)?;
        let mut stmt = conn
            .prepare(
                "SELECT package_pk, package_id, version, authority_owner, authority_project, publisher_owner, publisher_id, source_owner, source_project, source_kind, source_ref, artifact_rel_path, artifact_sha256, manifest_json, retracted_at, retracted_reason, created_at
                 FROM hub_asset_versions WHERE package_id = ?1 AND version = ?2",
            )
            .map_err(Self::qe)?;
        match stmt.query_row(params![package_id, version], |row| {
            Ok(HubAssetVersion {
                package_pk: row.get(0)?,
                package_id: row.get(1)?,
                version: row.get(2)?,
                authority_owner: row.get(3)?,
                authority_project: row.get(4)?,
                publisher_owner: row.get(5)?,
                publisher_id: row.get(6)?,
                source_owner: row.get(7)?,
                source_project: row.get(8)?,
                source_kind: row.get(9)?,
                source_ref: row.get(10)?,
                artifact_rel_path: row.get(11)?,
                artifact_sha256: row.get(12)?,
                manifest: serde_json::from_str::<Value>(&row.get::<_, String>(13)?)
                    .unwrap_or(Value::Null),
                retracted_at: row.get(14)?,
                retracted_reason: row.get(15)?,
                created_at: row.get(16)?,
            })
        }) {
            Ok(item) => Ok(Some(item)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    /// Mark one release retracted, and only if it is not already.
    ///
    /// The `retracted_at IS NULL` guard makes the write idempotent and keeps
    /// the first reason and moment, which are the ones a project pinning this
    /// coordinate needs to read.
    fn retract_hub_asset_version(
        &self,
        package_id: &str,
        version: &str,
        retracted_at: i64,
        reason: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        Self::ensure_hub_asset_version_columns(&conn)?;
        conn.execute(
            "UPDATE hub_asset_versions
                SET retracted_at = ?3, retracted_reason = ?4
              WHERE package_id = ?1 AND version = ?2 AND retracted_at IS NULL",
            params![package_id, version, retracted_at, reason],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_hub_token(&self, token: &HubToken) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO hub_tokens
             (token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19)
             ON CONFLICT(token_id) DO UPDATE SET
                 authority_id = excluded.authority_id,
                 publisher_pk = excluded.publisher_pk,
                 owner = excluded.owner,
                 project = excluded.project,
                 publisher_id = excluded.publisher_id,
                 publisher_display_name = excluded.publisher_display_name,
                 publisher_url = excluded.publisher_url,
                 publisher_email = excluded.publisher_email,
                 title = excluded.title,
                 secret_hash = excluded.secret_hash,
                 scope_read = excluded.scope_read,
                 scope_publish = excluded.scope_publish,
                 scope_manage = excluded.scope_manage,
                 expires_at = excluded.expires_at,
                 last_used_at = excluded.last_used_at,
                 revoked_at = excluded.revoked_at,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &token.token_id,
                &token.authority_id,
                &token.publisher_pk,
                &token.owner,
                &token.project,
                &token.publisher_id,
                &token.publisher_display_name,
                &token.publisher_url,
                &token.publisher_email,
                &token.title,
                &token.secret_hash,
                if token.scope_read { 1 } else { 0 },
                if token.scope_publish { 1 } else { 0 },
                if token.scope_manage { 1 } else { 0 },
                token.expires_at,
                token.last_used_at,
                token.revoked_at,
                token.created_at,
                token.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_hub_token(&self, token_id: &str) -> Result<Option<HubToken>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at
                 FROM hub_tokens WHERE token_id = ?1",
            )
            .map_err(Self::qe)?;
        match stmt.query_row(params![token_id], |row| {
            let scope_read = row.get::<_, i64>(11)? != 0;
            let scope_publish = row.get::<_, i64>(12)? != 0;
            let scope_manage = row.get::<_, i64>(13)? != 0;
            Ok(HubToken {
                token_id: row.get(0)?,
                authority_id: row.get(1)?,
                publisher_pk: row.get(2)?,
                owner: row.get(3)?,
                project: row.get(4)?,
                publisher_id: row.get(5)?,
                publisher_display_name: row.get(6)?,
                publisher_url: row.get(7)?,
                publisher_email: row.get(8)?,
                title: row.get(9)?,
                secret_hash: row.get(10)?,
                scopes: Self::hub_token_scopes(scope_read, scope_publish, scope_manage),
                scope_read,
                scope_publish,
                scope_manage,
                expires_at: row.get(14)?,
                last_used_at: row.get(15)?,
                revoked_at: row.get(16)?,
                created_at: row.get(17)?,
                updated_at: row.get(18)?,
            })
        }) {
            Ok(item) => Ok(Some(item)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn list_hub_tokens(&self, owner: &str, project: &str) -> Result<Vec<HubToken>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at
                 FROM hub_tokens WHERE owner = ?1 AND project = ?2
                 ORDER BY updated_at DESC, token_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                let scope_read = row.get::<_, i64>(11)? != 0;
                let scope_publish = row.get::<_, i64>(12)? != 0;
                let scope_manage = row.get::<_, i64>(13)? != 0;
                Ok(HubToken {
                    token_id: row.get(0)?,
                    authority_id: row.get(1)?,
                    publisher_pk: row.get(2)?,
                    owner: row.get(3)?,
                    project: row.get(4)?,
                    publisher_id: row.get(5)?,
                    publisher_display_name: row.get(6)?,
                    publisher_url: row.get(7)?,
                    publisher_email: row.get(8)?,
                    title: row.get(9)?,
                    secret_hash: row.get(10)?,
                    scopes: Self::hub_token_scopes(scope_read, scope_publish, scope_manage),
                    scope_read,
                    scope_publish,
                    scope_manage,
                    expires_at: row.get(14)?,
                    last_used_at: row.get(15)?,
                    revoked_at: row.get(16)?,
                    created_at: row.get(17)?,
                    updated_at: row.get(18)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .collect();
        Ok(items)
    }

    fn list_all_hub_tokens(&self) -> Result<Vec<HubToken>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT token_id, authority_id, publisher_pk, owner, project, publisher_id, publisher_display_name, publisher_url, publisher_email, title, secret_hash, scope_read, scope_publish, scope_manage, expires_at, last_used_at, revoked_at, created_at, updated_at
                 FROM hub_tokens
                 ORDER BY updated_at DESC, token_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                let scope_read = row.get::<_, i64>(11)? != 0;
                let scope_publish = row.get::<_, i64>(12)? != 0;
                let scope_manage = row.get::<_, i64>(13)? != 0;
                Ok(HubToken {
                    token_id: row.get(0)?,
                    authority_id: row.get(1)?,
                    publisher_pk: row.get(2)?,
                    owner: row.get(3)?,
                    project: row.get(4)?,
                    publisher_id: row.get(5)?,
                    publisher_display_name: row.get(6)?,
                    publisher_url: row.get(7)?,
                    publisher_email: row.get(8)?,
                    title: row.get(9)?,
                    secret_hash: row.get(10)?,
                    scopes: Self::hub_token_scopes(scope_read, scope_publish, scope_manage),
                    scope_read,
                    scope_publish,
                    scope_manage,
                    expires_at: row.get(14)?,
                    last_used_at: row.get(15)?,
                    revoked_at: row.get(16)?,
                    created_at: row.get(17)?,
                    updated_at: row.get(18)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .collect();
        Ok(items)
    }

    fn delete_hub_token(&self, token_id: &str) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM hub_tokens WHERE token_id = ?1",
            params![token_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    // ──────────────────────── Pipeline Metadata ────────────────────────

    fn put_pipeline_meta(&self, meta: &PipelineMeta) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT OR REPLACE INTO pipeline_meta
             (owner, project, file_rel_path, name, title, virtual_path, description,
              trigger_kind, hash, active_hash, activated_at, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
            params![
                &meta.owner,
                &meta.project,
                &meta.file_rel_path,
                &meta.name,
                &meta.title,
                &meta.virtual_path,
                &meta.description,
                &meta.trigger_kind,
                &meta.hash,
                meta.active_hash.as_deref(),
                meta.activated_at,
                meta.created_at,
                meta.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn delete_pipeline_meta(
        &self,
        owner: &str,
        project: &str,
        file_rel_path: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM pipeline_meta WHERE owner = ?1 AND project = ?2 AND file_rel_path = ?3",
            params![owner, project, file_rel_path],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_pipeline_meta(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<PipelineMeta>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT owner, project, file_rel_path, name, title, virtual_path, description,
                        trigger_kind, hash, active_hash, activated_at, created_at, updated_at
                 FROM pipeline_meta WHERE owner = ?1 AND project = ?2
                 ORDER BY file_rel_path ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok(PipelineMeta {
                    owner: row.get(0)?,
                    project: row.get(1)?,
                    file_rel_path: row.get(2)?,
                    name: row.get(3)?,
                    title: row.get(4)?,
                    virtual_path: row.get(5)?,
                    description: row.get(6)?,
                    trigger_kind: row.get(7)?,
                    hash: row.get(8)?,
                    active_hash: row.get(9)?,
                    activated_at: row.get(10)?,
                    created_at: row.get(11)?,
                    updated_at: row.get(12)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .collect();
        Ok(items)
    }

    // ──────────────────────── Policies ────────────────────────

    fn put_project_policy(&self, policy: &ProjectPolicy) -> Result<(), PlatformError> {
        let caps_json = serde_json::to_string(&policy.capabilities)?;
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO project_policies
             (project_id, owner, project, policy_id, title, capabilities_json, managed, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
             ON CONFLICT(owner, project, policy_id) DO UPDATE SET
                 project_id = excluded.project_id,
                 title = excluded.title,
                 capabilities_json = excluded.capabilities_json,
                 managed = excluded.managed,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &policy.project_id,
                &policy.owner,
                &policy.project,
                &policy.policy_id,
                &policy.title,
                &caps_json,
                policy.managed as i64,
                policy.created_at,
                policy.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_policies(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectPolicy>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT project_id, owner, project, policy_id, title, capabilities_json, managed, created_at, updated_at
                 FROM project_policies WHERE owner = ?1 AND project = ?2
                 ORDER BY policy_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, i64>(6)?,
                    row.get::<_, i64>(7)?,
                    row.get::<_, i64>(8)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .filter_map(
                |(
                    project_id,
                    owner,
                    project,
                    policy_id,
                    title,
                    caps_json,
                    managed,
                    created_at,
                    updated_at,
                )| {
                    let capabilities: Vec<ProjectCapability> =
                        serde_json::from_str(&caps_json).unwrap_or_default();
                    Some(ProjectPolicy {
                        project_id,
                        owner,
                        project,
                        policy_id,
                        title,
                        capabilities,
                        managed: managed != 0,
                        created_at,
                        updated_at,
                    })
                },
            )
            .collect();
        Ok(items)
    }

    fn delete_project_policy(
        &self,
        owner: &str,
        project: &str,
        policy_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_policies WHERE owner = ?1 AND project = ?2 AND policy_id = ?3",
            params![owner, project, policy_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn put_project_policy_binding(
        &self,
        binding: &ProjectPolicyBinding,
    ) -> Result<(), PlatformError> {
        let subject_kind = binding.subject_kind.key();
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO project_policy_bindings
             (project_id, owner, project, subject_kind, subject_id, policy_id, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
             ON CONFLICT(owner, project, subject_id, policy_id) DO UPDATE SET
                 project_id = excluded.project_id,
                 subject_kind = excluded.subject_kind,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &binding.project_id,
                &binding.owner,
                &binding.project,
                subject_kind,
                &binding.subject_id,
                &binding.policy_id,
                binding.created_at,
                binding.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_policy_bindings(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectPolicyBinding>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT project_id, owner, project, subject_kind, subject_id, policy_id, created_at, updated_at
                 FROM project_policy_bindings WHERE owner = ?1 AND project = ?2
                 ORDER BY subject_kind ASC, subject_id ASC, policy_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, i64>(6)?,
                    row.get::<_, i64>(7)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .filter_map(
                |(
                    project_id,
                    owner,
                    project,
                    subject_kind_str,
                    subject_id,
                    policy_id,
                    created_at,
                    updated_at,
                )| {
                    let subject_kind: ProjectSubjectKind =
                        serde_json::from_value(Value::String(subject_kind_str)).ok()?;
                    Some(ProjectPolicyBinding {
                        project_id,
                        owner,
                        project,
                        subject_kind,
                        subject_id,
                        policy_id,
                        created_at,
                        updated_at,
                    })
                },
            )
            .collect();
        Ok(items)
    }

    fn delete_project_policy_binding(
        &self,
        owner: &str,
        project: &str,
        subject_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_policy_bindings
             WHERE owner = ?1 AND project = ?2 AND subject_id = ?3",
            params![owner, project, subject_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_project_member(
        &self,
        owner: &str,
        project: &str,
        user_id: &str,
    ) -> Result<Option<ProjectMember>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT project_id, owner, project, user_id, member_user_id, role_preset, custom_policy_ids_json, mcp_capabilities_json, created_by, created_by_user_id, created_at, updated_at
             FROM project_members WHERE owner = ?1 AND project = ?2 AND user_id = ?3",
            params![owner, project, user_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                    row.get::<_, String>(9)?,
                    row.get::<_, i64>(10)?,
                    row.get::<_, i64>(11)?,
                ))
            },
        );
        match result {
            Ok((
                project_id,
                owner,
                project,
                user_id,
                member_user_id,
                role_preset,
                custom_policy_ids_json,
                mcp_capabilities_json,
                created_by,
                created_by_user_id,
                created_at,
                updated_at,
            )) => {
                let role_preset: ProjectAccessRolePreset =
                    serde_json::from_value(Value::String(role_preset)).unwrap_or_default();
                let custom_policy_ids: Vec<String> =
                    serde_json::from_str(&custom_policy_ids_json).unwrap_or_default();
                let mcp_capability_ceiling: Vec<ProjectCapability> =
                    serde_json::from_str(&mcp_capabilities_json).unwrap_or_default();
                Ok(Some(ProjectMember {
                    project_id,
                    owner,
                    project,
                    user_id,
                    member_user_id,
                    role_preset,
                    custom_policy_ids,
                    mcp_capability_ceiling,
                    created_by,
                    created_by_user_id,
                    created_at,
                    updated_at,
                }))
            }
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn put_project_member(&self, member: &ProjectMember) -> Result<(), PlatformError> {
        let custom_policy_ids_json = serde_json::to_string(&member.custom_policy_ids)?;
        let mcp_capabilities_json = serde_json::to_string(&member.mcp_capability_ceiling)?;
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO project_members
             (project_id, owner, project, user_id, member_user_id, role_preset, custom_policy_ids_json, mcp_capabilities_json, created_by, created_by_user_id, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
             ON CONFLICT(owner, project, user_id) DO UPDATE SET
                 project_id = excluded.project_id,
                 member_user_id = excluded.member_user_id,
                 role_preset = excluded.role_preset,
                 custom_policy_ids_json = excluded.custom_policy_ids_json,
                 mcp_capabilities_json = excluded.mcp_capabilities_json,
                 created_by = excluded.created_by,
                 created_by_user_id = excluded.created_by_user_id,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &member.project_id,
                &member.owner,
                &member.project,
                &member.user_id,
                &member.member_user_id,
                member.role_preset.key(),
                custom_policy_ids_json,
                mcp_capabilities_json,
                &member.created_by,
                &member.created_by_user_id,
                member.created_at,
                member.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_memberships(
        &self,
        user_id: &str,
    ) -> Result<Vec<ProjectMember>, PlatformError> {
        // Finds the projects first, then reuses `list_project_members` for the
        // rows, so the twelve-column mapping stays in one place.
        let scopes: Vec<(String, String)> = {
            let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
            let mut stmt = conn
                .prepare(
                    "SELECT DISTINCT owner, project FROM project_members WHERE user_id = ?1",
                )
                .map_err(Self::qe)?;
            let rows = stmt
                .query_map(params![user_id], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
                })
                .map_err(Self::qe)?
                .filter_map(|row| row.ok())
                .collect();
            rows
        };

        let mut out = Vec::new();
        for (owner, project) in scopes {
            out.extend(
                self.list_project_members(&owner, &project)?
                    .into_iter()
                    .filter(|member| member.user_id == user_id),
            );
        }
        Ok(out)
    }

    fn list_project_members(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectMember>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT project_id, owner, project, user_id, member_user_id, role_preset, custom_policy_ids_json, mcp_capabilities_json, created_by, created_by_user_id, created_at, updated_at
                 FROM project_members WHERE owner = ?1 AND project = ?2 ORDER BY user_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                    row.get::<_, String>(9)?,
                    row.get::<_, i64>(10)?,
                    row.get::<_, i64>(11)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .filter_map(
                |(
                    project_id,
                    owner,
                    project,
                    user_id,
                    member_user_id,
                    role_preset,
                    custom_policy_ids_json,
                    mcp_capabilities_json,
                    created_by,
                    created_by_user_id,
                    created_at,
                    updated_at,
                )| {
                    let role_preset: ProjectAccessRolePreset =
                        serde_json::from_value(Value::String(role_preset)).ok()?;
                    let custom_policy_ids: Vec<String> =
                        serde_json::from_str(&custom_policy_ids_json).unwrap_or_default();
                    let mcp_capability_ceiling: Vec<ProjectCapability> =
                        serde_json::from_str(&mcp_capabilities_json).unwrap_or_default();
                    Some(ProjectMember {
                        project_id,
                        owner,
                        project,
                        user_id,
                        member_user_id,
                        role_preset,
                        custom_policy_ids,
                        mcp_capability_ceiling,
                        created_by,
                        created_by_user_id,
                        created_at,
                        updated_at,
                    })
                },
            )
            .collect();
        Ok(items)
    }

    fn delete_project_member(
        &self,
        owner: &str,
        project: &str,
        user_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_members WHERE owner = ?1 AND project = ?2 AND user_id = ?3",
            params![owner, project, user_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_project_invite(
        &self,
        owner: &str,
        project: &str,
        invite_id: &str,
    ) -> Result<Option<ProjectInvite>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT owner, project, invite_id, target_user, role_preset, custom_policy_ids_json, mcp_capabilities_json, note, invited_by, status, expires_at, created_at, updated_at
             FROM project_invites WHERE owner = ?1 AND project = ?2 AND invite_id = ?3",
            params![owner, project, invite_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                    row.get::<_, String>(9)?,
                    row.get::<_, Option<i64>>(10)?,
                    row.get::<_, i64>(11)?,
                    row.get::<_, i64>(12)?,
                ))
            },
        );
        match result {
            Ok((
                owner,
                project,
                invite_id,
                target_user,
                role_preset,
                custom_policy_ids_json,
                mcp_capabilities_json,
                note,
                invited_by,
                status,
                expires_at,
                created_at,
                updated_at,
            )) => {
                let role_preset: ProjectAccessRolePreset =
                    serde_json::from_value(Value::String(role_preset)).unwrap_or_default();
                let status: ProjectInviteStatus = serde_json::from_str(&status).unwrap_or_default();
                let custom_policy_ids: Vec<String> =
                    serde_json::from_str(&custom_policy_ids_json).unwrap_or_default();
                let mcp_capability_ceiling: Vec<ProjectCapability> =
                    serde_json::from_str(&mcp_capabilities_json).unwrap_or_default();
                Ok(Some(ProjectInvite {
                    owner,
                    project,
                    invite_id,
                    target_user,
                    role_preset,
                    custom_policy_ids,
                    mcp_capability_ceiling,
                    note,
                    invited_by,
                    status,
                    expires_at,
                    created_at,
                    updated_at,
                }))
            }
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(Self::qe(e)),
        }
    }

    fn put_project_invite(&self, invite: &ProjectInvite) -> Result<(), PlatformError> {
        let custom_policy_ids_json = serde_json::to_string(&invite.custom_policy_ids)?;
        let mcp_capabilities_json = serde_json::to_string(&invite.mcp_capability_ceiling)?;
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT OR REPLACE INTO project_invites
             (owner, project, invite_id, target_user, role_preset, custom_policy_ids_json, mcp_capabilities_json, note, invited_by, status, expires_at, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
            params![
                &invite.owner,
                &invite.project,
                &invite.invite_id,
                &invite.target_user,
                invite.role_preset.key(),
                custom_policy_ids_json,
                mcp_capabilities_json,
                &invite.note,
                &invite.invited_by,
                serde_json::to_string(&invite.status)?,
                invite.expires_at,
                invite.created_at,
                invite.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_invites_for_user(
        &self,
        target_user: &str,
    ) -> Result<Vec<ProjectInvite>, PlatformError> {
        // Finds the projects first, then reuses `list_project_invites` for the
        // rows. The row mapping is forty lines; a second copy of it here would
        // be a second thing to keep in step, and a person is invited to a
        // handful of projects, not thousands.
        let scopes: Vec<(String, String)> = {
            let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
            let mut stmt = conn
                .prepare(
                    "SELECT DISTINCT owner, project FROM project_invites WHERE target_user = ?1",
                )
                .map_err(Self::qe)?;
            let rows = stmt
                .query_map(params![target_user], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
                })
                .map_err(Self::qe)?
                .filter_map(|row| row.ok())
                .collect();
            rows
        };

        let mut out = Vec::new();
        for (owner, project) in scopes {
            out.extend(
                self.list_project_invites(&owner, &project)?
                    .into_iter()
                    .filter(|invite| invite.target_user == target_user),
            );
        }
        Ok(out)
    }

    fn list_project_invites(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Vec<ProjectInvite>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT owner, project, invite_id, target_user, role_preset, custom_policy_ids_json, mcp_capabilities_json, note, invited_by, status, expires_at, created_at, updated_at
                 FROM project_invites WHERE owner = ?1 AND project = ?2 ORDER BY created_at DESC, invite_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, String>(8)?,
                    row.get::<_, String>(9)?,
                    row.get::<_, Option<i64>>(10)?,
                    row.get::<_, i64>(11)?,
                    row.get::<_, i64>(12)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .filter_map(
                |(
                    owner,
                    project,
                    invite_id,
                    target_user,
                    role_preset,
                    custom_policy_ids_json,
                    mcp_capabilities_json,
                    note,
                    invited_by,
                    status,
                    expires_at,
                    created_at,
                    updated_at,
                )| {
                    let role_preset: ProjectAccessRolePreset =
                        serde_json::from_value(Value::String(role_preset)).ok()?;
                    let status: ProjectInviteStatus = serde_json::from_str(&status).ok()?;
                    let custom_policy_ids: Vec<String> =
                        serde_json::from_str(&custom_policy_ids_json).unwrap_or_default();
                    let mcp_capability_ceiling: Vec<ProjectCapability> =
                        serde_json::from_str(&mcp_capabilities_json).unwrap_or_default();
                    Some(ProjectInvite {
                        owner,
                        project,
                        invite_id,
                        target_user,
                        role_preset,
                        custom_policy_ids,
                        mcp_capability_ceiling,
                        note,
                        invited_by,
                        status,
                        expires_at,
                        created_at,
                        updated_at,
                    })
                },
            )
            .collect();
        Ok(items)
    }

    fn delete_project_invite(
        &self,
        owner: &str,
        project: &str,
        invite_id: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_invites WHERE owner = ?1 AND project = ?2 AND invite_id = ?3",
            params![owner, project, invite_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_hub_authority(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Option<HubAuthority>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT authority_id, host_project_id, owner, project, enabled, public_base_url, created_at, updated_at
             FROM hub_authorities
             WHERE owner = ?1 AND project = ?2",
            params![owner, project],
            |row| {
                Ok(HubAuthority {
                    authority_id: row.get(0)?,
                    host_project_id: row.get(1)?,
                    owner: row.get(2)?,
                    project: row.get(3)?,
                    enabled: row.get::<_, i64>(4)? != 0,
                    public_base_url: row.get(5)?,
                    created_at: row.get(6)?,
                    updated_at: row.get(7)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_hub_authority(&self, authority: &HubAuthority) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO hub_authorities
             (authority_id, host_project_id, owner, project, enabled, public_base_url, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
             ON CONFLICT(authority_id) DO UPDATE SET
                 host_project_id = excluded.host_project_id,
                 owner = excluded.owner,
                 project = excluded.project,
                 enabled = excluded.enabled,
                 public_base_url = excluded.public_base_url,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &authority.authority_id,
                &authority.host_project_id,
                &authority.owner,
                &authority.project,
                if authority.enabled { 1 } else { 0 },
                &authority.public_base_url,
                authority.created_at,
                authority.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_hub_authorities(&self) -> Result<Vec<HubAuthority>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT authority_id, host_project_id, owner, project, enabled, public_base_url, created_at, updated_at
                 FROM hub_authorities
                 ORDER BY owner ASC, project ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                Ok(HubAuthority {
                    authority_id: row.get(0)?,
                    host_project_id: row.get(1)?,
                    owner: row.get(2)?,
                    project: row.get(3)?,
                    enabled: row.get::<_, i64>(4)? != 0,
                    public_base_url: row.get(5)?,
                    created_at: row.get(6)?,
                    updated_at: row.get(7)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect();
        Ok(items)
    }

    fn get_platform_office(
        &self,
        office_id: &str,
    ) -> Result<Option<PlatformOffice>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT office_id, office_slug, label, office_kind, base_url, status, created_at, updated_at
             FROM offices
             WHERE office_id = ?1",
            params![office_id],
            |row| {
                Ok(PlatformOffice {
                    office_id: row.get(0)?,
                    office_slug: row.get(1)?,
                    label: row.get(2)?,
                    office_kind: row.get(3)?,
                    base_url: row.get(4)?,
                    status: row.get(5)?,
                    created_at: row.get(6)?,
                    updated_at: row.get(7)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_platform_office(&self, office: &PlatformOffice) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO offices
             (office_id, office_slug, label, office_kind, base_url, status, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
             ON CONFLICT(office_id) DO UPDATE SET
                 office_slug = excluded.office_slug,
                 label = excluded.label,
                 office_kind = excluded.office_kind,
                 base_url = excluded.base_url,
                 status = excluded.status,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &office.office_id,
                &office.office_slug,
                &office.label,
                &office.office_kind,
                &office.base_url,
                &office.status,
                office.created_at,
                office.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_office_join_token(
        &self,
        office_id: &str,
    ) -> Result<Option<PlatformOfficeJoinToken>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT office_id, secret_digest, status, note, created_at, last_used_at
             FROM office_join_tokens
             WHERE office_id = ?1",
            params![office_id],
            |row| {
                Ok(PlatformOfficeJoinToken {
                    office_id: row.get(0)?,
                    secret_digest: row.get(1)?,
                    status: row.get(2)?,
                    note: row.get(3)?,
                    created_at: row.get(4)?,
                    last_used_at: row.get(5)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_office_join_token(&self, token: &PlatformOfficeJoinToken) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO office_join_tokens
             (office_id, secret_digest, status, note, created_at, last_used_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)
             ON CONFLICT(office_id) DO UPDATE SET
                 secret_digest = excluded.secret_digest,
                 status = excluded.status,
                 note = excluded.note,
                 created_at = excluded.created_at,
                 last_used_at = excluded.last_used_at",
            params![
                &token.office_id,
                &token.secret_digest,
                &token.status,
                &token.note,
                token.created_at,
                token.last_used_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn touch_office_join_token(
        &self,
        office_id: &str,
        secret_digest: &str,
        last_used_at: i64,
    ) -> Result<bool, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        // One statement, one column. `status` and `secret_digest` appear in the
        // `WHERE` and never in the `SET`: a use record must never be able to
        // put back a status somebody just changed or a digest somebody just
        // replaced.
        let updated = conn
            .execute(
                "UPDATE office_join_tokens
                 SET last_used_at = ?3
                 WHERE office_id = ?1 AND status = 'active' AND secret_digest = ?2",
                params![office_id, secret_digest, last_used_at],
            )
            .map_err(Self::qe)?;
        Ok(updated == 1)
    }

    fn list_office_join_tokens(&self) -> Result<Vec<PlatformOfficeJoinToken>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT office_id, secret_digest, status, note, created_at, last_used_at
                 FROM office_join_tokens
                 ORDER BY office_id ASC",
            )
            .map_err(Self::qe)?;
        let rows = stmt
            .query_map([], |row| {
                Ok(PlatformOfficeJoinToken {
                    office_id: row.get(0)?,
                    secret_digest: row.get(1)?,
                    status: row.get(2)?,
                    note: row.get(3)?,
                    created_at: row.get(4)?,
                    last_used_at: row.get(5)?,
                })
            })
            .map_err(Self::qe)?;
        let mut records = Vec::new();
        for row in rows {
            records.push(row.map_err(Self::qe)?);
        }
        Ok(records)
    }

    fn claim_office_vouch_nonce(
        &self,
        redemption: &PlatformOfficeVouchRedemption,
    ) -> Result<bool, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        // Pruned first, in the same lock: a row whose vouch has already
        // expired can never refuse anything the expiry check would have let
        // through, so keeping it would only grow the table.
        conn.execute(
            "DELETE FROM office_vouch_redemptions WHERE expires_at <= ?1",
            params![redemption.redeemed_at],
        )
        .map_err(Self::qe)?;
        let inserted = conn
            .execute(
                "INSERT OR IGNORE INTO office_vouch_redemptions
                 (nonce, office_id, identity, expires_at, redeemed_at)
                 VALUES (?1, ?2, ?3, ?4, ?5)",
                params![
                    &redemption.nonce,
                    &redemption.office_id,
                    &redemption.identity,
                    redemption.expires_at,
                    redemption.redeemed_at,
                ],
            )
            .map_err(Self::qe)?;
        Ok(inserted == 1)
    }

    fn put_office_identity_write(
        &self,
        entry: &PlatformOfficeIdentityWrite,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO office_identity_writes
             (write_id, office_id, owner, action, role, source, detail, written_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
            params![
                &entry.write_id,
                &entry.office_id,
                &entry.owner,
                &entry.action,
                &entry.role,
                &entry.source,
                &entry.detail,
                entry.written_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_office_identity_writes(
        &self,
        limit: usize,
    ) -> Result<Vec<PlatformOfficeIdentityWrite>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT write_id, office_id, owner, action, role, source, detail, written_at
                 FROM office_identity_writes
                 ORDER BY written_at DESC, rowid DESC
                 LIMIT ?1",
            )
            .map_err(Self::qe)?;
        let rows = stmt
            .query_map(params![limit as i64], |row| {
                Ok(PlatformOfficeIdentityWrite {
                    write_id: row.get(0)?,
                    office_id: row.get(1)?,
                    owner: row.get(2)?,
                    action: row.get(3)?,
                    role: row.get(4)?,
                    source: row.get(5)?,
                    detail: row.get(6)?,
                    written_at: row.get(7)?,
                })
            })
            .map_err(Self::qe)?;
        let mut entries = Vec::new();
        for row in rows {
            entries.push(row.map_err(Self::qe)?);
        }
        Ok(entries)
    }

    fn office_identity_write_exists(
        &self,
        owner: &str,
        action: &str,
    ) -> Result<bool, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let found: Option<i64> = conn
            .query_row(
                "SELECT 1 FROM office_identity_writes
                 WHERE owner = ?1 AND action = ?2
                 LIMIT 1",
                params![owner, action],
                |row| row.get(0),
            )
            .optional()
            .map_err(Self::qe)?;
        Ok(found.is_some())
    }

    fn put_office_local_authority_event(
        &self,
        entry: &PlatformOfficeLocalAuthorityEvent,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        // `OR REPLACE` on the id, not a plain insert: the controller stores a
        // copy of a row the office already holds, and the office re-sends
        // anything unacknowledged, so the same event id arriving twice is the
        // expected case rather than a conflict.
        conn.execute(
            "INSERT OR REPLACE INTO office_local_authority
             (event_id, office_id, event, join_fingerprint, owner, detail, acted_at, reported_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
            params![
                &entry.event_id,
                &entry.office_id,
                &entry.event,
                &entry.join_fingerprint,
                &entry.owner,
                &entry.detail,
                entry.acted_at,
                entry.reported_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_office_local_authority_events(
        &self,
        limit: usize,
    ) -> Result<Vec<PlatformOfficeLocalAuthorityEvent>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT event_id, office_id, event, join_fingerprint, owner, detail, acted_at,
                        reported_at
                 FROM office_local_authority
                 ORDER BY acted_at DESC, rowid DESC
                 LIMIT ?1",
            )
            .map_err(Self::qe)?;
        let rows = stmt
            .query_map(params![limit as i64], |row| {
                Ok(PlatformOfficeLocalAuthorityEvent {
                    event_id: row.get(0)?,
                    office_id: row.get(1)?,
                    event: row.get(2)?,
                    join_fingerprint: row.get(3)?,
                    owner: row.get(4)?,
                    detail: row.get(5)?,
                    acted_at: row.get(6)?,
                    reported_at: row.get(7)?,
                })
            })
            .map_err(Self::qe)?;
        let mut entries = Vec::new();
        for row in rows {
            entries.push(row.map_err(Self::qe)?);
        }
        Ok(entries)
    }

    fn find_office_break_glass(
        &self,
        join_fingerprint: &str,
    ) -> Result<Option<PlatformOfficeLocalAuthorityEvent>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT event_id, office_id, event, join_fingerprint, owner, detail, acted_at,
                    reported_at
             FROM office_local_authority
             WHERE join_fingerprint = ?1 AND event = ?2
             ORDER BY acted_at DESC, rowid DESC
             LIMIT 1",
            params![join_fingerprint, LOCAL_AUTHORITY_EVENT_BREAK_GLASS],
            |row| {
                Ok(PlatformOfficeLocalAuthorityEvent {
                    event_id: row.get(0)?,
                    office_id: row.get(1)?,
                    event: row.get(2)?,
                    join_fingerprint: row.get(3)?,
                    owner: row.get(4)?,
                    detail: row.get(5)?,
                    acted_at: row.get(6)?,
                    reported_at: row.get(7)?,
                })
            },
        );
        match result {
            Ok(entry) => Ok(Some(entry)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn list_unreported_office_break_glass(
        &self,
    ) -> Result<Vec<PlatformOfficeLocalAuthorityEvent>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT event_id, office_id, event, join_fingerprint, owner, detail, acted_at,
                        reported_at
                 FROM office_local_authority
                 WHERE event = ?1 AND reported_at = 0
                 ORDER BY acted_at ASC, rowid ASC",
            )
            .map_err(Self::qe)?;
        let rows = stmt
            .query_map(params![LOCAL_AUTHORITY_EVENT_BREAK_GLASS], |row| {
                Ok(PlatformOfficeLocalAuthorityEvent {
                    event_id: row.get(0)?,
                    office_id: row.get(1)?,
                    event: row.get(2)?,
                    join_fingerprint: row.get(3)?,
                    owner: row.get(4)?,
                    detail: row.get(5)?,
                    acted_at: row.get(6)?,
                    reported_at: row.get(7)?,
                })
            })
            .map_err(Self::qe)?;
        let mut entries = Vec::new();
        for row in rows {
            entries.push(row.map_err(Self::qe)?);
        }
        Ok(entries)
    }

    fn mark_office_local_authority_reported(
        &self,
        event_id: &str,
        reported_at: i64,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        // `reported_at` is the one mutable column, and it is delivery
        // bookkeeping rather than a change to what happened: the act, its time,
        // its office, and its fingerprint are written once and never rewritten.
        conn.execute(
            "UPDATE office_local_authority SET reported_at = ?2 WHERE event_id = ?1",
            params![event_id, reported_at],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_platform_offices(&self) -> Result<Vec<PlatformOffice>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT office_id, office_slug, label, office_kind, base_url, status, created_at, updated_at
                 FROM offices
                 ORDER BY office_slug ASC, office_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                Ok(PlatformOffice {
                    office_id: row.get(0)?,
                    office_slug: row.get(1)?,
                    label: row.get(2)?,
                    office_kind: row.get(3)?,
                    base_url: row.get(4)?,
                    status: row.get(5)?,
                    created_at: row.get(6)?,
                    updated_at: row.get(7)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect();
        Ok(items)
    }

    fn get_platform_service_instance(
        &self,
        service_instance_id: &str,
    ) -> Result<Option<PlatformServiceInstance>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT service_instance_id, service_kind, display_label, host_office_id, state_office_id, public_base_url, enabled, status, placement_generation, created_at, updated_at
             FROM platform_service_instances
             WHERE service_instance_id = ?1",
            params![service_instance_id],
            |row| {
                Ok(PlatformServiceInstance {
                    service_instance_id: row.get(0)?,
                    service_kind: row.get(1)?,
                    display_label: row.get(2)?,
                    host_office_id: row.get(3)?,
                    state_office_id: row.get(4)?,
                    public_base_url: row.get(5)?,
                    enabled: row.get::<_, i64>(6)? != 0,
                    status: row.get(7)?,
                    placement_generation: row.get(8)?,
                    created_at: row.get(9)?,
                    updated_at: row.get(10)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_platform_service_instance(
        &self,
        service: &PlatformServiceInstance,
    ) -> Result<(), PlatformError> {
        if service.host_office_id != service.state_office_id {
            return Err(PlatformError::new(
                "PLATFORM_SERVICE_STATE_HOST_MISMATCH",
                "platform service host_office_id and state_office_id must match in v1",
            ));
        }
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO platform_service_instances
             (service_instance_id, service_kind, display_label, host_office_id, state_office_id, public_base_url, enabled, status, placement_generation, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
             ON CONFLICT(service_instance_id) DO UPDATE SET
                 service_kind = excluded.service_kind,
                 display_label = excluded.display_label,
                 host_office_id = excluded.host_office_id,
                 state_office_id = excluded.state_office_id,
                 public_base_url = excluded.public_base_url,
                 enabled = excluded.enabled,
                 status = excluded.status,
                 placement_generation = excluded.placement_generation,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &service.service_instance_id,
                &service.service_kind,
                &service.display_label,
                &service.host_office_id,
                &service.state_office_id,
                &service.public_base_url,
                if service.enabled { 1_i64 } else { 0_i64 },
                &service.status,
                service.placement_generation,
                service.created_at,
                service.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_platform_service_instances(
        &self,
    ) -> Result<Vec<PlatformServiceInstance>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT service_instance_id, service_kind, display_label, host_office_id, state_office_id, public_base_url, enabled, status, placement_generation, created_at, updated_at
                 FROM platform_service_instances
                 ORDER BY service_kind ASC, service_instance_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                Ok(PlatformServiceInstance {
                    service_instance_id: row.get(0)?,
                    service_kind: row.get(1)?,
                    display_label: row.get(2)?,
                    host_office_id: row.get(3)?,
                    state_office_id: row.get(4)?,
                    public_base_url: row.get(5)?,
                    enabled: row.get::<_, i64>(6)? != 0,
                    status: row.get(7)?,
                    placement_generation: row.get(8)?,
                    created_at: row.get(9)?,
                    updated_at: row.get(10)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect();
        Ok(items)
    }

    fn get_platform_office_node(
        &self,
        node_id: &str,
    ) -> Result<Option<PlatformOfficeNode>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT office_id, node_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at
             FROM office_nodes
             WHERE node_id = ?1",
            params![node_id],
            |row| {
                let capabilities_json: String = row.get(5)?;
                Ok(PlatformOfficeNode {
                    office_id: row.get(0)?,
                    node_id: row.get(1)?,
                    label: row.get(2)?,
                    base_url: row.get(3)?,
                    status: row.get(4)?,
                    capabilities: serde_json::from_str(&capabilities_json).unwrap_or_default(),
                    registered_at: row.get(6)?,
                    last_heartbeat_at: row.get(7)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_platform_office_node(&self, node: &PlatformOfficeNode) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let capabilities_json =
            serde_json::to_string(&node.capabilities).unwrap_or_else(|_| "{}".to_string());
        conn.execute(
            "INSERT INTO office_nodes
             (office_id, node_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
             ON CONFLICT(node_id) DO UPDATE SET
                 office_id = excluded.office_id,
                 label = excluded.label,
                 base_url = excluded.base_url,
                 status = excluded.status,
                 capabilities_json = excluded.capabilities_json,
                 registered_at = excluded.registered_at,
                 last_heartbeat_at = excluded.last_heartbeat_at",
            params![
                &node.office_id,
                &node.node_id,
                &node.label,
                &node.base_url,
                &node.status,
                capabilities_json,
                node.registered_at,
                node.last_heartbeat_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_platform_office_nodes(&self) -> Result<Vec<PlatformOfficeNode>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT office_id, node_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at
                 FROM office_nodes
                 ORDER BY office_id ASC, node_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                let capabilities_json: String = row.get(5)?;
                Ok(PlatformOfficeNode {
                    office_id: row.get(0)?,
                    node_id: row.get(1)?,
                    label: row.get(2)?,
                    base_url: row.get(3)?,
                    status: row.get(4)?,
                    capabilities: serde_json::from_str(&capabilities_json).unwrap_or_default(),
                    registered_at: row.get(6)?,
                    last_heartbeat_at: row.get(7)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect();
        Ok(items)
    }

    fn get_worker_registry_record(
        &self,
        node_id: &str,
    ) -> Result<Option<WorkerRegistryRecord>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT office_id, office_slug, node_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at
             FROM worker_registry
             WHERE node_id = ?1",
            params![node_id],
            |row| {
                let capabilities_json: String = row.get(6)?;
                let capabilities = serde_json::from_str(&capabilities_json).unwrap_or_default();
                Ok(WorkerRegistryRecord {
                    office_id: row.get(0)?,
                    office_slug: row.get(1)?,
                    node_id: row.get(2)?,
                    label: row.get(3)?,
                    base_url: row.get(4)?,
                    status: row.get(5)?,
                    capabilities,
                    registered_at: row.get(7)?,
                    last_heartbeat_at: row.get(8)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_worker_registry_record(
        &self,
        record: &WorkerRegistryRecord,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let capabilities_json =
            serde_json::to_string(&record.capabilities).unwrap_or_else(|_| "{}".to_string());
        conn.execute(
            "INSERT OR REPLACE INTO worker_registry
             (office_id, office_slug, node_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                &record.office_id,
                &record.office_slug,
                &record.node_id,
                &record.label,
                &record.base_url,
                &record.status,
                capabilities_json,
                record.registered_at,
                record.last_heartbeat_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_worker_registry_records(&self) -> Result<Vec<WorkerRegistryRecord>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT office_id, office_slug, node_id, label, base_url, status, capabilities_json, registered_at, last_heartbeat_at
                 FROM worker_registry
                 ORDER BY node_id ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                let capabilities_json: String = row.get(6)?;
                let capabilities = serde_json::from_str(&capabilities_json).unwrap_or_default();
                Ok(WorkerRegistryRecord {
                    office_id: row.get(0)?,
                    office_slug: row.get(1)?,
                    node_id: row.get(2)?,
                    label: row.get(3)?,
                    base_url: row.get(4)?,
                    status: row.get(5)?,
                    capabilities,
                    registered_at: row.get(7)?,
                    last_heartbeat_at: row.get(8)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect();
        Ok(items)
    }

    fn delete_worker_registry_record(&self, node_id: &str) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM worker_registry WHERE node_id = ?1",
            params![node_id],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_project_runtime_placement(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<Option<ProjectRuntimePlacement>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT project_id, owner, project, mode, target, target_office_id, target_node_id, worker_id, resource_profile, desired_replicas, effective_state, created_at, updated_at
             FROM project_runtime_placements
             WHERE owner = ?1 AND project = ?2",
            params![owner, project],
            |row| {
                let mode: String = row.get(3)?;
                let target: String = row.get(4)?;
                Ok(ProjectRuntimePlacement {
                    project_id: row.get(0)?,
                    owner: row.get(1)?,
                    project: row.get(2)?,
                    mode: serde_json::from_value(Value::String(mode))
                        .unwrap_or(ProjectRuntimeMode::Shared),
                    target: serde_json::from_value(Value::String(target))
                        .unwrap_or(ProjectRuntimePlacementTarget::Local),
                    target_office_id: row.get(5)?,
                    target_node_id: row.get(6)?,
                    worker_id: row.get::<_, Option<String>>(7)?.or_else(|| row.get::<_, Option<String>>(6).ok().flatten()),
                    resource_profile: row.get(8)?,
                    desired_replicas: row.get::<_, i64>(9)?.max(0) as u32,
                    effective_state: row.get(10)?,
                    created_at: row.get(11)?,
                    updated_at: row.get(12)?,
                })
            },
        );
        match result {
            Ok(record) => Ok(Some(record)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_project_runtime_placement(
        &self,
        placement: &ProjectRuntimePlacement,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mode = serde_json::to_value(placement.mode)
            .ok()
            .and_then(|value| value.as_str().map(ToString::to_string))
            .unwrap_or_else(|| "shared".to_string());
        let target = serde_json::to_value(placement.target)
            .ok()
            .and_then(|value| value.as_str().map(ToString::to_string))
            .unwrap_or_else(|| "local".to_string());
        conn.execute(
            "INSERT INTO project_runtime_placements
             (project_id, owner, project, mode, target, target_office_id, target_node_id, worker_id, resource_profile, desired_replicas, effective_state, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
             ON CONFLICT(owner, project) DO UPDATE SET
                 project_id = excluded.project_id,
                 mode = excluded.mode,
                 target = excluded.target,
                 target_office_id = excluded.target_office_id,
                 target_node_id = excluded.target_node_id,
                 worker_id = excluded.worker_id,
                 resource_profile = excluded.resource_profile,
                 desired_replicas = excluded.desired_replicas,
                 effective_state = excluded.effective_state,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at",
            params![
                &placement.project_id,
                &placement.owner,
                &placement.project,
                mode,
                target,
                &placement.target_office_id,
                &placement.target_node_id,
                &placement.worker_id,
                &placement.resource_profile,
                i64::from(placement.desired_replicas),
                &placement.effective_state,
                placement.created_at,
                placement.updated_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_runtime_placements(
        &self,
    ) -> Result<Vec<ProjectRuntimePlacement>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT project_id, owner, project, mode, target, target_office_id, target_node_id, worker_id, resource_profile, desired_replicas, effective_state, created_at, updated_at
                 FROM project_runtime_placements
                 ORDER BY owner ASC, project ASC",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                let mode: String = row.get(3)?;
                let target: String = row.get(4)?;
                Ok(ProjectRuntimePlacement {
                    project_id: row.get(0)?,
                    owner: row.get(1)?,
                    project: row.get(2)?,
                    mode: serde_json::from_value(Value::String(mode))
                        .unwrap_or(ProjectRuntimeMode::Shared),
                    target: serde_json::from_value(Value::String(target))
                        .unwrap_or(ProjectRuntimePlacementTarget::Local),
                    target_office_id: row.get(5)?,
                    target_node_id: row.get(6)?,
                    worker_id: row
                        .get::<_, Option<String>>(7)?
                        .or_else(|| row.get::<_, Option<String>>(6).ok().flatten()),
                    resource_profile: row.get(8)?,
                    desired_replicas: row.get::<_, i64>(9)?.max(0) as u32,
                    effective_state: row.get(10)?,
                    created_at: row.get(11)?,
                    updated_at: row.get(12)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect();
        Ok(items)
    }

    fn delete_project_runtime_placement(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "DELETE FROM project_runtime_placements WHERE owner = ?1 AND project = ?2",
            params![owner, project],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn get_project_operation(
        &self,
        owner: &str,
        project: &str,
        operation_id: &str,
    ) -> Result<Option<ProjectOperationRecord>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let result = conn.query_row(
            "SELECT project_id, owner, project, operation_id, kind, status, current_step, source_office_id, target_office_id, artifact_rel_path, artifact_sha256, artifact_bytes, error_message, retry_count, created_at, updated_at, completed_at
             FROM project_operations
             WHERE owner = ?1 AND project = ?2 AND operation_id = ?3",
            params![owner, project, operation_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, Option<String>>(7)?,
                    row.get::<_, Option<String>>(8)?,
                    row.get::<_, Option<String>>(9)?,
                    row.get::<_, Option<String>>(10)?,
                    row.get::<_, Option<u64>>(11)?,
                    row.get::<_, Option<String>>(12)?,
                    row.get::<_, u32>(13)?,
                    row.get::<_, i64>(14)?,
                    row.get::<_, i64>(15)?,
                    row.get::<_, Option<i64>>(16)?,
                ))
            },
        );
        match result {
            Ok((
                project_id,
                owner,
                project,
                operation_id,
                kind,
                status,
                current_step,
                source_office_id,
                target_office_id,
                artifact_rel_path,
                artifact_sha256,
                artifact_bytes,
                error_message,
                retry_count,
                created_at,
                updated_at,
                completed_at,
            )) => Ok(Some(ProjectOperationRecord {
                project_id,
                owner,
                project,
                operation_id,
                kind: Self::decode_project_operation_kind(&kind)?,
                status: Self::decode_project_operation_status(&status)?,
                current_step,
                source_office_id,
                target_office_id,
                artifact_rel_path,
                artifact_sha256,
                artifact_bytes,
                error_message,
                retry_count,
                created_at,
                updated_at,
                completed_at,
            })),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(err) => Err(Self::qe(err)),
        }
    }

    fn put_project_operation(&self, record: &ProjectOperationRecord) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO project_operations
             (project_id, owner, project, operation_id, kind, status, current_step, source_office_id, target_office_id, artifact_rel_path, artifact_sha256, artifact_bytes, error_message, retry_count, created_at, updated_at, completed_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)
             ON CONFLICT(owner, project, operation_id) DO UPDATE SET
                 project_id = excluded.project_id,
                 kind = excluded.kind,
                 status = excluded.status,
                 current_step = excluded.current_step,
                 source_office_id = excluded.source_office_id,
                 target_office_id = excluded.target_office_id,
                 artifact_rel_path = excluded.artifact_rel_path,
                 artifact_sha256 = excluded.artifact_sha256,
                 artifact_bytes = excluded.artifact_bytes,
                 error_message = excluded.error_message,
                 retry_count = excluded.retry_count,
                 created_at = excluded.created_at,
                 updated_at = excluded.updated_at,
                 completed_at = excluded.completed_at",
            params![
                &record.project_id,
                &record.owner,
                &record.project,
                &record.operation_id,
                record.kind.key(),
                record.status.key(),
                &record.current_step,
                &record.source_office_id,
                &record.target_office_id,
                &record.artifact_rel_path,
                &record.artifact_sha256,
                record.artifact_bytes,
                &record.error_message,
                record.retry_count,
                record.created_at,
                record.updated_at,
                record.completed_at,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn list_project_operations(
        &self,
        owner: &str,
        project: &str,
        limit: usize,
    ) -> Result<Vec<ProjectOperationRecord>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT project_id, owner, project, operation_id, kind, status, current_step, source_office_id, target_office_id, artifact_rel_path, artifact_sha256, artifact_bytes, error_message, retry_count, created_at, updated_at, completed_at
                 FROM project_operations
                 WHERE owner = ?1 AND project = ?2
                 ORDER BY updated_at DESC, operation_id DESC
                 LIMIT ?3",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map(params![owner, project, limit as i64], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, Option<String>>(7)?,
                    row.get::<_, Option<String>>(8)?,
                    row.get::<_, Option<String>>(9)?,
                    row.get::<_, Option<String>>(10)?,
                    row.get::<_, Option<u64>>(11)?,
                    row.get::<_, Option<String>>(12)?,
                    row.get::<_, u32>(13)?,
                    row.get::<_, i64>(14)?,
                    row.get::<_, i64>(15)?,
                    row.get::<_, Option<i64>>(16)?,
                ))
            })
            .map_err(Self::qe)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(Self::qe)?;
        let mut out = Vec::with_capacity(items.len());
        for (
            project_id,
            owner,
            project,
            operation_id,
            kind,
            status,
            current_step,
            source_office_id,
            target_office_id,
            artifact_rel_path,
            artifact_sha256,
            artifact_bytes,
            error_message,
            retry_count,
            created_at,
            updated_at,
            completed_at,
        ) in items
        {
            out.push(ProjectOperationRecord {
                project_id,
                owner,
                project,
                operation_id,
                kind: Self::decode_project_operation_kind(&kind)?,
                status: Self::decode_project_operation_status(&status)?,
                current_step,
                source_office_id,
                target_office_id,
                artifact_rel_path,
                artifact_sha256,
                artifact_bytes,
                error_message,
                retry_count,
                created_at,
                updated_at,
                completed_at,
            });
        }
        Ok(out)
    }

    // ──────────────────────── MCP Sessions ────────────────────────

    fn list_all_mcp_sessions(&self) -> Result<Vec<McpSession>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let mut stmt = conn
            .prepare(
                "SELECT token, owner, project, capabilities_json, created_at, auto_reset_seconds, enabled
                 FROM mcp_sessions",
            )
            .map_err(Self::qe)?;
        let items = stmt
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, i64>(4)?,
                    row.get::<_, Option<i64>>(5)?,
                    row.get::<_, i64>(6)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .filter_map(
                |(token, owner, project, caps_json, created_at, auto_reset_seconds, enabled)| {
                    let capabilities: Vec<ProjectCapability> =
                        serde_json::from_str(&caps_json).unwrap_or_default();
                    Some(McpSession {
                        token,
                        owner,
                        project,
                        capabilities,
                        created_at,
                        auto_reset_seconds: auto_reset_seconds.map(|s| s as u64),
                        enabled: enabled != 0,
                    })
                },
            )
            .collect();
        Ok(items)
    }

    fn put_mcp_session(&self, session: &McpSession) -> Result<(), PlatformError> {
        let caps_json = serde_json::to_string(&session.capabilities)?;
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT OR REPLACE INTO mcp_sessions
             (token, owner, project, capabilities_json, created_at, auto_reset_seconds, enabled)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                &session.token,
                &session.owner,
                &session.project,
                &caps_json,
                session.created_at,
                session.auto_reset_seconds.map(|s| s as i64),
                session.enabled as i64,
            ],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn delete_mcp_session(&self, token: &str) -> Result<(), PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute("DELETE FROM mcp_sessions WHERE token = ?1", params![token])
            .map_err(Self::qe)?;
        Ok(())
    }

    // ──────────────────────── Admin ────────────────────────

    fn admin_list_collections(&self) -> Result<Vec<(String, usize)>, PlatformError> {
        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let tables = [
            "users",
            "projects",
            "project_credentials",
            "project_db_connections",
            "pipeline_meta",
            "project_policies",
            "project_policy_bindings",
            "mcp_sessions",
            "pipeline_invocations",
        ];
        let mut out = Vec::with_capacity(tables.len());
        for table in &tables {
            let count: i64 = conn
                .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| {
                    row.get(0)
                })
                .unwrap_or(0);
            out.push((table.to_string(), count as usize));
        }
        Ok(out)
    }

    // ──────────────────────── Invocations ────────────────────────

    fn log_pipeline_invocation(
        &self,
        owner: &str,
        project: &str,
        file_rel_path: &str,
        entry: &PipelineInvocationEntry,
        max_n: usize,
        max_age_secs: Option<i64>,
    ) -> Result<(), PlatformError> {
        let trace_json = serde_json::to_string(&entry.trace)?;
        if let Some(conn) = self.open_project_invocations_conn(owner, project)? {
            conn.execute(
                "INSERT INTO pipeline_invocations
                 (file_rel_path, run_id, at, duration_ms, status, trigger, error, trace_json)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                params![
                    file_rel_path,
                    &entry.run_id,
                    entry.at,
                    entry.duration_ms as i64,
                    &entry.status,
                    &entry.trigger,
                    entry.error.as_deref(),
                    &trace_json,
                ],
            )
            .map_err(Self::qe)?;
            if let Some(max_age_secs) = max_age_secs {
                let cutoff = entry.at.saturating_sub(max_age_secs.max(0));
                conn.execute(
                    "DELETE FROM pipeline_invocations
                     WHERE file_rel_path = ?1 AND at < ?2",
                    params![file_rel_path, cutoff],
                )
                .map_err(Self::qe)?;
            }
            conn.execute(
                "DELETE FROM pipeline_invocations
                 WHERE file_rel_path = ?1
                   AND id NOT IN (
                     SELECT id FROM pipeline_invocations
                     WHERE file_rel_path = ?1
                     ORDER BY at DESC LIMIT ?2
                   )",
                params![file_rel_path, max_n as i64],
            )
            .map_err(Self::qe)?;
            return Ok(());
        }

        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.execute(
            "INSERT INTO pipeline_invocations
             (owner, project, file_rel_path, run_id, at, duration_ms, status, trigger, error, trace_json)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
            params![
                owner,
                project,
                file_rel_path,
                &entry.run_id,
                entry.at,
                entry.duration_ms as i64,
                &entry.status,
                &entry.trigger,
                entry.error.as_deref(),
                &trace_json,
            ],
        )
        .map_err(Self::qe)?;
        if let Some(max_age_secs) = max_age_secs {
            let cutoff = entry.at.saturating_sub(max_age_secs.max(0));
            conn.execute(
                "DELETE FROM pipeline_invocations
                 WHERE owner = ?1 AND project = ?2 AND file_rel_path = ?3 AND at < ?4",
                params![owner, project, file_rel_path, cutoff],
            )
            .map_err(Self::qe)?;
        }
        // Trim rows beyond max_n (keep the most-recent by `at`)
        conn.execute(
            "DELETE FROM pipeline_invocations
             WHERE owner = ?1 AND project = ?2 AND file_rel_path = ?3
               AND id NOT IN (
                 SELECT id FROM pipeline_invocations
                 WHERE owner = ?1 AND project = ?2 AND file_rel_path = ?3
                 ORDER BY at DESC LIMIT ?4
               )",
            params![owner, project, file_rel_path, max_n as i64],
        )
        .map_err(Self::qe)?;
        Ok(())
    }

    fn record_pipeline_error(
        &self,
        owner: &str,
        project: &str,
        file_rel_path: &str,
        entry: &PipelineInvocationEntry,
        bounds: ErrorGroupBounds,
    ) -> Result<(), PlatformError> {
        if entry.status != "error" {
            return Ok(());
        }
        let Some(conn) = self.open_project_invocations_conn(owner, project)? else {
            return Ok(());
        };
        let (node_id, code, message) = crate::platform::model::error_group_parts(entry);
        let pattern = crate::platform::model::error_message_pattern(&message);
        let signature = format!("{file_rel_path} · {node_id} · {code} · {pattern}");
        let entry_json = serde_json::to_string(entry)?;
        let existing = conn
            .query_row(
                "SELECT count, first_seen, last_seen, occurrences_json, capture, captured_json FROM pipeline_error_groups WHERE file_rel_path = ?1 AND signature = ?2",
                params![file_rel_path, signature],
                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?, row.get::<_, i64>(2)?, row.get::<_, String>(3)?, row.get::<_, String>(4)?, row.get::<_, String>(5)?)),
            )
            .optional()
            .map_err(|e| PlatformError::new("PLATFORM_ERROR_GROUP_READ", e.to_string()))?;
        match existing {
            Some((count, _first_seen, last_seen, occurrences_json, capture, captured_json)) => {
                let mut occurrences: Vec<(String, i64)> = serde_json::from_str(&occurrences_json).unwrap_or_default();
                occurrences.push((entry.run_id.clone(), entry.at));
                if occurrences.len() > bounds.occurrence_ring {
                    let drop = occurrences.len() - bounds.occurrence_ring;
                    occurrences.drain(..drop);
                }
                let mut captured: Vec<PipelineInvocationEntry> = if capture == "all" { serde_json::from_str(&captured_json).unwrap_or_default() } else { Vec::new() };
                if capture == "all" && captured.len() < bounds.capture_cap {
                    captured.push(entry.clone());
                }
                // Quiet for longer than the threshold, then back: a reintroduced bug, not the old count continuing unremarked.
                let reopened = if entry.at - last_seen > bounds.reopen_after_secs { Some(entry.at) } else { None };
                conn.execute(
                    "UPDATE pipeline_error_groups SET count = ?3, last_seen = ?4, latest_json = ?5, occurrences_json = ?6, captured_json = ?7, reopened_at = COALESCE(?8, reopened_at) WHERE file_rel_path = ?1 AND signature = ?2",
                    params![file_rel_path, signature, count + 1, entry.at, entry_json, serde_json::to_string(&occurrences)?, serde_json::to_string(&captured)?, reopened],
                )
                .map_err(|e| PlatformError::new("PLATFORM_ERROR_GROUP_WRITE", e.to_string()))?;
            }
            None => {
                conn.execute(
                    "INSERT INTO pipeline_error_groups (file_rel_path, signature, node_id, code, message_pattern, count, first_seen, last_seen, first_json, latest_json, occurrences_json) VALUES (?1, ?2, ?3, ?4, ?5, 1, ?6, ?6, ?7, ?7, ?8)",
                    params![file_rel_path, signature, node_id, code, pattern, entry.at, entry_json, serde_json::to_string(&vec![(entry.run_id.clone(), entry.at)])?],
                )
                .map_err(|e| PlatformError::new("PLATFORM_ERROR_GROUP_WRITE", e.to_string()))?;
                // Bounded by distinct errors: the groups least recently seen go first.
                conn.execute(
                    "DELETE FROM pipeline_error_groups WHERE rowid IN (SELECT rowid FROM pipeline_error_groups ORDER BY last_seen DESC LIMIT -1 OFFSET ?1)",
                    params![bounds.max_groups as i64],
                )
                .map_err(|e| PlatformError::new("PLATFORM_ERROR_GROUP_WRITE", e.to_string()))?;
            }
        }
        Ok(())
    }

    fn list_pipeline_error_groups(
        &self,
        owner: &str,
        project: &str,
        file_rel_path: Option<&str>,
    ) -> Result<Vec<PipelineErrorGroup>, PlatformError> {
        let Some(conn) = self.open_project_invocations_conn(owner, project)? else {
            return Ok(vec![]);
        };
        let sql = match file_rel_path {
            Some(_) => "SELECT * FROM pipeline_error_groups WHERE file_rel_path = ?1 ORDER BY last_seen DESC",
            None => "SELECT * FROM pipeline_error_groups WHERE ?1 = ?1 ORDER BY last_seen DESC",
        };
        let mut stmt = conn.prepare(sql).map_err(|e| PlatformError::new("PLATFORM_ERROR_GROUP_READ", e.to_string()))?;
        let rows = stmt
            .query_map(params![file_rel_path.unwrap_or("")], Self::error_group_from_row)
            .map_err(|e| PlatformError::new("PLATFORM_ERROR_GROUP_READ", e.to_string()))?;
        Ok(rows.filter_map(Result::ok).collect())
    }

    fn find_pipeline_error_group_by_run(
        &self,
        owner: &str,
        project: &str,
        run_id_prefix: &str,
    ) -> Result<Option<PipelineErrorGroup>, PlatformError> {
        let prefix = run_id_prefix.trim().to_ascii_lowercase();
        if prefix.len() < 8 {
            return Ok(None);
        }
        Ok(self
            .list_pipeline_error_groups(owner, project, None)?
            .into_iter()
            .find(|g| g.occurrences.iter().any(|(id, _)| id.starts_with(&prefix)) || g.first.run_id.starts_with(&prefix) || g.latest.run_id.starts_with(&prefix)))
    }

    fn get_pipeline_invocations(
        &self,
        owner: &str,
        project: &str,
        file_rel_path: &str,
        max_age_secs: Option<i64>,
    ) -> Result<Vec<PipelineInvocationEntry>, PlatformError> {
        let cutoff = max_age_secs.map(|age| now_ts().saturating_sub(age.max(0)));
        if let Some(conn) = self.open_project_invocations_conn(owner, project)? {
            let items = if let Some(cutoff) = cutoff {
                let mut stmt = conn
                    .prepare(
                        "SELECT run_id, at, duration_ms, status, trigger, error, trace_json
                         FROM pipeline_invocations
                         WHERE file_rel_path = ?1 AND at >= ?2
                         ORDER BY at DESC",
                    )
                    .map_err(Self::qe)?;
                stmt.query_map(params![file_rel_path, cutoff], |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, i64>(1)?,
                        row.get::<_, i64>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, Option<String>>(5)?,
                        row.get::<_, String>(6)?,
                    ))
                })
                .map_err(Self::qe)?
                .filter_map(|r| r.ok())
                .map(
                    |(run_id, at, duration_ms, status, trigger, error, trace_json)| {
                        let trace = serde_json::from_str(&trace_json).unwrap_or_default();
                        PipelineInvocationEntry {
                            run_id,
                            at,
                            duration_ms: duration_ms as u64,
                            status,
                            trigger,
                            error,
                            trace,
                        }
                    },
                )
                .collect::<Vec<_>>()
            } else {
                let mut stmt = conn
                    .prepare(
                        "SELECT run_id, at, duration_ms, status, trigger, error, trace_json
                         FROM pipeline_invocations
                         WHERE file_rel_path = ?1
                         ORDER BY at DESC",
                    )
                    .map_err(Self::qe)?;
                stmt.query_map(params![file_rel_path], |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, i64>(1)?,
                        row.get::<_, i64>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, Option<String>>(5)?,
                        row.get::<_, String>(6)?,
                    ))
                })
                .map_err(Self::qe)?
                .filter_map(|r| r.ok())
                .map(
                    |(run_id, at, duration_ms, status, trigger, error, trace_json)| {
                        let trace = serde_json::from_str(&trace_json).unwrap_or_default();
                        PipelineInvocationEntry {
                            run_id,
                            at,
                            duration_ms: duration_ms as u64,
                            status,
                            trigger,
                            error,
                            trace,
                        }
                    },
                )
                .collect::<Vec<_>>()
            };
            return Ok(items);
        }

        let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let items = if let Some(cutoff) = cutoff {
            let mut stmt = conn
                .prepare(
                    "SELECT run_id, at, duration_ms, status, trigger, error, trace_json
                     FROM pipeline_invocations
                     WHERE owner = ?1 AND project = ?2 AND file_rel_path = ?3 AND at >= ?4
                     ORDER BY at DESC",
                )
                .map_err(Self::qe)?;
            stmt.query_map(params![owner, project, file_rel_path, cutoff], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, Option<String>>(5)?,
                    row.get::<_, String>(6)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .map(
                |(run_id, at, duration_ms, status, trigger, error, trace_json)| {
                    let trace = serde_json::from_str(&trace_json).unwrap_or_default();
                    PipelineInvocationEntry {
                        run_id,
                        at,
                        duration_ms: duration_ms as u64,
                        status,
                        trigger,
                        error,
                        trace,
                    }
                },
            )
            .collect::<Vec<_>>()
        } else {
            let mut stmt = conn
                .prepare(
                    "SELECT run_id, at, duration_ms, status, trigger, error, trace_json
                     FROM pipeline_invocations
                     WHERE owner = ?1 AND project = ?2 AND file_rel_path = ?3
                     ORDER BY at DESC",
                )
                .map_err(Self::qe)?;
            stmt.query_map(params![owner, project, file_rel_path], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, Option<String>>(5)?,
                    row.get::<_, String>(6)?,
                ))
            })
            .map_err(Self::qe)?
            .filter_map(|r| r.ok())
            .map(
                |(run_id, at, duration_ms, status, trigger, error, trace_json)| {
                    let trace = serde_json::from_str(&trace_json).unwrap_or_default();
                    PipelineInvocationEntry {
                        run_id,
                        at,
                        duration_ms: duration_ms as u64,
                        status,
                        trigger,
                        error,
                        trace,
                    }
                },
            )
            .collect::<Vec<_>>()
        };
        Ok(items)
    }

    fn get_pipeline_invocation_log_stats(
        &self,
        owner: &str,
        project: &str,
    ) -> Result<PipelineInvocationLogStats, PlatformError> {
        let Some(path) = self.project_invocations_db_path(owner, project)? else {
            return Ok(PipelineInvocationLogStats::default());
        };
        if !path.exists() {
            return Ok(PipelineInvocationLogStats::default());
        }
        let conn = self
            .open_project_invocations_conn(owner, project)?
            .ok_or_else(|| {
                PlatformError::new(
                    "PLATFORM_INVOCATION_LOG_OPEN",
                    "project invocation log adapter unavailable",
                )
            })?;
        let (count, trace_bytes): (i64, i64) = conn
            .query_row(
                "SELECT COUNT(*), COALESCE(SUM(LENGTH(trace_json)), 0) FROM pipeline_invocations",
                [],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .map_err(Self::qe)?;
        let mut stmt = conn
            .prepare(
                "SELECT file_rel_path,
                        COUNT(*) AS count,
                        COALESCE(SUM(LENGTH(trace_json)), 0) AS trace_bytes,
                        COALESCE(MAX(LENGTH(trace_json)), 0) AS largest_trace_bytes,
                        COALESCE(MAX(at), 0) AS latest_at
                 FROM pipeline_invocations
                 GROUP BY file_rel_path
                 ORDER BY trace_bytes DESC, count DESC, file_rel_path ASC",
            )
            .map_err(Self::qe)?;
        let pipelines = stmt
            .query_map([], |row| {
                Ok(PipelineInvocationLogPipelineStats {
                    file_rel_path: row.get(0)?,
                    count: row.get::<_, i64>(1)? as u64,
                    trace_bytes: row.get::<_, i64>(2)? as u64,
                    largest_trace_bytes: row.get::<_, i64>(3)? as u64,
                    latest_at: row.get(4)?,
                })
            })
            .map_err(Self::qe)?
            .filter_map(|row| row.ok())
            .collect::<Vec<_>>();
        let file_size =
            |path: &Path| -> u64 { std::fs::metadata(path).map(|meta| meta.len()).unwrap_or(0) };
        Ok(PipelineInvocationLogStats {
            count: count.max(0) as u64,
            trace_bytes: trace_bytes.max(0) as u64,
            db_bytes: file_size(&path),
            wal_bytes: file_size(Path::new(&format!("{}-wal", path.display()))),
            shm_bytes: file_size(Path::new(&format!("{}-shm", path.display()))),
            pipelines,
        })
    }

    fn clear_pipeline_invocation_logs(
        &self,
        owner: &str,
        project: &str,
        file_rel_path: Option<&str>,
    ) -> Result<u64, PlatformError> {
        let Some(path) = self.project_invocations_db_path(owner, project)? else {
            return Ok(0);
        };
        if !path.exists() {
            return Ok(0);
        }
        let conn = self
            .open_project_invocations_conn(owner, project)?
            .ok_or_else(|| {
                PlatformError::new(
                    "PLATFORM_INVOCATION_LOG_OPEN",
                    "project invocation log adapter unavailable",
                )
            })?;
        let affected = if let Some(file_rel_path) = file_rel_path {
            conn.execute(
                "DELETE FROM pipeline_invocations WHERE file_rel_path = ?1",
                params![file_rel_path],
            )
        } else {
            conn.execute("DELETE FROM pipeline_invocations", [])
        }
        .map_err(Self::qe)? as u64;
        conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE); VACUUM;")
            .map_err(Self::qe)?;
        Ok(affected)
    }

    fn transfer_project_owner(
        &self,
        old_owner: &str,
        project: &str,
        new_owner: &str,
        new_owner_user_id: &str,
    ) -> Result<(), PlatformError> {
        let mut conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
        let tx = conn
            .transaction_with_behavior(TransactionBehavior::Exclusive)
            .map_err(Self::qe)?;

        // Core project record — also update owner_user_id.
        tx.execute(
            "UPDATE projects SET owner = ?1, owner_user_id = ?2 WHERE owner = ?3 AND project = ?4",
            params![new_owner, new_owner_user_id, old_owner, project],
        )
        .map_err(Self::qe)?;

        // All tables with (owner, project) in their primary key.
        let rekey_tables = [
            "project_credentials",
            "project_db_connections",
            "project_hub_repositories",
            "hub_publishers",
            "pipeline_meta",
            "project_policies",
            "project_policy_bindings",
            "project_members",
            "project_invites",
            "project_runtime_placements",
            "project_operations",
            "pipeline_invocations",
        ];
        for table in &rekey_tables {
            tx.execute(
                &format!("UPDATE {table} SET owner = ?1 WHERE owner = ?2 AND project = ?3"),
                params![new_owner, old_owner, project],
            )
            .map_err(Self::qe)?;
        }

        // Hub asset versions: multiple owner columns that may reference the old owner.
        for col in &["authority_owner", "publisher_owner", "source_owner"] {
            tx.execute(
                &format!(
                    "UPDATE hub_asset_versions SET {col} = ?1 WHERE {col} = ?2 AND source_project = ?3"
                ),
                params![new_owner, old_owner, project],
            )
            .map_err(Self::qe)?;
        }

        // Invalidate MCP sessions — force re-auth under new owner.
        tx.execute(
            "DELETE FROM mcp_sessions WHERE owner = ?1 AND project = ?2",
            params![old_owner, project],
        )
        .map_err(Self::qe)?;

        tx.commit().map_err(Self::qe)?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use serde_json::json;

    use super::*;
    use crate::pipeline::model::NodeTraceEntry;
    use crate::platform::adapters::data::DataAdapter;

    fn seed_project(adapter: &SqliteDataAdapter, owner: &str, project: &str) {
        adapter
            .put_user(&StoredUser {
                profile: PlatformUser {
                    user_id: String::new(),
                    owner: owner.to_string(),
                    role: "superadmin".to_string(),
                    git_name: String::new(),
                    git_email: String::new(),
                    created_at: 1,
                    updated_at: 1,
                },
                auth: PlatformUserLocalAuth {
                    user_id: String::new(),
                    password_hash: "x".to_string(),
                    password_alg: "test".to_string(),
                    password_updated_at: 1,
                    credential_state: CREDENTIAL_STATE_CHOSEN.to_string(),
                },
            })
            .expect("put user");
        adapter
            .put_project(&PlatformProject {
                project_id: String::new(),
                owner: owner.to_string(),
                project: project.to_string(),
                title: project.to_string(),
                owner_user_id: String::new(),
                created_at: 1,
                updated_at: 1,
            })
            .expect("put project");
    }

    fn credential(owner: &str, project: &str, id: &str, secret: Value) -> ProjectCredential {
        ProjectCredential {
            owner: owner.to_string(),
            project: project.to_string(),
            credential_id: id.to_string(),
            title: id.to_string(),
            kind: "postgres".to_string(),
            secret,
            notes: String::new(),
            created_at: 1,
            updated_at: 1,
        }
    }

    fn raw_secret_json(adapter: &SqliteDataAdapter, credential_id: &str) -> String {
        let conn = adapter.conn.lock().unwrap_or_else(|e| e.into_inner());
        conn.query_row(
            "SELECT secret_json FROM project_credentials WHERE credential_id = ?1",
            params![credential_id],
            |row| row.get::<_, String>(0),
        )
        .expect("stored row")
    }

    /// The rule the whole feature exists for, stated against the column.
    #[test]
    fn a_stored_credential_is_ciphertext_in_the_column_and_plaintext_through_the_adapter() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        seed_project(&adapter, "superadmin", "default");
        adapter
            .put_project_credential(&credential(
                "superadmin",
                "default",
                "db",
                json!({"host": "db.example", "password": "correct-horse-battery-staple"}),
            ))
            .expect("put");

        let stored = raw_secret_json(&adapter, "db");
        assert!(
            !stored.contains("correct-horse-battery-staple"),
            "the secret is readable in the column: {stored}"
        );
        assert!(
            !stored.contains("db.example"),
            "the whole secret half is encrypted regardless of what a type declared: {stored}"
        );
        assert!(stored.starts_with("\"zfc1:1:"), "{stored}");

        let read = adapter
            .get_project_credential("superadmin", "default", "db")
            .expect("get")
            .expect("present");
        assert_eq!(
            read.secret,
            json!({"host": "db.example", "password": "correct-horse-battery-staple"})
        );
        let listed = adapter
            .list_project_credentials("superadmin", "default")
            .expect("list");
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].secret, read.secret);
    }

    /// A row written before this release is plaintext. It is accepted on read
    /// and re-emitted as ciphertext on the next write — `ZebFsAcl`'s rule,
    /// because a credential cannot be regenerated either.
    #[test]
    fn a_pre_encryption_row_is_read_as_it_is_and_rewritten_as_ciphertext() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        seed_project(&adapter, "superadmin", "default");
        {
            let conn = adapter.conn.lock().unwrap_or_else(|e| e.into_inner());
            conn.execute(
                "INSERT INTO project_credentials
                 (owner, project, credential_id, title, kind, secret_json, notes, created_at, updated_at)
                 VALUES ('superadmin', 'default', 'legacy', 'Legacy', 'postgres', ?1, '', 1, 1)",
                params![r#"{"password":"from-before-encryption"}"#],
            )
            .expect("legacy insert");
        }

        let read = adapter
            .get_project_credential("superadmin", "default", "legacy")
            .expect("get")
            .expect("present");
        assert_eq!(read.secret, json!({"password": "from-before-encryption"}));

        adapter.put_project_credential(&read).expect("rewrite");
        let stored = raw_secret_json(&adapter, "legacy");
        assert!(stored.starts_with("\"zfc1:"), "{stored}");
        assert!(!stored.contains("from-before-encryption"), "{stored}");

        // And the sweep converts one that is never written again.
        {
            let conn = adapter.conn.lock().unwrap_or_else(|e| e.into_inner());
            conn.execute(
                "INSERT INTO project_credentials
                 (owner, project, credential_id, title, kind, secret_json, notes, created_at, updated_at)
                 VALUES ('superadmin', 'default', 'untouched', 'L', 'postgres', ?1, '', 1, 1)",
                params![r#"{"token":"never-rewritten"}"#],
            )
            .expect("legacy insert");
        }
        let sweep = adapter.reencrypt_project_credentials().expect("sweep");
        assert_eq!(sweep.examined, 2);
        assert_eq!(sweep.rewritten, 1);
        assert_eq!(sweep.migrated_from_plaintext, 1);
        assert!(sweep.compacted, "a migrating sweep rebuilds the catalog");
        assert!(!raw_secret_json(&adapter, "untouched").contains("never-rewritten"));
        // The plaintext is gone from the file, not merely from the row: a
        // migrated credential whose old bytes sit in a freed page is a
        // migration that did not happen for the reader this defends against.
        let catalog = std::fs::read(tmp.path().join("platform/catalog.db")).expect("catalog");
        assert!(
            !catalog
                .windows("never-rewritten".len())
                .any(|w| w == b"never-rewritten"),
            "the migrated plaintext is still in the database file"
        );
    }

    /// Rotation is online, instant, and re-encrypts nothing.
    #[test]
    fn rotation_seals_new_writes_under_the_new_key_and_leaves_old_ones_readable() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        seed_project(&adapter, "superadmin", "default");
        adapter
            .put_project_credential(&credential(
                "superadmin",
                "default",
                "before",
                json!({"key": "sealed-under-one"}),
            ))
            .expect("put");
        assert!(raw_secret_json(&adapter, "before").starts_with("\"zfc1:1:"));

        let report = adapter.rotate_credential_key().expect("rotate");
        assert_eq!(report.current_key_id, 2);
        assert_eq!(report.generations.len(), 2);
        assert!(
            report
                .generations
                .iter()
                .any(|g| g.key_id == 1 && !g.current)
        );
        assert!(
            report
                .generations
                .iter()
                .any(|g| g.key_id == 2 && g.current)
        );

        adapter
            .put_project_credential(&credential(
                "superadmin",
                "default",
                "after",
                json!({"key": "sealed-under-two"}),
            ))
            .expect("put");
        assert!(raw_secret_json(&adapter, "after").starts_with("\"zfc1:2:"));
        // The older generation stayed for reads. This is the assertion that
        // fails on every design that rotates by re-encrypting in place.
        assert!(raw_secret_json(&adapter, "before").starts_with("\"zfc1:1:"));
        assert_eq!(
            adapter
                .get_project_credential("superadmin", "default", "before")
                .expect("get")
                .expect("present")
                .secret,
            json!({"key": "sealed-under-one"})
        );

        // The sweep is what makes generation 1 stop being needed.
        let sweep = adapter.reencrypt_project_credentials().expect("sweep");
        assert_eq!(sweep.key_id, 2);
        assert_eq!(sweep.rewritten, 1);
        assert_eq!(sweep.migrated_from_plaintext, 0);
        assert!(raw_secret_json(&adapter, "before").starts_with("\"zfc1:2:"));
    }

    /// Rekey replaces the file and touches no ciphertext.
    #[test]
    fn a_rekey_replaces_the_instance_key_and_changes_no_credential() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        seed_project(&adapter, "superadmin", "default");
        adapter
            .put_project_credential(&credential(
                "superadmin",
                "default",
                "db",
                json!({"password": "unchanged-by-a-rekey"}),
            ))
            .expect("put");
        let ciphertext = raw_secret_json(&adapter, "db");
        let key_path = tmp.path().join(CREDENTIAL_KEY_REL);
        let before = std::fs::read_to_string(&key_path).expect("key file");

        adapter.rekey_credential_keyring().expect("rekey");
        let after = std::fs::read_to_string(&key_path).expect("key file");
        assert_ne!(before, after, "the instance key must actually change");
        assert_eq!(after.lines().filter(|l| !l.is_empty()).count(), 1);
        assert_eq!(
            raw_secret_json(&adapter, "db"),
            ciphertext,
            "a rekey re-wraps data keys and never rewrites a credential"
        );

        // A restart under the new file reads everything.
        drop(adapter);
        let reopened = SqliteDataAdapter::new(tmp.path()).expect("reopen");
        assert_eq!(
            reopened
                .get_project_credential("superadmin", "default", "db")
                .expect("get")
                .expect("present")
                .secret,
            json!({"password": "unchanged-by-a-rekey"})
        );
    }

    /// The refusal, at the boundary an operator meets it.
    #[test]
    fn an_instance_whose_key_is_gone_refuses_to_open_its_catalog() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        seed_project(&adapter, "superadmin", "default");
        adapter
            .put_project_credential(&credential(
                "superadmin",
                "default",
                "db",
                json!({"password": "irreplaceable"}),
            ))
            .expect("put");
        drop(adapter);

        let key_path = tmp.path().join(CREDENTIAL_KEY_REL);
        std::fs::remove_file(&key_path).expect("remove key");
        let err = SqliteDataAdapter::new(tmp.path())
            .err()
            .expect("must refuse");
        assert_eq!(err.code, "PLATFORM_CREDENTIAL_KEY_MISSING");
        assert!(err.message.contains("cannot be regenerated"), "{err:?}");
        assert!(!key_path.exists(), "a refusal must not write a new key");

        // A different key is a different refusal, and just as loud.
        std::fs::write(
            &key_path,
            crate::infra::secrets::SecretKey::generate()
                .expect("generate")
                .render(),
        )
        .expect("write");
        let err = SqliteDataAdapter::new(tmp.path())
            .err()
            .expect("must refuse");
        assert_eq!(err.code, "PLATFORM_CREDENTIAL_KEY_UNUSABLE");
    }

    /// A catalog opened without a data root holds no keyring and says so
    /// rather than writing a credential in the clear.
    #[test]
    fn a_rootless_catalog_refuses_credentials_instead_of_storing_them_plainly() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter =
            SqliteDataAdapter::new_at_db_path(&tmp.path().join("hub.db")).expect("adapter");
        let err = adapter
            .put_project_credential(&credential("superadmin", "default", "db", json!({"a": 1})))
            .expect_err("must refuse");
        assert_eq!(err.code, "PLATFORM_CREDENTIAL_KEY_MISSING");
        assert_eq!(
            adapter
                .rotate_credential_key()
                .expect_err("no keyring")
                .code,
            "PLATFORM_CREDENTIAL_KEY_MISSING"
        );
    }

    /// A project with the rows a real project has must actually delete.
    ///
    /// Six tables reference `projects(project_id)` with `ON DELETE RESTRICT`,
    /// so deleting the project row alone failed with a foreign key error for
    /// every project that had ever had a policy or a member — which creation
    /// gives it. This asserts both halves: the delete succeeds, and it takes
    /// the project's own rows with it rather than leaving them for a project
    /// later recreated under the same slug to inherit.
    #[test]
    fn deleting_a_project_takes_its_dependent_rows_with_it() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        adapter
            .put_user(&StoredUser {
                profile: PlatformUser {
                    user_id: String::new(),
                    owner: "superadmin".to_string(),
                    role: "superadmin".to_string(),
                    git_name: String::new(),
                    git_email: String::new(),
                    created_at: 1,
                    updated_at: 1,
                },
                auth: PlatformUserLocalAuth {
                    user_id: String::new(),
                    password_hash: "x".to_string(),
                    password_alg: "test".to_string(),
                    password_updated_at: 1,
                    credential_state: CREDENTIAL_STATE_CHOSEN.to_string(),
                },
            })
            .expect("put user");
        adapter
            .put_project(&PlatformProject {
                project_id: String::new(),
                owner_user_id: String::new(),
                owner: "superadmin".to_string(),
                project: "doomed".to_string(),
                title: "Doomed".to_string(),
                created_at: 1,
                updated_at: 1,
            })
            .expect("put project");
        adapter
            .put_project_policy(&ProjectPolicy {
                project_id: String::new(),
                owner: "superadmin".to_string(),
                project: "doomed".to_string(),
                policy_id: "owner".to_string(),
                title: "Owner".to_string(),
                capabilities: vec![],
                managed: true,
                created_at: 1,
                updated_at: 1,
            })
            .expect("put policy");
        adapter
            .put_project_credential(&ProjectCredential {
                owner: "superadmin".to_string(),
                project: "doomed".to_string(),
                credential_id: "db".to_string(),
                title: "DB".to_string(),
                kind: "postgres".to_string(),
                secret: json!({"password": "hunter2"}),
                notes: String::new(),
                created_at: 1,
                updated_at: 1,
            })
            .expect("put credential");

        adapter
            .delete_project("superadmin", "doomed")
            .expect("delete project");

        assert!(
            adapter
                .list_projects("superadmin")
                .expect("list")
                .is_empty()
        );
        assert!(
            adapter
                .list_project_policies("superadmin", "doomed")
                .expect("policies")
                .is_empty(),
            "policies must not outlive their project"
        );
        assert!(
            adapter
                .get_project_credential("superadmin", "doomed", "db")
                .expect("credential")
                .is_none(),
            "a recreated project must not inherit the old one's secrets"
        );
    }

    #[test]
    fn invocation_logs_are_project_local_under_data_logs() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let adapter = SqliteDataAdapter::new(tmp.path()).expect("adapter");
        let entry = PipelineInvocationEntry {
            run_id: "run-1".to_string(),
            at: 1_700_000_000,
            duration_ms: 42,
            status: "ok".to_string(),
            trigger: "manual".to_string(),
            error: None,
            trace: vec![NodeTraceEntry {
                node_id: "a".to_string(),
                node_kind: "n.test".to_string(),
                config: None,
                duration_ms: 42,
                input: json!({"in": true}),
                output: json!({"out": true}),
                error: None,
                status: "ok".to_string(),
                error_code: None,
                preview_snapshot: None,
            }],
        };

        adapter
            .log_pipeline_invocation(
                "superadmin",
                "default",
                "pipelines/demo.zf.json",
                &entry,
                20,
                None,
            )
            .expect("log invocation");

        let project_log = tmp
            .path()
            .join("users")
            .join("superadmin")
            .join("default")
            .join("data")
            .join("logs")
            .join("invocations.db");
        assert!(project_log.exists(), "project invocation DB should exist");

        let stored = adapter
            .get_pipeline_invocations("superadmin", "default", "pipelines/demo.zf.json", None)
            .expect("read invocations");
        assert_eq!(stored.len(), 1);
        assert_eq!(stored[0].run_id, "run-1");
        assert_eq!(stored[0].trace.len(), 1);

        let stats = adapter
            .get_pipeline_invocation_log_stats("superadmin", "default")
            .expect("invocation stats");
        assert_eq!(stats.count, 1);
        assert_eq!(stats.pipelines.len(), 1);
        assert_eq!(stats.pipelines[0].file_rel_path, "pipelines/demo.zf.json");
        assert!(stats.trace_bytes > 0);
        assert!(stats.db_bytes > 0);

        let platform_db =
            Connection::open(tmp.path().join("platform").join("catalog.db")).expect("platform db");
        let legacy_count: i64 = platform_db
            .query_row("SELECT COUNT(*) FROM pipeline_invocations", [], |row| {
                row.get(0)
            })
            .expect("legacy count");
        assert_eq!(legacy_count, 0);

        let deleted = adapter
            .clear_pipeline_invocation_logs("superadmin", "default", Some("pipelines/demo.zf.json"))
            .expect("clear invocation logs");
        assert_eq!(deleted, 1);
        let stats = adapter
            .get_pipeline_invocation_log_stats("superadmin", "default")
            .expect("invocation stats after clear");
        assert_eq!(stats.count, 0);
    }
}