feat: add Workspace Plugin catalog and immutable artifact authority

crates/workspace-server/src/assignment_tracking_tests.rs

@@ -479,7 +479,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
                     (102, "Durable failed Subject creation recovery".into()),
                     (103, "Passkey metadata and atomic addition results".into()),
                     (104, "Converge Passkey results and Job input retention".into()),
-                    (105, "Durable Subject deletion snapshots and Job admission fences".into())
+                    (105, "Durable Subject deletion snapshots and Job admission fences".into()),
+                    (106, crate::plugin_catalog::MIGRATION_NAME.into())
                 ]
             );
             for (table, rows) in retained.iter().zip(original.iter()) {

crates/workspace-server/src/implementation_result_tests.rs

@@ -368,10 +368,13 @@ async fn implementation_result_rejects_forged_evidence_and_rolls_back_without_no
 }
 
 // Preserve seeded assignment evidence while rebuilding the actual pre-business
-// capture composition. Changing only a receipt leaves schema105 admission fences
+// capture composition. Changing only a receipt leaves current admission fences
 // and erasure guards in an otherwise legacy fixture.
 fn reconstruct_implementation_schema96(conn: &Connection) -> Result<()> {
-    assert_eq!(current_schema_version(conn)?, 105);
+    assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
+    conn.execute_batch("DROP TABLE plugin_references; DROP TABLE plugin_operations;
+        DROP TABLE plugin_uploads; DROP TABLE plugin_artifacts; DROP TABLE plugin_installations;
+        UPDATE __yoi_schema_migrations SET version=105 WHERE version=106;")?;
     let directory = tempfile::tempdir().unwrap();
     let path = directory.path().join("retained96.db");
     prepare_retained_schema(&path, 96);

crates/workspace-server/src/latest_schema.sql

@@ -1,4 +1,5 @@
 -- Canonical composed Workspace Server schema. New databases are created directly at this generation.
+-- Plugin catalog schema is composed by store.rs.
 -- Ticket creates its standalone schema first; replace those empty tables with the composed authority.
 DROP TABLE typed_ticket_event_attributes;
 DROP TABLE typed_ticket_event_references;

crates/workspace-server/src/lib.rs

@@ -12,6 +12,7 @@ pub mod hosts;
 pub mod memory_backend;
 pub mod memory_staging;
 pub mod observation;
+pub mod plugin_catalog;
 pub mod profile_settings;
 pub mod prompt_settings;
 pub mod records;
@@ -161,6 +162,8 @@ pub enum Error {
     WorkerSourceIdentity(String),
     #[error("Workspace signing identity error ({code}): {message}")]
     WorkspaceSigningIdentity { code: String, message: String },
+    #[error("Plugin catalog error: {0}")]
+    PluginCatalog(#[from] plugin_catalog::PluginCatalogError),
     #[error("Workspace Drive error: {0}")]
     Drive(#[from] workspace_drive::Error),
     #[error("DB execution capacity exceeded")]

crates/workspace-server/src/notification_routing_tests.rs

@@ -851,11 +851,11 @@ async fn notification_routing_fresh_and_upgraded_schema_and_guards_match() {
     };
     assert_eq!(current_schema_version(&conn).unwrap(), 98);
     drop(conn);
-    // Compare the complete current composition, not upgraded98 against fresh105.
+    // Compare the complete current composition, not upgraded98 against fresh current schema.
     let upgraded_store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
     let upgraded = upgraded_store
         .with_conn(|conn| {
-            assert_eq!(current_schema_version(conn)?, 105);
+            assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
             Ok(snapshot(conn))
         })
         .unwrap();

crates/workspace-server/src/plugin_catalog.rs

@@ -0,0 +1,733 @@
+//! Workspace catalog authority. The authenticated service checks current read/manage
+//! permission before entering these methods; neither a digest nor an actor string is
+//! an authorization credential. Blob publication, retention and GC share SQLite's
+//! immediate write transaction, including across independent Server connections.
+use crate::{Error, Result, store::SqliteWorkspaceStore};
+use chrono::{Duration, Utc};
+use manifest::plugin::{self, PluginPackageLimits, PluginSourceKind};
+use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
+use server_api::*;
+use sha2::{Digest, Sha256};
+use std::{
+    collections::BTreeSet,
+    fs,
+    io::{Read, Write},
+    path::{Path, PathBuf},
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+pub enum PluginCatalogError {
+    #[error("invalid Plugin catalog request")]
+    Invalid,
+    #[error("Plugin catalog resource not found")]
+    NotFound,
+    #[error("Plugin catalog operation conflicts with observed state")]
+    Conflict,
+    #[error("Plugin installation has retained references")]
+    Referenced,
+    #[error("Plugin package exceeds size limits")]
+    TooLarge,
+    #[error("Plugin storage quota exceeded")]
+    Quota,
+    #[error("Plugin artifact storage unavailable")]
+    Unavailable,
+}
+impl PluginCatalogError {
+    pub fn status(self) -> u16 {
+        match self {
+            Self::Invalid => 400,
+            Self::NotFound => 404,
+            Self::Conflict | Self::Referenced => 409,
+            Self::TooLarge => 413,
+            Self::Quota => 429,
+            Self::Unavailable => 503,
+        }
+    }
+    pub fn code(self) -> &'static str {
+        match self {
+            Self::Invalid => "plugin_invalid",
+            Self::NotFound => "plugin_not_found",
+            Self::Conflict => "plugin_conflict",
+            Self::Referenced => "plugin_referenced",
+            Self::TooLarge => "plugin_too_large",
+            Self::Quota => "plugin_quota",
+            Self::Unavailable => "plugin_unavailable",
+        }
+    }
+}
+
+const WORKSPACE_BYTES: i64 = 1024 * 1024 * 1024;
+const ACTOR_UPLOAD_BYTES: i64 = 256 * 1024 * 1024;
+const MAX_UPLOADS: i64 = 128;
+const MAX_ACTOR_UPLOADS: i64 = 32;
+const MAX_INSTALLATIONS: i64 = 128;
+const MAX_ARTIFACTS: i64 = 64;
+
+pub(crate) const MIGRATION_NAME: &str =
+    "Workspace Plugin catalog and immutable artifact references";
+pub(crate) fn migrate(conn: &Connection) -> Result<()> {
+    let version: i64 = conn.query_row(
+        "SELECT max(version) FROM __yoi_schema_migrations",
+        [],
+        |r| r.get(0),
+    )?;
+    if version != 105 {
+        return Err(Error::Store(
+            "Plugin catalog migration requires schema 105".into(),
+        ));
+    }
+    let tx = conn.unchecked_transaction()?;
+    tx.execute_batch(include_str!("plugin_catalog_schema.sql"))?;
+    tx.execute(
+        "INSERT INTO __yoi_schema_migrations(version,name) VALUES(106,?1)",
+        [MIGRATION_NAME],
+    )?;
+    verify(&tx)?;
+    tx.commit()?;
+    Ok(())
+}
+pub(crate) fn verify(conn: &Connection) -> Result<()> {
+    let expected = Connection::open_in_memory()?;
+    expected.execute_batch(include_str!("plugin_catalog_schema.sql"))?;
+    let objects = expected.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND tbl_name LIKE 'plugin_%' ORDER BY name")?
+        .query_map([], |r| Ok((r.get::<_,String>(0)?, r.get::<_,String>(1)?)))?
+        .collect::<rusqlite::Result<Vec<_>>>()?;
+    for (name, sql) in objects {
+        let actual = conn
+            .query_row(
+                "SELECT sql FROM sqlite_schema WHERE name=?1",
+                [&name],
+                |r| r.get::<_, String>(0),
+            )
+            .optional()?;
+        if actual.as_deref() != Some(sql.as_str()) {
+            return Err(Error::Store(
+                "Plugin catalog storage schema has drifted".into(),
+            ));
+        }
+    }
+    Ok(())
+}
+fn fail<T>(error: PluginCatalogError) -> Result<T> {
+    Err(Error::PluginCatalog(error))
+}
+fn encode<T: Serialize>(value: &T) -> Result<String> {
+    serde_json::to_string(value).map_err(|_| PluginCatalogError::Unavailable.into())
+}
+fn decode<T: DeserializeOwned>(value: &str) -> Result<T> {
+    serde_json::from_str(value).map_err(|_| PluginCatalogError::Unavailable.into())
+}
+fn hash(bytes: &[u8]) -> String {
+    Sha256::digest(bytes)
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect()
+}
+fn id(prefix: &str) -> String {
+    format!("{prefix}-{}", uuid::Uuid::now_v7())
+}
+fn bounded(value: &str, max: usize) -> bool {
+    !value.is_empty() && value.len() <= max && !value.chars().any(char::is_control)
+}
+fn digest(value: &str) -> bool {
+    value.strip_prefix("sha256:").is_some_and(hex_digest)
+}
+fn hex_digest(value: &str) -> bool {
+    value.len() == 64
+        && value
+            .bytes()
+            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+}
+fn workspace_exists(conn: &Connection, workspace: &str) -> Result<()> {
+    if !conn.query_row(
+        "SELECT EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=?1)",
+        [workspace],
+        |r| r.get::<_, bool>(0),
+    )? {
+        return fail(PluginCatalogError::NotFound);
+    }
+    Ok(())
+}
+fn validate_identity(actor: &str, key: &str) -> Result<()> {
+    if !bounded(actor, 512) || !bounded(key, 128) {
+        return fail(PluginCatalogError::Invalid);
+    }
+    Ok(())
+}
+fn storage<T>(result: std::io::Result<T>) -> Result<T> {
+    result.map_err(|_| PluginCatalogError::Unavailable.into())
+}
+fn blob_path(root: &Path, artifact: &str) -> Result<PathBuf> {
+    if !digest(artifact) {
+        return fail(PluginCatalogError::Invalid);
+    }
+    Ok(root.join("blobs").join(&artifact[7..]))
+}
+struct PendingFile(PathBuf);
+impl Drop for PendingFile {
+    fn drop(&mut self) {
+        let _ = fs::remove_file(&self.0);
+    }
+}
+fn read_blob(root: &Path, artifact: &str, size: u64) -> Result<Vec<u8>> {
+    let path = blob_path(root, artifact)?;
+    let metadata = storage(fs::symlink_metadata(&path))?;
+    if !metadata.is_file()
+        || metadata.len() != size
+        || size > PluginPackageLimits::default().max_package_size_bytes
+    {
+        return fail(PluginCatalogError::Unavailable);
+    }
+    let file = storage(fs::File::open(path))?;
+    let mut bytes = Vec::new();
+    storage(file.take(size + 1).read_to_end(&mut bytes))?;
+    if bytes.len() as u64 != size || format!("sha256:{}", hash(&bytes)) != artifact {
+        return fail(PluginCatalogError::Unavailable);
+    }
+    Ok(bytes)
+}
+// Caller holds SQLite write ownership until the upload reference commits. GC
+// cannot unlink a candidate between publication and the first durable reference.
+fn publish_blob(root: &Path, artifact: &str, bytes: &[u8]) -> Result<()> {
+    let path = blob_path(root, artifact)?;
+    storage(fs::create_dir_all(root.join("blobs")))?;
+    if path.exists() {
+        read_blob(root, artifact, bytes.len() as u64)?;
+        return Ok(());
+    }
+    let pending = PendingFile(root.join("blobs").join(id(".pending")));
+    let mut file = storage(
+        fs::OpenOptions::new()
+            .create_new(true)
+            .write(true)
+            .open(&pending.0),
+    )?;
+    storage(file.write_all(bytes))?;
+    storage(file.sync_all())?;
+    match fs::hard_link(&pending.0, &path) {
+        Ok(()) => (),
+        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
+            read_blob(root, artifact, bytes.len() as u64)?;
+        }
+        Err(_) => return fail(PluginCatalogError::Unavailable),
+    }
+    storage(fs::File::open(root.join("blobs")))?
+        .sync_all()
+        .map_err(|_| Error::PluginCatalog(PluginCatalogError::Unavailable))?;
+    Ok(())
+}
+fn artifact_metadata(
+    package: &plugin::MaterializedPluginPackage,
+    canonical: &plugin::CanonicalPluginArtifact,
+) -> Result<PluginArtifact> {
+    let m = &package.package.manifest;
+    let schema_digest = format!("sha256:{}", hash(m.config_schema.as_bytes()));
+    Ok(PluginArtifact {
+        package_id: m.id.clone(),
+        name: m.name.clone(),
+        version: m.version.clone(),
+        description: m.description.clone(),
+        package_digest: canonical.package_digest.clone(),
+        artifact_digest: canonical.artifact_digest.clone(),
+        artifact_size: u32::try_from(canonical.artifact_size)
+            .map_err(|_| PluginCatalogError::TooLarge)?,
+        schema_version: m.schema_version,
+        world: m.runtime.world.clone(),
+        required_imports: m.runtime.required_imports.clone(),
+        tools: m
+            .tools
+            .iter()
+            .map(|t| PluginToolMetadata {
+                name: t.name.clone(),
+                description: t.description.clone(),
+                input_schema: t.input_schema.clone(),
+                output_schema: t.output_schema.clone(),
+                error_codes: t.error_codes.clone(),
+                external_write: t.external_write,
+            })
+            .collect(),
+        connections: decode(&encode(&m.connections)?)?,
+        requested_grants: decode(&encode(&m.requested_grants)?)?,
+        config_schema: PluginSchemaContribution {
+            identity: format!("{}/config", m.id),
+            version: m.version.clone(),
+            namespace: format!("{}/config/{}/{}", m.id, m.version, schema_digest),
+            digest: schema_digest,
+            dcdl: m.config_schema.clone(),
+        },
+        revoked: false,
+        revocation_reason: None,
+    })
+}
+fn load_artifact(
+    conn: &Connection,
+    workspace: &str,
+    installation: &str,
+    artifact: &str,
+) -> Result<PluginArtifact> {
+    let row=conn.query_row("SELECT metadata_json,revoked,revocation_reason FROM plugin_artifacts WHERE workspace_id=?1 AND installation_id=?2 AND artifact_digest=?3",params![workspace,installation,artifact],|r|Ok((r.get::<_,String>(0)?,r.get::<_,bool>(1)?,r.get::<_,Option<String>>(2)?))).optional()?;
+    let Some((json, revoked, reason)) = row else {
+        return fail(PluginCatalogError::NotFound);
+    };
+    let mut result: PluginArtifact = decode(&json)?;
+    result.revoked = revoked;
+    result.revocation_reason = reason;
+    Ok(result)
+}
+fn load_installation(
+    conn: &Connection,
+    workspace: &str,
+    installation: &str,
+) -> Result<PluginInstallation> {
+    let row=conn.query_row("SELECT alias,package_id,last_operation,uninstalled FROM plugin_installations WHERE workspace_id=?1 AND installation_id=?2",params![workspace,installation],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,bool>(3)?))).optional()?;
+    let Some((alias, package_id, last_operation, uninstalled)) = row else {
+        return fail(PluginCatalogError::NotFound);
+    };
+    let digests=conn.prepare("SELECT artifact_digest FROM plugin_artifacts WHERE workspace_id=?1 AND installation_id=?2 ORDER BY artifact_digest")?.query_map(params![workspace,installation],|r|r.get::<_,String>(0))?.collect::<rusqlite::Result<Vec<_>>>()?;
+    let artifacts = digests
+        .iter()
+        .map(|d| load_artifact(conn, workspace, installation, d))
+        .collect::<Result<Vec<_>>>()?;
+    Ok(PluginInstallation {
+        installation_id: installation.into(),
+        alias,
+        package_id,
+        last_operation,
+        uninstalled,
+        artifacts,
+    })
+}
+fn compare_installation(
+    conn: &Connection,
+    workspace: &str,
+    installation: &str,
+    expected: &str,
+) -> Result<PluginInstallation> {
+    let current = load_installation(conn, workspace, installation)?;
+    if current.uninstalled {
+        return fail(PluginCatalogError::NotFound);
+    }
+    if current.last_operation != expected {
+        return fail(PluginCatalogError::Conflict);
+    }
+    Ok(current)
+}
+fn load_upload(
+    conn: &Connection,
+    workspace: &str,
+    actor: &str,
+    upload: &str,
+) -> Result<(PluginArtifact, i64)> {
+    let row=conn.query_row("SELECT metadata_json,expires_at FROM plugin_uploads WHERE workspace_id=?1 AND actor=?2 AND upload_id=?3",params![workspace,actor,upload],|r|Ok((r.get::<_,String>(0)?,r.get::<_,i64>(1)?))).optional()?;
+    let Some((json, expires)) = row else {
+        return fail(PluginCatalogError::NotFound);
+    };
+    Ok((decode(&json)?, expires))
+}
+fn kind_name(kind: PluginOperationKind) -> &'static str {
+    match kind {
+        PluginOperationKind::Install => "install",
+        PluginOperationKind::AddArtifact => "add_artifact",
+        PluginOperationKind::Revoke => "revoke",
+        PluginOperationKind::Uninstall => "uninstall",
+    }
+}
+fn replay(
+    conn: &Connection,
+    workspace: &str,
+    actor: &str,
+    kind: PluginOperationKind,
+    key: &str,
+    fingerprint: &str,
+) -> Result<Option<PluginOperationResponse>> {
+    let saved=conn.query_row("SELECT fingerprint,result_json FROM plugin_operations WHERE workspace_id=?1 AND actor=?2 AND kind=?3 AND operation_key=?4",params![workspace,actor,kind_name(kind),key],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?))).optional()?;
+    if let Some((old, json)) = saved {
+        if old != fingerprint {
+            return fail(PluginCatalogError::Conflict);
+        }
+        return Ok(Some(decode(&json)?));
+    }
+    Ok(None)
+}
+fn save_operation(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    actor: &str,
+    key: &str,
+    fingerprint: &str,
+    kind: PluginOperationKind,
+    operation: &str,
+    installation: &str,
+) -> Result<PluginOperationResponse> {
+    tx.execute("UPDATE plugin_installations SET last_operation=?3 WHERE workspace_id=?1 AND installation_id=?2",params![workspace,installation,operation])?;
+    let result = PluginOperationResponse {
+        operation_id: operation.into(),
+        kind,
+        installation: load_installation(tx, workspace, installation)?,
+    };
+    tx.execute("INSERT INTO plugin_operations(workspace_id,operation_id,actor,kind,operation_key,fingerprint,result_json) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,operation,actor,kind_name(kind),key,fingerprint,encode(&result)?])?;
+    Ok(result)
+}
+fn add_artifact(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    installation: &str,
+    artifact: &PluginArtifact,
+) -> Result<()> {
+    let existing=tx.query_row("SELECT artifact_digest,package_digest,revoked FROM plugin_artifacts WHERE workspace_id=?1 AND installation_id=?2 AND version=?3",params![workspace,installation,artifact.version],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,bool>(2)?))).optional()?;
+    if let Some((old, package, revoked)) = existing {
+        if old != artifact.artifact_digest || package != artifact.package_digest || revoked {
+            return fail(PluginCatalogError::Conflict);
+        }
+        return Ok(());
+    }
+    let count: i64 = tx.query_row(
+        "SELECT count(*) FROM plugin_artifacts WHERE workspace_id=?1 AND installation_id=?2",
+        params![workspace, installation],
+        |r| r.get(0),
+    )?;
+    if count >= MAX_ARTIFACTS {
+        return fail(PluginCatalogError::Quota);
+    }
+    check_workspace_bytes(tx, workspace, artifact.artifact_size as i64)?;
+    tx.execute("INSERT INTO plugin_artifacts(workspace_id,installation_id,artifact_digest,package_digest,version,artifact_size,metadata_json) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,installation,artifact.artifact_digest,artifact.package_digest,artifact.version,artifact.artifact_size as i64,encode(artifact)?])?;
+    Ok(())
+}
+fn check_workspace_bytes(conn: &Connection, workspace: &str, additional: i64) -> Result<()> {
+    let bytes:i64=conn.query_row("SELECT (SELECT coalesce(sum(a.artifact_size),0) FROM plugin_artifacts a JOIN plugin_installations i USING(workspace_id,installation_id) WHERE a.workspace_id=?1 AND i.uninstalled=0) + (SELECT coalesce(sum(artifact_size),0) FROM plugin_uploads WHERE workspace_id=?1 AND expires_at>?2)",params![workspace,Utc::now().timestamp()],|r|r.get(0))?;
+    if bytes + additional > WORKSPACE_BYTES {
+        return fail(PluginCatalogError::Quota);
+    }
+    Ok(())
+}
+
+impl SqliteWorkspaceStore {
+    pub fn plugin_upload(
+        &self,
+        root: &Path,
+        workspace: &str,
+        actor: &str,
+        bytes: &[u8],
+    ) -> Result<PluginUploadResponse> {
+        if !bounded(actor, 512) {
+            return fail(PluginCatalogError::Invalid);
+        }
+        let limits = PluginPackageLimits::default();
+        if bytes.len() as u64 > limits.max_package_size_bytes {
+            return fail(PluginCatalogError::TooLarge);
+        }
+        let package = plugin::read_plugin_package_bytes(bytes, PluginSourceKind::Explicit, &limits)
+            .map_err(|e| {
+                if e.kind == plugin::PluginDiagnosticKind::Bounds {
+                    PluginCatalogError::TooLarge
+                } else {
+                    PluginCatalogError::Invalid
+                }
+            })?;
+        let canonical = plugin::pack_plugin_package(&package, &limits)
+            .map_err(|_| PluginCatalogError::Invalid)?;
+        let artifact = artifact_metadata(&package, &canonical)?;
+        self.with_conn_mut(|conn| {
+            let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            workspace_exists(&tx,workspace)?;
+            let now=Utc::now();
+            let (count,own,own_bytes):(i64,i64,i64)=tx.query_row("SELECT count(*),coalesce(sum(actor=?2),0),coalesce(sum(CASE WHEN actor=?2 THEN artifact_size ELSE 0 END),0) FROM plugin_uploads WHERE workspace_id=?1 AND expires_at>?3",params![workspace,actor,now.timestamp()],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?;
+            if count>=MAX_UPLOADS || own>=MAX_ACTOR_UPLOADS || own_bytes+canonical.artifact_size as i64>ACTOR_UPLOAD_BYTES {return fail(PluginCatalogError::Quota);}
+            check_workspace_bytes(&tx,workspace,canonical.artifact_size as i64)?;
+            publish_blob(root,&canonical.artifact_digest,&canonical.bytes)?;
+            let upload_id=id("pu"); let expires=now+Duration::hours(24);
+            tx.execute("INSERT INTO plugin_uploads(workspace_id,upload_id,actor,expires_at,artifact_digest,artifact_size,metadata_json) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,upload_id,actor,expires.timestamp(),artifact.artifact_digest,artifact.artifact_size as i64,encode(&artifact)?])?;
+            tx.commit()?;
+            Ok(PluginUploadResponse{upload_id,expires_at:expires.to_rfc3339_opts(chrono::SecondsFormat::Secs,true),artifact:artifact.clone()})
+        })
+    }
+    pub fn plugin_install(
+        &self,
+        workspace: &str,
+        actor: &str,
+        request: &PluginInstallRequest,
+    ) -> Result<PluginOperationResponse> {
+        validate_identity(actor, &request.operation_key)?;
+        if !bounded(&request.alias, 128) || request.alias.trim() != request.alias {
+            return fail(PluginCatalogError::Invalid);
+        }
+        self.with_conn_mut(|conn| {
+            let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let (artifact,expires)=load_upload(&tx,workspace,actor,&request.upload_id)?;
+            let fingerprint=hash(encode(&(request,&artifact.package_digest,&artifact.artifact_digest))?.as_bytes());
+            if let Some(saved)=replay(&tx,workspace,actor,PluginOperationKind::Install,&request.operation_key,&fingerprint)? {return Ok(saved);}
+            if expires<=Utc::now().timestamp() {return fail(PluginCatalogError::NotFound);}
+            let (count,alias):(i64,bool)=tx.query_row("SELECT count(*),coalesce(max(alias=?2),0) FROM plugin_installations WHERE workspace_id=?1 AND uninstalled=0",params![workspace,request.alias],|r|Ok((r.get(0)?,r.get(1)?)))?;
+            if alias {return fail(PluginCatalogError::Conflict);}
+            if count>=MAX_INSTALLATIONS {return fail(PluginCatalogError::Quota);}
+            let installation=id("pi"); let operation=id("po");
+            tx.execute("INSERT INTO plugin_installations(workspace_id,installation_id,alias,package_id,last_operation) VALUES(?1,?2,?3,?4,?5)",params![workspace,installation,request.alias,artifact.package_id,operation])?;
+            add_artifact(&tx,workspace,&installation,&artifact)?;
+            let result=save_operation(&tx,workspace,actor,&request.operation_key,&fingerprint,PluginOperationKind::Install,&operation,&installation)?;
+            tx.commit()?; Ok(result)
+        })
+    }
+    pub fn plugin_add_artifact(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &PluginAddArtifactRequest,
+    ) -> Result<PluginOperationResponse> {
+        validate_identity(actor, &request.operation_key)?;
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let (artifact, expires) = load_upload(&tx, workspace, actor, &request.upload_id)?;
+            let fingerprint = hash(
+                encode(&(
+                    installation,
+                    request,
+                    &artifact.package_digest,
+                    &artifact.artifact_digest,
+                ))?
+                .as_bytes(),
+            );
+            if let Some(saved) = replay(
+                &tx,
+                workspace,
+                actor,
+                PluginOperationKind::AddArtifact,
+                &request.operation_key,
+                &fingerprint,
+            )? {
+                return Ok(saved);
+            }
+            if expires <= Utc::now().timestamp() {
+                return fail(PluginCatalogError::NotFound);
+            }
+            let current = compare_installation(
+                &tx,
+                workspace,
+                installation,
+                &request.expected_last_operation,
+            )?;
+            if current.package_id != artifact.package_id {
+                return fail(PluginCatalogError::Conflict);
+            }
+            add_artifact(&tx, workspace, installation, &artifact)?;
+            let result = save_operation(
+                &tx,
+                workspace,
+                actor,
+                &request.operation_key,
+                &fingerprint,
+                PluginOperationKind::AddArtifact,
+                &id("po"),
+                installation,
+            )?;
+            tx.commit()?;
+            Ok(result)
+        })
+    }
+    pub fn plugin_revoke(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &PluginRevokeRequest,
+    ) -> Result<PluginOperationResponse> {
+        validate_identity(actor, &request.operation_key)?;
+        if !digest(&request.artifact_digest) || !bounded(&request.reason, 1024) {
+            return fail(PluginCatalogError::Invalid);
+        }
+        self.with_conn_mut(|conn| {
+            let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let fingerprint=hash(encode(&(installation,request))?.as_bytes());
+            if let Some(saved)=replay(&tx,workspace,actor,PluginOperationKind::Revoke,&request.operation_key,&fingerprint)? {return Ok(saved);}
+            compare_installation(&tx,workspace,installation,&request.expected_last_operation)?;
+            let artifact=load_artifact(&tx,workspace,installation,&request.artifact_digest)?;
+            if artifact.revoked {return fail(PluginCatalogError::Conflict);}
+            tx.execute("UPDATE plugin_artifacts SET revoked=1,revocation_reason=?4 WHERE workspace_id=?1 AND installation_id=?2 AND artifact_digest=?3",params![workspace,installation,request.artifact_digest,request.reason])?;
+            let result=save_operation(&tx,workspace,actor,&request.operation_key,&fingerprint,PluginOperationKind::Revoke,&id("po"),installation)?;
+            tx.commit()?; Ok(result)
+        })
+    }
+    pub fn plugin_uninstall(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &PluginUninstallRequest,
+    ) -> Result<PluginOperationResponse> {
+        validate_identity(actor, &request.operation_key)?;
+        self.with_conn_mut(|conn| {
+            let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let fingerprint=hash(encode(&(installation,request))?.as_bytes());
+            if let Some(saved)=replay(&tx,workspace,actor,PluginOperationKind::Uninstall,&request.operation_key,&fingerprint)? {return Ok(saved);}
+            compare_installation(&tx,workspace,installation,&request.expected_last_operation)?;
+            let referenced:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM plugin_references WHERE workspace_id=?1 AND installation_id=?2)",params![workspace,installation],|r|r.get(0))?;
+            if referenced {return fail(PluginCatalogError::Referenced);}
+            tx.execute("UPDATE plugin_installations SET uninstalled=1 WHERE workspace_id=?1 AND installation_id=?2",params![workspace,installation])?;
+            let result=save_operation(&tx,workspace,actor,&request.operation_key,&fingerprint,PluginOperationKind::Uninstall,&id("po"),installation)?;
+            tx.commit()?; Ok(result)
+        })
+    }
+    pub fn plugin_show(&self, workspace: &str, installation: &str) -> Result<PluginInstallation> {
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            let result = load_installation(&tx, workspace, installation)?;
+            tx.commit()?;
+            Ok(result)
+        })
+    }
+    pub fn plugin_operation(
+        &self,
+        workspace: &str,
+        operation: &str,
+    ) -> Result<PluginOperationResponse> {
+        self.with_conn(|conn| {
+            let json=conn.query_row("SELECT result_json FROM plugin_operations WHERE workspace_id=?1 AND operation_id=?2",params![workspace,operation],|r|r.get::<_,String>(0)).optional()?;
+            match json {Some(json)=>decode(&json),None=>fail(PluginCatalogError::NotFound)}
+        })
+    }
+    pub fn plugin_list(
+        &self,
+        workspace: &str,
+        request: &PluginListRequest,
+    ) -> Result<PluginListResponse> {
+        let limit = request.limit.unwrap_or(50);
+        if limit == 0 || limit > 100 {
+            return fail(PluginCatalogError::Invalid);
+        }
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            let conn = &tx;
+            workspace_exists(conn,workspace)?;
+            let after=if let Some(cursor)=&request.cursor {
+                // Server generated operation identity is an opaque cursor scoped by
+                // a persisted membership check, never interpreted as an arbitrary offset.
+                let value=conn.query_row("SELECT json_extract(result_json,'$.installation.installation_id') FROM plugin_operations WHERE workspace_id=?1 AND operation_id=?2",params![workspace,cursor],|r|r.get::<_,String>(0)).optional()?;
+                value.ok_or(PluginCatalogError::Invalid)?
+            } else {String::new()};
+            let mut installations=conn.prepare("SELECT installation_id,alias,package_id,last_operation,uninstalled FROM plugin_installations WHERE workspace_id=?1 AND installation_id>?2 ORDER BY installation_id LIMIT ?3")?.query_map(params![workspace,after,limit+1],|r|Ok(PluginInstallationSummary{installation_id:r.get(0)?,alias:r.get(1)?,package_id:r.get(2)?,last_operation:r.get(3)?,uninstalled:r.get(4)?}))?.collect::<rusqlite::Result<Vec<_>>>()?;
+            let more=installations.len()>limit as usize;
+            installations.truncate(limit as usize);
+            let next_cursor=if more {installations.last().map(|i|i.last_operation.clone())} else {None};
+            tx.commit()?;
+            Ok(PluginListResponse{installations,next_cursor})
+        })
+    }
+    pub fn plugin_artifact_bytes(
+        &self,
+        root: &Path,
+        workspace: &str,
+        installation: &str,
+        artifact_digest: &str,
+    ) -> Result<Vec<u8>> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let current = load_installation(&tx, workspace, installation)?;
+            if current.uninstalled {
+                return fail(PluginCatalogError::NotFound);
+            }
+            let artifact = load_artifact(&tx, workspace, installation, artifact_digest)?;
+            if artifact.revoked {
+                return fail(PluginCatalogError::Conflict);
+            }
+            let bytes = read_blob(root, artifact_digest, u64::from(artifact.artifact_size))?;
+            tx.commit()?;
+            Ok(bytes)
+        })
+    }
+    pub fn plugin_gc(&self, root: &Path) -> Result<u64> {
+        self.with_conn_mut(|conn| {
+            let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let retained=tx.prepare("SELECT a.artifact_digest FROM plugin_artifacts a JOIN plugin_installations i USING(workspace_id,installation_id) WHERE i.uninstalled=0 UNION SELECT artifact_digest FROM plugin_uploads WHERE expires_at>?1")?.query_map([Utc::now().timestamp()],|r|r.get::<_,String>(0))?.collect::<rusqlite::Result<BTreeSet<_>>>()?;
+            let entries=match fs::read_dir(root.join("blobs")) {Ok(entries)=>entries,Err(e) if e.kind()==std::io::ErrorKind::NotFound=>return Ok(0),Err(_)=>return fail(PluginCatalogError::Unavailable)};
+            let mut removed=0;
+            for entry in entries {
+                let entry=storage(entry)?; let name=entry.file_name(); let Some(name)=name.to_str() else {continue;};
+                if (hex_digest(name) || name.starts_with(".pending-")) && !retained.contains(&format!("sha256:{name}")) {
+                    storage(fs::remove_file(entry.path()))?; removed+=1;
+                }
+            }
+            storage(fs::File::open(root.join("blobs")))?.sync_all().map_err(|_|Error::PluginCatalog(PluginCatalogError::Unavailable))?;
+            tx.commit()?; Ok(removed)
+        })
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum PluginReferenceKind {
+    Config,
+    Pending,
+    Active,
+    Restorable,
+}
+impl PluginReferenceKind {
+    fn name(self) -> &'static str {
+        match self {
+            Self::Config => "config",
+            Self::Pending => "pending",
+            Self::Active => "active",
+            Self::Restorable => "restorable",
+        }
+    }
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PluginReference {
+    pub kind: PluginReferenceKind,
+    pub owner: String,
+    pub installation_id: String,
+    pub expected_last_operation: String,
+    pub package_digest: String,
+    pub artifact_digest: String,
+}
+/// Called inside the config/plan owner's transaction. The same write transaction
+/// must commit that owner's content CAS and its reference changes. This function
+/// does not persist a second copy of config or policy.
+pub fn plugin_artifact_admit(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    installation: &str,
+    expected_last_operation: &str,
+    package_digest: &str,
+    artifact_digest: &str,
+) -> Result<PluginArtifact> {
+    compare_installation(tx, workspace, installation, expected_last_operation)?;
+    let artifact = load_artifact(tx, workspace, installation, artifact_digest)?;
+    if artifact.revoked || artifact.package_digest != package_digest {
+        return fail(PluginCatalogError::Conflict);
+    }
+    Ok(artifact)
+}
+pub fn plugin_reference_put(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    reference: &PluginReference,
+) -> Result<()> {
+    if !bounded(&reference.owner, 512) {
+        return fail(PluginCatalogError::Invalid);
+    }
+    plugin_artifact_admit(
+        tx,
+        workspace,
+        &reference.installation_id,
+        &reference.expected_last_operation,
+        &reference.package_digest,
+        &reference.artifact_digest,
+    )?;
+    tx.execute("INSERT INTO plugin_references(workspace_id,kind,owner,installation_id,artifact_digest) VALUES(?1,?2,?3,?4,?5) ON CONFLICT DO NOTHING",params![workspace,reference.kind.name(),reference.owner,reference.installation_id,reference.artifact_digest])?;
+    Ok(())
+}
+pub fn plugin_reference_remove(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    kind: PluginReferenceKind,
+    owner: &str,
+    installation: &str,
+    artifact_digest: &str,
+) -> Result<()> {
+    tx.execute("DELETE FROM plugin_references WHERE workspace_id=?1 AND kind=?2 AND owner=?3 AND installation_id=?4 AND artifact_digest=?5",params![workspace,kind.name(),owner,installation,artifact_digest])?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests;

crates/workspace-server/src/plugin_catalog/migration_tests.rs

@@ -0,0 +1,81 @@
+//! Exercise production migration/open against retained SQLite fixtures, including
+//! the oldest retained transition, marker failure rollback and real FK/CHECKs.
+use super::*;
+
+fn catalog_schema(conn: &Connection) -> Vec<(String, String)> {
+    conn.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND tbl_name LIKE 'plugin_%' ORDER BY name").unwrap().query_map([],|r|Ok((r.get(0)?,r.get(1)?))).unwrap().collect::<rusqlite::Result<Vec<_>>>().unwrap()
+}
+#[test]
+fn plugin_catalog_migration_matches_fresh_schema_from_oldest_retained_and_105() {
+    let dir = tempfile::tempdir().unwrap();
+    let fresh = SqliteWorkspaceStore::open_for_test(dir.path().join("fresh.db")).unwrap();
+    let canonical = fresh.with_conn(|c| Ok(catalog_schema(c))).unwrap();
+    for version in [50, 105] {
+        let path = dir.path().join(format!("v{version}.db"));
+        if version == 50 {
+            tests::prepare_schema_v50(&path, Some("workspace-a"));
+        } else {
+            tests::prepare_retained_schema(&path, version);
+        }
+        let conn = Connection::open(&path).unwrap();
+        configure_sqlite(&conn).unwrap();
+        let before = workspace_schema_migration_history(&conn).unwrap();
+        conn.execute_batch("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) VALUES('plugin-owner','user','plugin-owner','Owner','1','1'); INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('plugin-workspace','plugin-owner','Plugins','active','1','1');").unwrap();
+        drop(conn);
+        let migrated = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+        migrated.with_conn(|c| {
+            assert_eq!(catalog_schema(c),canonical);
+            let after=workspace_schema_migration_history(c)?;
+            assert!(after.starts_with(&before));
+            assert_eq!(after.last().unwrap(),&(106,crate::plugin_catalog::MIGRATION_NAME.into()));
+            assert_eq!(c.query_row("SELECT display_name FROM workspaces WHERE workspace_id='plugin-workspace'",[],|r|r.get::<_,String>(0))?,"Plugins");
+            assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
+            assert!(c.execute("INSERT INTO plugin_installations VALUES('absent','i','alias','p','operation',0)",[]).is_err());
+            c.execute("INSERT INTO plugin_installations VALUES('plugin-workspace','i','alias','p','operation',0)",[])?;
+            assert!(c.execute("INSERT INTO plugin_installations VALUES('plugin-workspace','other','alias','p','operation',0)",[]).is_err());
+            assert!(c.execute("INSERT INTO plugin_installations VALUES('plugin-workspace','other','other','p','operation',2)",[]).is_err());
+            assert!(c.execute("INSERT INTO plugin_references VALUES('plugin-workspace','active','owner','i','unknown')",[]).is_err());
+            Ok(())
+        }).unwrap();
+    }
+}
+#[test]
+fn plugin_catalog_migration_marker_failure_rolls_back_then_retries_without_changing_applied() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("retained.db");
+    tests::prepare_retained_schema(&path, 105);
+    let conn = Connection::open(&path).unwrap();
+    configure_sqlite(&conn).unwrap();
+    let before = workspace_schema_migration_history(&conn).unwrap();
+    conn.execute_batch("CREATE TRIGGER interrupt_plugin_migration BEFORE INSERT ON __yoi_schema_migrations WHEN NEW.version=106 BEGIN SELECT RAISE(ABORT,'injected plugin migration interruption'); END;").unwrap();
+    assert!(SqliteWorkspaceStore::open_for_test(&path).is_err());
+    assert_eq!(current_schema_version(&conn).unwrap(), 105);
+    assert_eq!(workspace_schema_migration_history(&conn).unwrap(), before);
+    assert!(catalog_schema(&conn).is_empty());
+    conn.execute_batch("DROP TRIGGER interrupt_plugin_migration")
+        .unwrap();
+    let migrated = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+    migrated
+        .with_conn(|c| {
+            assert_eq!(current_schema_version(c)?, 106);
+            crate::plugin_catalog::verify(c)?;
+            Ok(())
+        })
+        .unwrap();
+}
+#[test]
+fn plugin_catalog_startup_rejects_drifted_reference_fence() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("fresh.db");
+    let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+    store
+        .with_conn(|c| {
+            c.execute_batch("DROP TRIGGER plugin_reference_live")?;
+            Ok(())
+        })
+        .unwrap();
+    drop(store);
+    assert!(
+        matches!(SqliteWorkspaceStore::open_for_test(path),Err(Error::Store(message)) if message.contains("Plugin catalog storage schema has drifted"))
+    );
+}

crates/workspace-server/src/plugin_catalog/tests.rs

@@ -0,0 +1,911 @@
+//! Real SQLite plus disk-component tests. Only authentication/HTTP and execution
+//! are outside this suite; packages use the checked-in real Rust PDK Component.
+use super::*;
+use crate::store::{ControlPlaneStore, WorkspaceRecord};
+use tempfile::{TempDir, tempdir};
+
+struct Fixture {
+    dir: TempDir,
+    store: SqliteWorkspaceStore,
+    bytes: Vec<u8>,
+}
+impl Fixture {
+    async fn new() -> Self {
+        let dir = tempdir().unwrap();
+        let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap();
+        for workspace in ["a", "b"] {
+            store
+                .upsert_workspace(&WorkspaceRecord {
+                    workspace_id: workspace.into(),
+                    owner_account_id: "owner-account".into(),
+                    display_name: workspace.into(),
+                    state: "active".into(),
+                    created_at: "2026-10-01T00:00:00Z".into(),
+                    updated_at: "2026-10-01T00:00:00Z".into(),
+                })
+                .await
+                .unwrap();
+        }
+        Self {
+            dir,
+            store,
+            bytes: package("0.1.0", ""),
+        }
+    }
+    fn root(&self) -> PathBuf {
+        self.dir.path().join("artifacts")
+    }
+    fn upload(&self) -> PluginUploadResponse {
+        self.store
+            .plugin_upload(&self.root(), "a", "actor", &self.bytes)
+            .unwrap()
+    }
+    fn install(&self, upload: &PluginUploadResponse, alias: &str) -> PluginOperationResponse {
+        self.store
+            .plugin_install(
+                "a",
+                "actor",
+                &PluginInstallRequest {
+                    upload_id: upload.upload_id.clone(),
+                    alias: alias.into(),
+                    operation_key: alias.into(),
+                },
+            )
+            .unwrap()
+    }
+    fn expire(&self) {
+        self.store
+            .with_conn(|c| {
+                c.execute("UPDATE plugin_uploads SET expires_at=0", [])?;
+                Ok(())
+            })
+            .unwrap();
+    }
+    fn count(&self, table: &str) -> i64 {
+        self.store
+            .with_conn(|c| {
+                Ok(c.query_row(&format!("SELECT count(*) FROM {table}"), [], |r| r.get(0))?)
+            })
+            .unwrap()
+    }
+}
+fn package(version: &str, extra: &str) -> Vec<u8> {
+    let dir = tempdir().unwrap();
+    let manifest =
+        include_str!("../../../../resources/plugin/templates/rust-component-tool/plugin.toml")
+            .replace("version = \"0.1.0\"", &format!("version = \"{version}\""));
+    fs::write(dir.path().join("plugin.toml"), manifest).unwrap();
+    fs::write(
+        dir.path().join("plugin.component.wasm"),
+        include_bytes!("../../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"),
+    )
+    .unwrap();
+    if !extra.is_empty() {
+        fs::write(dir.path().join("extra.txt"), extra).unwrap();
+    }
+    let parsed = plugin::read_plugin_directory(
+        dir.path(),
+        PluginSourceKind::Explicit,
+        &PluginPackageLimits::default(),
+    )
+    .unwrap();
+    plugin::pack_plugin_package(&parsed, &PluginPackageLimits::default())
+        .unwrap()
+        .bytes
+}
+fn error<T: std::fmt::Debug>(result: Result<T>, expected: PluginCatalogError) {
+    assert!(
+        matches!(result,Err(Error::PluginCatalog(actual)) if actual==expected),
+        "expected {expected:?}, got {result:?}"
+    );
+}
+fn reference(result: &PluginOperationResponse, kind: PluginReferenceKind) -> PluginReference {
+    let artifact = &result.installation.artifacts[0];
+    PluginReference {
+        kind,
+        owner: "owner/config-or-plan".into(),
+        installation_id: result.installation.installation_id.clone(),
+        expected_last_operation: result.operation_id.clone(),
+        package_digest: artifact.package_digest.clone(),
+        artifact_digest: artifact.artifact_digest.clone(),
+    }
+}
+fn uninstall_request(result: &PluginOperationResponse, key: &str) -> PluginUninstallRequest {
+    PluginUninstallRequest {
+        expected_last_operation: result.operation_id.clone(),
+        operation_key: key.into(),
+    }
+}
+
+#[tokio::test]
+async fn validated_upload_is_not_install_and_exact_published_bytes_survive_restart() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    assert_eq!(f.count("plugin_installations"), 0);
+    let installed = f.install(&upload, "echo");
+    let installation = &installed.installation.installation_id;
+    let artifact = &upload.artifact;
+    assert_ne!(artifact.package_digest, artifact.artifact_digest);
+    assert_eq!(
+        artifact.config_schema.digest,
+        format!("sha256:{}", hash(artifact.config_schema.dcdl.as_bytes()))
+    );
+    let fetched = f
+        .store
+        .plugin_artifact_bytes(&f.root(), "a", installation, &artifact.artifact_digest)
+        .unwrap();
+    assert_eq!(fetched, f.bytes);
+    assert_eq!(fetched.len() as u64, u64::from(artifact.artifact_size));
+    plugin::verify_plugin_artifact(
+        &fetched,
+        &artifact.package_digest,
+        &artifact.artifact_digest,
+        u64::from(artifact.artifact_size),
+        &PluginPackageLimits::default(),
+    )
+    .unwrap();
+    let reopened = SqliteWorkspaceStore::open_for_test(f.dir.path().join("server.db")).unwrap();
+    assert_eq!(
+        reopened.plugin_show("a", installation).unwrap(),
+        installed.installation
+    );
+    assert_eq!(
+        reopened
+            .plugin_artifact_bytes(&f.root(), "a", installation, &artifact.artifact_digest)
+            .unwrap(),
+        fetched
+    );
+    error(
+        reopened.plugin_artifact_bytes(&f.root(), "b", installation, &artifact.artifact_digest),
+        PluginCatalogError::NotFound,
+    );
+    error(
+        reopened.plugin_operation("b", &installed.operation_id),
+        PluginCatalogError::NotFound,
+    );
+    error(
+        reopened.plugin_show("b", installation),
+        PluginCatalogError::NotFound,
+    );
+}
+
+#[tokio::test]
+async fn replay_returns_original_receipt_not_latest_catalog_and_survives_uninstall() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let first = f.install(&upload, "echo");
+    let mut request = PluginInstallRequest {
+        upload_id: upload.upload_id.clone(),
+        alias: "echo".into(),
+        operation_key: "echo".into(),
+    };
+    assert_eq!(
+        f.store.plugin_install("a", "actor", &request).unwrap(),
+        first
+    );
+    request.alias = "changed".into();
+    error(
+        f.store.plugin_install("a", "actor", &request),
+        PluginCatalogError::Conflict,
+    );
+    request.alias = "echo".into();
+    error(
+        f.store.plugin_install("a", "other-actor", &request),
+        PluginCatalogError::NotFound,
+    );
+    let next = f
+        .store
+        .plugin_upload(&f.root(), "a", "actor", &package("0.2.0", ""))
+        .unwrap();
+    let added = f
+        .store
+        .plugin_add_artifact(
+            "a",
+            "actor",
+            &first.installation.installation_id,
+            &PluginAddArtifactRequest {
+                upload_id: next.upload_id,
+                expected_last_operation: first.operation_id.clone(),
+                operation_key: "add".into(),
+            },
+        )
+        .unwrap();
+    assert_eq!(
+        f.store.plugin_install("a", "actor", &request).unwrap(),
+        first
+    );
+    error(
+        f.store.plugin_uninstall(
+            "a",
+            "actor",
+            &first.installation.installation_id,
+            &uninstall_request(&first, "remove"),
+        ),
+        PluginCatalogError::Conflict,
+    );
+    let uninstalled = f
+        .store
+        .plugin_uninstall(
+            "a",
+            "actor",
+            &first.installation.installation_id,
+            &uninstall_request(&added, "remove"),
+        )
+        .unwrap();
+    assert!(uninstalled.installation.uninstalled);
+    f.expire();
+    f.store.plugin_gc(&f.root()).unwrap();
+    assert_eq!(
+        f.store.plugin_install("a", "actor", &request).unwrap(),
+        first
+    );
+    assert_eq!(
+        f.store
+            .plugin_uninstall(
+                "a",
+                "actor",
+                &first.installation.installation_id,
+                &uninstall_request(&added, "remove")
+            )
+            .unwrap(),
+        uninstalled
+    );
+    assert_eq!(
+        f.store.plugin_operation("a", &first.operation_id).unwrap(),
+        first
+    );
+    assert_eq!(f.count("plugin_operations"), 3);
+    assert_eq!(f.count("plugin_artifacts"), 2);
+}
+
+#[tokio::test]
+async fn version_collision_and_revoke_are_not_overwrite_or_reactivation() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let first = f.install(&upload, "echo");
+    let installation = &first.installation.installation_id;
+    let collision = f
+        .store
+        .plugin_upload(
+            &f.root(),
+            "a",
+            "actor",
+            &package("0.1.0", "different bytes"),
+        )
+        .unwrap();
+    error(
+        f.store.plugin_add_artifact(
+            "a",
+            "actor",
+            installation,
+            &PluginAddArtifactRequest {
+                upload_id: collision.upload_id,
+                expected_last_operation: first.operation_id.clone(),
+                operation_key: "collide".into(),
+            },
+        ),
+        PluginCatalogError::Conflict,
+    );
+    let revoked = f
+        .store
+        .plugin_revoke(
+            "a",
+            "actor",
+            installation,
+            &PluginRevokeRequest {
+                artifact_digest: upload.artifact.artifact_digest.clone(),
+                expected_last_operation: first.operation_id.clone(),
+                reason: "withdrawn".into(),
+                operation_key: "revoke".into(),
+            },
+        )
+        .unwrap();
+    assert!(revoked.installation.artifacts[0].revoked);
+    let add = PluginAddArtifactRequest {
+        upload_id: upload.upload_id.clone(),
+        expected_last_operation: revoked.operation_id.clone(),
+        operation_key: "reactivate".into(),
+    };
+    error(
+        f.store
+            .plugin_add_artifact("a", "actor", installation, &add),
+        PluginCatalogError::Conflict,
+    );
+    error(
+        f.store.plugin_artifact_bytes(
+            &f.root(),
+            "a",
+            installation,
+            &upload.artifact.artifact_digest,
+        ),
+        PluginCatalogError::Conflict,
+    );
+    let mut pin = reference(&revoked, PluginReferenceKind::Restorable);
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            error(
+                plugin_reference_put(&tx, "a", &pin),
+                PluginCatalogError::Conflict,
+            );
+            Ok(())
+        })
+        .unwrap();
+    let other = f.install(&upload, "new-installation");
+    assert_ne!(other.installation.installation_id, *installation);
+    pin.installation_id = other.installation.installation_id.clone();
+    pin.expected_last_operation = other.operation_id;
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            plugin_reference_put(&tx, "a", &pin)?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+    let rejected=f.store.with_conn(|c|{c.execute("UPDATE plugin_artifacts SET revoked=0,revocation_reason=NULL WHERE workspace_id='a' AND installation_id=?1",[installation])?;Ok(())});
+    assert!(matches!(rejected, Err(Error::Sqlite(_))));
+}
+
+#[tokio::test]
+async fn references_compose_with_owner_transaction_rollback_and_fence_all_lifetimes() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let first = f.install(&upload, "echo");
+    let installation = &first.installation.installation_id;
+    f.store
+        .with_conn_mut(|c| {
+            c.execute(
+                "CREATE TABLE fixture_config_commits (content_digest TEXT PRIMARY KEY)",
+                [],
+            )?;
+            let tx = c.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            tx.execute(
+                "INSERT INTO fixture_config_commits VALUES ('uncommitted')",
+                [],
+            )?;
+            plugin_reference_put(&tx, "a", &reference(&first, PluginReferenceKind::Config))?;
+            tx.rollback()?;
+            Ok(())
+        })
+        .unwrap();
+    assert_eq!(f.count("plugin_references"), 0);
+    assert_eq!(f.count("fixture_config_commits"), 0);
+    for kind in [
+        PluginReferenceKind::Config,
+        PluginReferenceKind::Pending,
+        PluginReferenceKind::Active,
+        PluginReferenceKind::Restorable,
+    ] {
+        f.store
+            .with_conn_mut(|c| {
+                let tx = c.transaction_with_behavior(TransactionBehavior::Immediate)?;
+                plugin_reference_put(&tx, "a", &reference(&first, kind))?;
+                tx.commit()?;
+                Ok(())
+            })
+            .unwrap();
+        error(
+            f.store.plugin_uninstall(
+                "a",
+                "actor",
+                installation,
+                &uninstall_request(&first, "remove"),
+            ),
+            PluginCatalogError::Referenced,
+        );
+    }
+    let mut stale = reference(&first, PluginReferenceKind::Config);
+    stale.expected_last_operation = "stale".into();
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            error(
+                plugin_reference_put(&tx, "a", &stale),
+                PluginCatalogError::Conflict,
+            );
+            stale.expected_last_operation = first.operation_id.clone();
+            stale.package_digest = "0".repeat(64);
+            error(
+                plugin_reference_put(&tx, "a", &stale),
+                PluginCatalogError::Conflict,
+            );
+            error(
+                plugin_reference_put(&tx, "b", &reference(&first, PluginReferenceKind::Config)),
+                PluginCatalogError::NotFound,
+            );
+            Ok(())
+        })
+        .unwrap();
+    let revoked = f
+        .store
+        .plugin_revoke(
+            "a",
+            "actor",
+            installation,
+            &PluginRevokeRequest {
+                artifact_digest: upload.artifact.artifact_digest.clone(),
+                expected_last_operation: first.operation_id.clone(),
+                operation_key: "revoke".into(),
+                reason: "withdrawn".into(),
+            },
+        )
+        .unwrap();
+    assert_eq!(f.count("plugin_references"), 4);
+    error(
+        f.store.plugin_uninstall(
+            "a",
+            "actor",
+            installation,
+            &uninstall_request(&revoked, "remove"),
+        ),
+        PluginCatalogError::Referenced,
+    );
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            for kind in [
+                PluginReferenceKind::Config,
+                PluginReferenceKind::Pending,
+                PluginReferenceKind::Active,
+                PluginReferenceKind::Restorable,
+            ] {
+                plugin_reference_remove(
+                    &tx,
+                    "a",
+                    kind,
+                    "owner/config-or-plan",
+                    installation,
+                    &upload.artifact.artifact_digest,
+                )?;
+            }
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+    let removed = f
+        .store
+        .plugin_uninstall(
+            "a",
+            "actor",
+            installation,
+            &uninstall_request(&revoked, "remove"),
+        )
+        .unwrap();
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            let mut pin = reference(&removed, PluginReferenceKind::Pending);
+            pin.expected_last_operation = removed.operation_id.clone();
+            error(
+                plugin_reference_put(&tx, "a", &pin),
+                PluginCatalogError::NotFound,
+            );
+            Ok(())
+        })
+        .unwrap();
+}
+
+#[tokio::test]
+async fn database_failure_rolls_back_domain_and_receipt_then_exact_retry_succeeds() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    f.store.with_conn(|c|{c.execute_batch("CREATE TRIGGER fail_plugin_receipt BEFORE INSERT ON plugin_operations BEGIN SELECT RAISE(ABORT,'injected receipt failure'); END;")?;Ok(())}).unwrap();
+    let request = PluginInstallRequest {
+        upload_id: upload.upload_id,
+        alias: "echo".into(),
+        operation_key: "install".into(),
+    };
+    assert!(matches!(
+        f.store.plugin_install("a", "actor", &request),
+        Err(Error::Sqlite(_))
+    ));
+    assert_eq!(f.count("plugin_installations"), 0);
+    assert_eq!(f.count("plugin_artifacts"), 0);
+    assert_eq!(f.count("plugin_operations"), 0);
+    f.store
+        .with_conn(|c| {
+            c.execute_batch("DROP TRIGGER fail_plugin_receipt")?;
+            Ok(())
+        })
+        .unwrap();
+    let installed = f.store.plugin_install("a", "actor", &request).unwrap();
+    assert_eq!(
+        f.store.plugin_install("a", "actor", &request).unwrap(),
+        installed
+    );
+}
+
+#[tokio::test]
+async fn upload_and_publish_failures_leave_no_authority_and_restart_gc_recovers_orphans() {
+    let f = Fixture::new().await;
+    for bytes in [&b"not a zip"[..], &f.bytes[..f.bytes.len() - 2]] {
+        error(
+            f.store.plugin_upload(&f.root(), "a", "actor", bytes),
+            PluginCatalogError::Invalid,
+        );
+    }
+    error(
+        f.store.plugin_upload(
+            &f.root(),
+            "a",
+            "actor",
+            &vec![0; PluginPackageLimits::default().max_package_size_bytes as usize + 1],
+        ),
+        PluginCatalogError::TooLarge,
+    );
+    assert_eq!(f.count("plugin_uploads"), 0);
+    fs::write(f.dir.path().join("blocked"), "not a directory").unwrap();
+    error(
+        f.store
+            .plugin_upload(&f.dir.path().join("blocked"), "a", "actor", &f.bytes),
+        PluginCatalogError::Unavailable,
+    );
+    f.store.with_conn(|c|{c.execute_batch("CREATE TRIGGER fail_upload_commit BEFORE INSERT ON plugin_uploads BEGIN SELECT RAISE(ABORT,'injected upload DB failure'); END;")?;Ok(())}).unwrap();
+    assert!(matches!(
+        f.store.plugin_upload(&f.root(), "a", "actor", &f.bytes),
+        Err(Error::Sqlite(_))
+    ));
+    assert_eq!(f.count("plugin_uploads"), 0);
+    assert_eq!(fs::read_dir(f.root().join("blobs")).unwrap().count(), 1);
+    fs::write(f.root().join("blobs/.pending-interrupted"), "partial").unwrap();
+    let reopened = SqliteWorkspaceStore::open_for_test(f.dir.path().join("server.db")).unwrap();
+    assert_eq!(reopened.plugin_gc(&f.root()).unwrap(), 2);
+    assert_eq!(fs::read_dir(f.root().join("blobs")).unwrap().count(), 0);
+    f.store
+        .with_conn(|c| {
+            c.execute_batch("DROP TRIGGER fail_upload_commit")?;
+            Ok(())
+        })
+        .unwrap();
+    let upload = f.upload();
+    let first = f.install(&upload, "echo");
+    fs::write(
+        blob_path(&f.root(), &upload.artifact.artifact_digest).unwrap(),
+        "corrupt",
+    )
+    .unwrap();
+    error(
+        f.store.plugin_artifact_bytes(
+            &f.root(),
+            "a",
+            &first.installation.installation_id,
+            &upload.artifact.artifact_digest,
+        ),
+        PluginCatalogError::Unavailable,
+    );
+    error(
+        f.store.plugin_upload(&f.root(), "a", "actor", &f.bytes),
+        PluginCatalogError::Unavailable,
+    );
+}
+
+#[tokio::test]
+async fn upload_actor_workspace_expiry_and_quotas_are_enforced_without_partial_rows() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let request = PluginInstallRequest {
+        upload_id: upload.upload_id.clone(),
+        alias: "echo".into(),
+        operation_key: "install".into(),
+    };
+    error(
+        f.store.plugin_install("a", "other", &request),
+        PluginCatalogError::NotFound,
+    );
+    error(
+        f.store.plugin_install("b", "actor", &request),
+        PluginCatalogError::NotFound,
+    );
+    f.expire();
+    error(
+        f.store.plugin_install("a", "actor", &request),
+        PluginCatalogError::NotFound,
+    );
+    assert_eq!(f.count("plugin_installations"), 0);
+    for _ in 0..MAX_ACTOR_UPLOADS {
+        f.upload();
+    }
+    error(
+        f.store.plugin_upload(&f.root(), "a", "actor", &f.bytes),
+        PluginCatalogError::Quota,
+    );
+    f.store
+        .plugin_upload(&f.root(), "a", "other", &f.bytes)
+        .unwrap();
+    f.store
+        .plugin_upload(&f.root(), "b", "actor", &f.bytes)
+        .unwrap();
+    f.expire();
+    f.upload();
+    assert_eq!(f.count("plugin_operations"), 0);
+}
+
+#[tokio::test]
+async fn global_gc_keeps_other_workspace_blob_and_historical_receipts_after_last_reference() {
+    let f = Fixture::new().await;
+    let a_upload = f.upload();
+    let a = f.install(&a_upload, "echo");
+    let b_upload = f
+        .store
+        .plugin_upload(&f.root(), "b", "actor", &f.bytes)
+        .unwrap();
+    let b = f
+        .store
+        .plugin_install(
+            "b",
+            "actor",
+            &PluginInstallRequest {
+                upload_id: b_upload.upload_id,
+                alias: "echo".into(),
+                operation_key: "install".into(),
+            },
+        )
+        .unwrap();
+    f.store
+        .plugin_uninstall(
+            "a",
+            "actor",
+            &a.installation.installation_id,
+            &uninstall_request(&a, "remove"),
+        )
+        .unwrap();
+    f.expire();
+    assert_eq!(f.store.plugin_gc(&f.root()).unwrap(), 0);
+    assert_eq!(
+        f.store
+            .plugin_artifact_bytes(
+                &f.root(),
+                "b",
+                &b.installation.installation_id,
+                &a_upload.artifact.artifact_digest
+            )
+            .unwrap(),
+        f.bytes
+    );
+    f.store
+        .plugin_uninstall(
+            "b",
+            "actor",
+            &b.installation.installation_id,
+            &uninstall_request(&b, "remove"),
+        )
+        .unwrap();
+    assert_eq!(f.store.plugin_gc(&f.root()).unwrap(), 1);
+    assert_eq!(f.store.plugin_operation("a", &a.operation_id).unwrap(), a);
+    assert_eq!(f.count("plugin_artifacts"), 2);
+}
+
+#[tokio::test]
+async fn independent_connections_cannot_uninstall_or_gc_across_a_reference_transaction() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let first = f.install(&upload, "echo");
+    f.expire();
+    let second = SqliteWorkspaceStore::open_for_test(f.dir.path().join("server.db")).unwrap();
+    second
+        .with_conn(|c| {
+            c.busy_timeout(std::time::Duration::ZERO)?;
+            Ok(())
+        })
+        .unwrap();
+    f.store.with_conn_mut(|c|{
+        let tx=c.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        plugin_reference_put(&tx,"a",&reference(&first,PluginReferenceKind::Pending))?;
+        let denied=second.plugin_uninstall("a","actor",&first.installation.installation_id,&uninstall_request(&first,"remove"));
+        assert!(matches!(denied,Err(Error::Sqlite(rusqlite::Error::SqliteFailure(e,_))) if e.code==rusqlite::ErrorCode::DatabaseBusy));
+        assert!(matches!(second.plugin_gc(&f.root()),Err(Error::Sqlite(rusqlite::Error::SqliteFailure(e,_))) if e.code==rusqlite::ErrorCode::DatabaseBusy));
+        assert!(blob_path(&f.root(),&upload.artifact.artifact_digest)?.exists());
+        tx.commit()?;Ok(())
+    }).unwrap();
+    error(
+        second.plugin_uninstall(
+            "a",
+            "actor",
+            &first.installation.installation_id,
+            &uninstall_request(&first, "remove"),
+        ),
+        PluginCatalogError::Referenced,
+    );
+    assert_eq!(second.plugin_gc(&f.root()).unwrap(), 0);
+    // Reverse ordering: after removal commits, a stale plan cannot recreate a pin.
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            plugin_reference_remove(
+                &tx,
+                "a",
+                PluginReferenceKind::Pending,
+                "owner/config-or-plan",
+                &first.installation.installation_id,
+                &upload.artifact.artifact_digest,
+            )?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+    second
+        .plugin_uninstall(
+            "a",
+            "actor",
+            &first.installation.installation_id,
+            &uninstall_request(&first, "remove"),
+        )
+        .unwrap();
+    f.store
+        .with_conn_mut(|c| {
+            let tx = c.transaction()?;
+            error(
+                plugin_reference_put(&tx, "a", &reference(&first, PluginReferenceKind::Active)),
+                PluginCatalogError::NotFound,
+            );
+            Ok(())
+        })
+        .unwrap();
+    assert_eq!(f.store.plugin_gc(&f.root()).unwrap(), 1);
+}
+
+#[tokio::test]
+async fn bounded_opaque_list_cursor_is_workspace_bound_and_survives_catalog_changes() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let first = f.install(&upload, "one");
+    f.install(&upload, "two");
+    let page = f
+        .store
+        .plugin_list(
+            "a",
+            &PluginListRequest {
+                cursor: None,
+                limit: Some(1),
+            },
+        )
+        .unwrap();
+    assert_eq!(page.installations.len(), 1);
+    assert!(page.next_cursor.is_some());
+    f.store
+        .plugin_uninstall(
+            "a",
+            "actor",
+            &first.installation.installation_id,
+            &uninstall_request(&first, "remove"),
+        )
+        .unwrap();
+    let next = f
+        .store
+        .plugin_list(
+            "a",
+            &PluginListRequest {
+                cursor: page.next_cursor.clone(),
+                limit: Some(1),
+            },
+        )
+        .unwrap();
+    assert_eq!(next.installations.len(), 1);
+    assert!(next.next_cursor.is_none());
+    error(
+        f.store.plugin_list(
+            "b",
+            &PluginListRequest {
+                cursor: page.next_cursor,
+                limit: Some(1),
+            },
+        ),
+        PluginCatalogError::Invalid,
+    );
+    error(
+        f.store.plugin_list(
+            "a",
+            &PluginListRequest {
+                cursor: None,
+                limit: Some(101),
+            },
+        ),
+        PluginCatalogError::Invalid,
+    );
+    error(
+        f.store.plugin_list(
+            "a",
+            &PluginListRequest {
+                cursor: Some("nonsense".into()),
+                limit: None,
+            },
+        ),
+        PluginCatalogError::Invalid,
+    );
+}
+
+#[tokio::test]
+async fn destructive_mutations_roll_back_when_their_atomic_receipt_cannot_commit() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    let installed = f.install(&upload, "echo");
+    let installation = &installed.installation.installation_id;
+    f.store.with_conn(|c| { c.execute_batch("CREATE TRIGGER fail_receipt BEFORE INSERT ON plugin_operations BEGIN SELECT RAISE(ABORT,'receipt storage failure'); END;")?; Ok(()) }).unwrap();
+    let revoke = PluginRevokeRequest {
+        artifact_digest: upload.artifact.artifact_digest,
+        expected_last_operation: installed.operation_id.clone(),
+        reason: "withdrawn".into(),
+        operation_key: "revoke".into(),
+    };
+    assert!(matches!(
+        f.store.plugin_revoke("a", "actor", installation, &revoke),
+        Err(Error::Sqlite(_))
+    ));
+    assert_eq!(
+        f.store.plugin_show("a", installation).unwrap(),
+        installed.installation
+    );
+    let uninstall = uninstall_request(&installed, "remove");
+    assert!(matches!(
+        f.store
+            .plugin_uninstall("a", "actor", installation, &uninstall),
+        Err(Error::Sqlite(_))
+    ));
+    assert_eq!(
+        f.store.plugin_show("a", installation).unwrap(),
+        installed.installation
+    );
+    assert_eq!(f.count("plugin_operations"), 1);
+    f.store
+        .with_conn(|c| {
+            c.execute_batch("DROP TRIGGER fail_receipt")?;
+            Ok(())
+        })
+        .unwrap();
+    let revoked = f
+        .store
+        .plugin_revoke("a", "actor", installation, &revoke)
+        .unwrap();
+    assert_eq!(
+        f.store
+            .plugin_revoke("a", "actor", installation, &revoke)
+            .unwrap(),
+        revoked
+    );
+    error(
+        f.store
+            .plugin_uninstall("a", "actor", installation, &uninstall),
+        PluginCatalogError::Conflict,
+    );
+    let mut changed = revoke.clone();
+    changed.reason = "changed reason".into();
+    error(
+        f.store.plugin_revoke("a", "actor", installation, &changed),
+        PluginCatalogError::Conflict,
+    );
+}
+
+#[tokio::test]
+async fn upload_workspace_quota_counts_all_actors_and_installation_quota_is_atomic() {
+    let f = Fixture::new().await;
+    let upload = f.upload();
+    // Clone real immutable upload references to the same published bytes to put
+    // accounting precisely at its global boundary without allocating duplicate blobs.
+    f.store.with_conn(|c| {
+        for n in 1..MAX_UPLOADS {
+            c.execute("INSERT INTO plugin_uploads SELECT workspace_id,?1,?2,expires_at,artifact_digest,artifact_size,metadata_json FROM plugin_uploads WHERE upload_id=?3",params![format!("upload-{n}"),format!("actor-{n}"),upload.upload_id])?;
+        }
+        Ok(())
+    }).unwrap();
+    error(
+        f.store
+            .plugin_upload(&f.root(), "a", "fresh-actor", &f.bytes),
+        PluginCatalogError::Quota,
+    );
+    assert_eq!(f.count("plugin_uploads"), MAX_UPLOADS);
+    for n in 0..MAX_INSTALLATIONS {
+        f.install(&upload, &format!("install-{n}"));
+    }
+    let request = PluginInstallRequest {
+        upload_id: upload.upload_id,
+        alias: "excess".into(),
+        operation_key: "excess".into(),
+    };
+    error(
+        f.store.plugin_install("a", "actor", &request),
+        PluginCatalogError::Quota,
+    );
+    assert_eq!(f.count("plugin_installations"), MAX_INSTALLATIONS);
+    assert_eq!(f.count("plugin_operations"), MAX_INSTALLATIONS);
+}

crates/workspace-server/src/plugin_catalog_schema.sql

@@ -0,0 +1,65 @@
+CREATE TABLE plugin_installations (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    installation_id TEXT NOT NULL,
+    alias TEXT NOT NULL CHECK(length(alias) BETWEEN 1 AND 128),
+    package_id TEXT NOT NULL,
+    last_operation TEXT NOT NULL,
+    uninstalled INTEGER NOT NULL DEFAULT 0 CHECK(uninstalled IN (0,1)),
+    PRIMARY KEY(workspace_id, installation_id)
+);
+CREATE UNIQUE INDEX plugin_live_alias ON plugin_installations(workspace_id, alias) WHERE uninstalled=0;
+CREATE TABLE plugin_artifacts (
+    workspace_id TEXT NOT NULL,
+    installation_id TEXT NOT NULL,
+    artifact_digest TEXT NOT NULL CHECK(length(artifact_digest)=71 AND substr(artifact_digest,1,7)='sha256:' AND substr(artifact_digest,8) NOT GLOB '*[^0-9a-f]*'),
+    package_digest TEXT NOT NULL CHECK(length(package_digest)=71 AND substr(package_digest,1,7)='sha256:' AND substr(package_digest,8) NOT GLOB '*[^0-9a-f]*'),
+    version TEXT NOT NULL,
+    artifact_size INTEGER NOT NULL CHECK(artifact_size > 0),
+    metadata_json TEXT NOT NULL CHECK(json_valid(metadata_json)),
+    revoked INTEGER NOT NULL DEFAULT 0 CHECK(revoked IN (0,1)),
+    revocation_reason TEXT,
+    CHECK((revoked=0 AND revocation_reason IS NULL) OR (revoked=1 AND revocation_reason IS NOT NULL)),
+    PRIMARY KEY(workspace_id, installation_id, artifact_digest),
+    UNIQUE(workspace_id, installation_id, version),
+    FOREIGN KEY(workspace_id, installation_id) REFERENCES plugin_installations(workspace_id, installation_id) ON DELETE CASCADE
+);
+CREATE TABLE plugin_uploads (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    upload_id TEXT NOT NULL,
+    actor TEXT NOT NULL,
+    expires_at INTEGER NOT NULL,
+    artifact_digest TEXT NOT NULL CHECK(length(artifact_digest)=71 AND substr(artifact_digest,1,7)='sha256:' AND substr(artifact_digest,8) NOT GLOB '*[^0-9a-f]*'),
+    artifact_size INTEGER NOT NULL CHECK(artifact_size > 0),
+    metadata_json TEXT NOT NULL CHECK(json_valid(metadata_json)),
+    PRIMARY KEY(workspace_id, upload_id)
+);
+CREATE INDEX plugin_upload_expiry ON plugin_uploads(expires_at);
+CREATE TABLE plugin_operations (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    operation_id TEXT NOT NULL,
+    actor TEXT NOT NULL,
+    kind TEXT NOT NULL CHECK(kind IN ('install','add_artifact','revoke','uninstall')),
+    operation_key TEXT NOT NULL CHECK(length(operation_key) BETWEEN 1 AND 128),
+    fingerprint TEXT NOT NULL CHECK(length(fingerprint)=64),
+    result_json TEXT NOT NULL CHECK(json_valid(result_json)),
+    PRIMARY KEY(workspace_id, operation_id),
+    UNIQUE(workspace_id, actor, kind, operation_key)
+);
+CREATE TABLE plugin_references (
+    workspace_id TEXT NOT NULL,
+    kind TEXT NOT NULL CHECK(kind IN ('config','pending','active','restorable')),
+    owner TEXT NOT NULL CHECK(length(owner) BETWEEN 1 AND 512),
+    installation_id TEXT NOT NULL,
+    artifact_digest TEXT NOT NULL,
+    PRIMARY KEY(workspace_id, kind, owner, installation_id, artifact_digest),
+    FOREIGN KEY(workspace_id, installation_id, artifact_digest) REFERENCES plugin_artifacts(workspace_id, installation_id, artifact_digest) ON DELETE RESTRICT
+);
+CREATE TRIGGER plugin_revocation_irreversible BEFORE UPDATE OF revoked ON plugin_artifacts
+WHEN OLD.revoked=1 AND NEW.revoked<>1 BEGIN SELECT RAISE(ABORT,'plugin revocation is irreversible'); END;
+CREATE TRIGGER plugin_reference_live BEFORE INSERT ON plugin_references
+WHEN NOT EXISTS(SELECT 1 FROM plugin_artifacts a JOIN plugin_installations i USING(workspace_id, installation_id)
+ WHERE a.workspace_id=NEW.workspace_id AND a.installation_id=NEW.installation_id AND a.artifact_digest=NEW.artifact_digest AND a.revoked=0 AND i.uninstalled=0)
+BEGIN SELECT RAISE(ABORT,'plugin reference target unavailable'); END;
+CREATE TRIGGER plugin_uninstall_referenced BEFORE UPDATE OF uninstalled ON plugin_installations
+WHEN NEW.uninstalled=1 AND EXISTS(SELECT 1 FROM plugin_references r WHERE r.workspace_id=NEW.workspace_id AND r.installation_id=NEW.installation_id)
+BEGIN SELECT RAISE(ABORT,'plugin installation referenced'); END;

crates/workspace-server/src/schema_convergence_tests.rs

@@ -100,6 +100,10 @@ mod schema_convergence {
                             version: 105,
                             name: "Durable Subject deletion snapshots and Job admission fences".into(),
                         },
+                        WorkspaceSchemaMigrationStep {
+                            version: 106,
+                            name: crate::plugin_catalog::MIGRATION_NAME.into(),
+                        },
                     ]
                 );
                 assert_eq!(
@@ -116,6 +120,7 @@ mod schema_convergence {
                     let mut expected = history.clone();
                     expected.push((104, "Converge Passkey results and Job input retention".into()));
                     expected.push((105, "Durable Subject deletion snapshots and Job admission fences".into()));
+                    expected.push((106, crate::plugin_catalog::MIGRATION_NAME.into()));
                     assert_eq!(workspace_schema_migration_history(conn)?, expected);
                     for table in ["passkey_credentials", "passkey_metadata", "passkey_addition_receipts", "backend_jobs", "backend_job_attempts"] {
                         assert_eq!(state_contract_schema_signature(conn, table), fresh.with_conn(|c| Ok(state_contract_schema_signature(c, table)))?, "{table}");
@@ -145,7 +150,7 @@ mod schema_convergence {
                         assert_eq!(feature_rows(conn, job_variant), rows);
                         assert_eq!(
                             workspace_schema_migration_history(conn)?.len(),
-                            history.len() + 2
+                            history.len() + plan.migrations.len()
                         );
                         Ok(())
                     })
@@ -183,7 +188,7 @@ mod schema_convergence {
                 upgraded
                     .with_conn(|conn| {
                         assert_eq!(feature_rows(conn, job_variant), rows);
-                        assert_eq!(current_schema_version(conn)?, 105);
+                        assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
                         Ok(())
                     })
                     .unwrap();

crates/workspace-server/src/server.rs

@@ -46,6 +46,8 @@ use worker_protocol::*;
 mod drive;
 #[path = "server_drive_grants.rs"]
 mod drive_grants;
+#[path = "server_plugin.rs"]
+mod plugin;
 #[path = "server_workspace_config.rs"]
 mod workspace_config;
 use std::collections::{BTreeMap, HashMap, HashSet};
@@ -3418,11 +3420,12 @@ async fn authorize_scoped_workspace_request(
         .map_err(|error| {
             runtime_request_proof_rejection(&method, &request_path, workspace_id, &error)
         })?;
-        if is_workspace_config_editor_path(&request_path) {
+        if is_workspace_config_editor_path(&request_path) || plugin::is_catalog_path(&request_path)
+        {
             return Err(repository_api_rejection(
                 &request_path,
                 StatusCode::FORBIDDEN,
-                "Runtime configuration editor access denied",
+                "Runtime user-management access denied",
             ));
         }
         request.extensions_mut().insert(source);
@@ -3576,11 +3579,12 @@ async fn authorize_workspace_api_request(
                 );
             }
         };
-        if is_workspace_config_editor_path(&request_path) {
+        if is_workspace_config_editor_path(&request_path) || plugin::is_catalog_path(&request_path)
+        {
             return repository_api_rejection(
                 &request_path,
                 StatusCode::FORBIDDEN,
-                "Runtime configuration editor access denied",
+                "Runtime user-management access denied",
             );
         }
         request.extensions_mut().insert(source);
@@ -4358,6 +4362,8 @@ impl WorkspaceApi {
         runtime_subscription_broker
             .register_embedded_runtime(embedded_runtime_id, embedded_subscription_runtime);
         let database_path = config.database_path.clone();
+        let plugin_root =
+            ServerConfig::server_data_root_for_data_dir(&config.data_root).join("plugin-artifacts");
         let executor = store.db_executor();
         let config_executor = executor.clone();
         let (config_store, repository_secrets) = executor
@@ -4367,6 +4373,7 @@ impl WorkspaceApi {
                         .open_store(&database_path)?
                         .with_db_executor(config_executor),
                 );
+                config_store.plugin_gc(&plugin_root)?;
                 let repository_secrets = Arc::new(RepositorySecretService::open(
                     config_store.clone(),
                     &database_path,
@@ -7603,6 +7610,7 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
         .merge(server_api::server_api_axum::current_worker_workdir_catalog(
             service.clone(),
         ))
+        .merge(plugin::router(service.clone()))
         .merge(server_api::server_api_axum::current_worker_workspace_config_get(service.clone()))
         .merge(server_api::server_api_axum::current_worker_workspace_config_attach(service.clone()))
         .merge(
@@ -13318,6 +13326,94 @@ impl server_api::ServerApi for ServerApiContractService {
         .map_err(ApiError::into_repository_api_error)
     }
 
+    async fn plugin_upload(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        body: server_api::BinaryBody,
+    ) -> std::result::Result<server_api::PluginUploadResponse, server_api::RepositoryApiError> {
+        let api = self.workspace_api()?;
+        plugin::upload(api, actor, workspace_id, body).await
+    }
+
+    async fn plugin_install(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        request: server_api::PluginInstallRequest,
+    ) -> std::result::Result<server_api::PluginOperationResponse, server_api::RepositoryApiError>
+    {
+        let api = self.workspace_api()?;
+        plugin::install(api, actor, workspace_id, request).await
+    }
+
+    async fn plugin_list(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        query: server_api::PluginListRequest,
+    ) -> std::result::Result<server_api::PluginListResponse, server_api::RepositoryApiError> {
+        let api = self.workspace_api()?;
+        plugin::list(api, actor, workspace_id, query).await
+    }
+
+    async fn plugin_show(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        installation_id: String,
+    ) -> std::result::Result<server_api::PluginInstallation, server_api::RepositoryApiError> {
+        let api = self.workspace_api()?;
+        plugin::show(api, actor, workspace_id, installation_id).await
+    }
+
+    async fn plugin_artifact_add(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        installation_id: String,
+        request: server_api::PluginAddArtifactRequest,
+    ) -> std::result::Result<server_api::PluginOperationResponse, server_api::RepositoryApiError>
+    {
+        let api = self.workspace_api()?;
+        plugin::artifact_add(api, actor, workspace_id, installation_id, request).await
+    }
+
+    async fn plugin_revoke(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        installation_id: String,
+        request: server_api::PluginRevokeRequest,
+    ) -> std::result::Result<server_api::PluginOperationResponse, server_api::RepositoryApiError>
+    {
+        let api = self.workspace_api()?;
+        plugin::revoke(api, actor, workspace_id, installation_id, request).await
+    }
+
+    async fn plugin_uninstall(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        installation_id: String,
+        request: server_api::PluginUninstallRequest,
+    ) -> std::result::Result<server_api::PluginOperationResponse, server_api::RepositoryApiError>
+    {
+        let api = self.workspace_api()?;
+        plugin::uninstall(api, actor, workspace_id, installation_id, request).await
+    }
+
+    async fn plugin_operation(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        operation_id: String,
+    ) -> std::result::Result<server_api::PluginOperationResponse, server_api::RepositoryApiError>
+    {
+        let api = self.workspace_api()?;
+        plugin::operation(api, actor, workspace_id, operation_id).await
+    }
+
     async fn drive_root(
         &self,
         context: server_api::ServerRequestContext,
@@ -38064,6 +38160,10 @@ mod tests {
         include!("server_ticket_case_signed_tests.rs");
     }
 
+    mod plugin_tests {
+        include!("server_plugin_tests.rs");
+    }
+
     mod ticket_tags_tests {
         include!("server_ticket_tags_tests.rs");
     }

crates/workspace-server/src/server_plugin.rs

@@ -0,0 +1,226 @@
+//! Human catalog management is separate from Runtime artifact delivery and execution grants.
+//! The outer dispatcher authenticates each request. Mutations additionally require the
+//! Workspace owner account; authenticated readers follow the existing Workspace read policy.
+//! Neither model text nor a Runtime workspace.request proof can manufacture a human actor.
+use super::*;
+use server_api::*;
+
+type CatalogResult<T> = std::result::Result<T, RepositoryApiError>;
+
+pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
+    Router::new()
+        .merge(
+            server_api_axum::plugin_upload(service.clone()).layer(DefaultBodyLimit::max(
+                manifest::plugin::PluginPackageLimits::default().max_package_size_bytes as usize,
+            )),
+        )
+        .merge(server_api_axum::plugin_install(service.clone()))
+        .merge(server_api_axum::plugin_list(service.clone()))
+        .merge(server_api_axum::plugin_show(service.clone()))
+        .merge(server_api_axum::plugin_artifact_add(service.clone()))
+        .merge(server_api_axum::plugin_revoke(service.clone()))
+        .merge(server_api_axum::plugin_uninstall(service.clone()))
+        .merge(server_api_axum::plugin_operation(service))
+        .layer(middleware::from_fn(normalize_rejections))
+}
+
+async fn normalize_rejections(request: Request, next: Next) -> Response {
+    let response = next.run(request).await;
+    // The management contract classifies malformed DTOs/queries as invalid=400,
+    // including Axum's JSON data/media rejections. Preserve the common envelope
+    // without reflecting attacker-supplied JSON values or transport diagnostics.
+    if matches!(
+        response.status(),
+        StatusCode::BAD_REQUEST
+            | StatusCode::UNPROCESSABLE_ENTITY
+            | StatusCode::UNSUPPORTED_MEDIA_TYPE
+    ) {
+        (
+            StatusCode::BAD_REQUEST,
+            Json(RepositoryApiError::new(
+                400,
+                "plugin_invalid",
+                "Invalid Plugin catalog request",
+                vec![],
+            )),
+        )
+            .into_response()
+    } else {
+        response
+    }
+}
+
+pub(super) fn is_catalog_path(path: &str) -> bool {
+    path.strip_prefix("/api/w/")
+        .and_then(|p| p.split_once('/'))
+        .is_some_and(|(_, resource)| {
+            matches!(resource, "plugins" | "plugin-uploads")
+                || resource.starts_with("plugins/")
+                || resource.starts_with("plugin-operations/")
+        })
+}
+
+fn failure(error: Error) -> RepositoryApiError {
+    let (status, code) = match error {
+        Error::PluginCatalog(error) => (error.status(), error.code()),
+        Error::WorkspacePermissionDenied(_) => (403, "access_denied"),
+        Error::WorkspaceIdMismatch => (404, "not_found"),
+        Error::DbCapacityExceeded => (503, "busy"),
+        // A storage failure may be a lost acknowledgement. Do not assert rollback
+        // or encourage a fresh operation key; the caller can reconcile its receipt.
+        _ => (503, "catalog_outcome_unknown"),
+    };
+    RepositoryApiError::new(status, code, format!("Plugin catalog: {code}"), vec![])
+}
+
+async fn authorize(
+    api: &WorkspaceApi,
+    workspace: &str,
+    actor: &RequestActor,
+    manage: bool,
+) -> CatalogResult<()> {
+    validate_workspace_scope(api, workspace).map_err(|e| failure(e.error))?;
+    if manage {
+        require_workspace_owner(api, workspace, actor, "Plugin management")
+            .await
+            .map_err(|e| failure(e.error))?;
+    }
+    Ok(())
+}
+
+fn actor_identity(actor: &RequestActor) -> String {
+    // Account alone is not a caller: two users of the same management account
+    // must not consume one another's temporary uploads or operation keys.
+    serde_json::to_string(&(&actor.account_id, &actor.user_id))
+        .expect("string tuple is serializable")
+}
+
+pub(super) async fn upload(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    body: BinaryBody,
+) -> CatalogResult<PluginUploadResponse> {
+    authorize(api, &workspace, &actor, true).await?;
+    let root =
+        ServerConfig::server_data_root_for_data_dir(&api.config.data_root).join("plugin-artifacts");
+    let store = api.config_store.clone();
+    let actor = actor_identity(&actor);
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_upload(&root, &workspace, &actor, body.as_ref()))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn install(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    request: PluginInstallRequest,
+) -> CatalogResult<PluginOperationResponse> {
+    authorize(api, &workspace, &actor, true).await?;
+    let store = api.config_store.clone();
+    let actor = actor_identity(&actor);
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_install(&workspace, &actor, &request))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn list(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    query: PluginListRequest,
+) -> CatalogResult<PluginListResponse> {
+    authorize(api, &workspace, &actor, false).await?;
+    let store = api.config_store.clone();
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_list(&workspace, &query))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn show(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    installation_id: String,
+) -> CatalogResult<PluginInstallation> {
+    authorize(api, &workspace, &actor, false).await?;
+    let store = api.config_store.clone();
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_show(&workspace, &installation_id))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn artifact_add(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    installation_id: String,
+    request: PluginAddArtifactRequest,
+) -> CatalogResult<PluginOperationResponse> {
+    authorize(api, &workspace, &actor, true).await?;
+    let store = api.config_store.clone();
+    let actor = actor_identity(&actor);
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_add_artifact(&workspace, &actor, &installation_id, &request))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn revoke(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    installation_id: String,
+    request: PluginRevokeRequest,
+) -> CatalogResult<PluginOperationResponse> {
+    authorize(api, &workspace, &actor, true).await?;
+    let store = api.config_store.clone();
+    let actor = actor_identity(&actor);
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_revoke(&workspace, &actor, &installation_id, &request))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn uninstall(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    installation_id: String,
+    request: PluginUninstallRequest,
+) -> CatalogResult<PluginOperationResponse> {
+    authorize(api, &workspace, &actor, true).await?;
+    let store = api.config_store.clone();
+    let actor = actor_identity(&actor);
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_uninstall(&workspace, &actor, &installation_id, &request))
+        .await
+        .map_err(failure)
+}
+
+pub(super) async fn operation(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    operation_id: String,
+) -> CatalogResult<PluginOperationResponse> {
+    authorize(api, &workspace, &actor, false).await?;
+    let store = api.config_store.clone();
+    let executor = store.db_executor();
+    executor
+        .run(move || store.plugin_operation(&workspace, &operation_id))
+        .await
+        .map_err(failure)
+}

crates/workspace-server/src/server_plugin_tests.rs

@@ -0,0 +1,414 @@
+use super::*;
+
+// Generated routes -> real token resolver and Workspace admission -> real catalog
+// SQLite/blob storage. The unrelated Worker executor is the existing test backend.
+fn package_bytes(version: &str) -> Vec<u8> {
+    use manifest::plugin::*;
+    let dir = tempfile::tempdir().unwrap();
+    fs::write(
+        dir.path().join("plugin.toml"),
+        include_str!("../../../resources/plugin/templates/rust-component-tool/plugin.toml")
+            .replace("0.1.0", version),
+    )
+    .unwrap();
+    fs::write(
+        dir.path().join("plugin.component.wasm"),
+        include_bytes!("../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"),
+    )
+    .unwrap();
+    let limits = PluginPackageLimits::default();
+    pack_plugin_package(
+        &read_plugin_directory(dir.path(), PluginSourceKind::Explicit, &limits).unwrap(),
+        &limits,
+    )
+    .unwrap()
+    .bytes
+}
+
+async fn request(
+    app: Router,
+    token: Option<&str>,
+    method: &str,
+    path: &str,
+    body: Vec<u8>,
+    binary: bool,
+    expected: StatusCode,
+) -> Value {
+    let mut req = Request::builder().method(method).uri(path).header(
+        CONTENT_TYPE,
+        if binary {
+            "application/octet-stream"
+        } else {
+            "application/json"
+        },
+    );
+    if let Some(token) = token {
+        req = req.header("authorization", format!("Bearer {token}"));
+    }
+    let response = app
+        .oneshot(req.body(Body::from(body)).unwrap())
+        .await
+        .unwrap();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), 32 * 1024 * 1024)
+        .await
+        .unwrap();
+    assert_eq!(
+        status,
+        expected,
+        "{method} {path}: {}",
+        String::from_utf8_lossy(&bytes)
+    );
+    serde_json::from_slice(&bytes).unwrap()
+}
+async fn json(app: Router, token: &str, path: &str, body: Value, expected: StatusCode) -> Value {
+    request(
+        app,
+        Some(token),
+        "POST",
+        path,
+        serde_json::to_vec(&body).unwrap(),
+        false,
+        expected,
+    )
+    .await
+}
+async fn get(app: Router, token: Option<&str>, path: &str, expected: StatusCode) -> Value {
+    request(app, token, "GET", path, vec![], false, expected).await
+}
+
+#[tokio::test]
+async fn authenticated_catalog_lifecycle_replay_cas_and_current_read_authority() {
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    let owner = seed_test_workspace_owner_token(api.store.as_ref(), "plugin-owner").await;
+    let reader = seed_test_api_token(api.store.as_ref(), "plugin-reader");
+    let app = build_router(api.clone());
+    let prefix = format!("/api/w/{TEST_WORKSPACE_ID}");
+    let catalog = format!("{prefix}/plugins");
+    let upload_path = format!("{prefix}/plugin-uploads");
+    get(app.clone(), None, &catalog, StatusCode::UNAUTHORIZED).await;
+    get(app.clone(), Some(&reader), &catalog, StatusCode::OK).await;
+    request(
+        app.clone(),
+        Some(&reader),
+        "POST",
+        &upload_path,
+        package_bytes("0.1.0"),
+        true,
+        StatusCode::FORBIDDEN,
+    )
+    .await;
+    let upload = request(
+        app.clone(),
+        Some(&owner),
+        "POST",
+        &upload_path,
+        package_bytes("0.1.0"),
+        true,
+        StatusCode::OK,
+    )
+    .await;
+    assert_ne!(
+        upload["artifact"]["package_digest"],
+        upload["artifact"]["artifact_digest"]
+    );
+    let install = serde_json::json!({"upload_id":upload["upload_id"],"alias":"primary","operation_key":"install"});
+    let original = json(
+        app.clone(),
+        &owner,
+        &catalog,
+        install.clone(),
+        StatusCode::OK,
+    )
+    .await;
+    assert_eq!(
+        json(
+            app.clone(),
+            &owner,
+            &catalog,
+            install.clone(),
+            StatusCode::OK
+        )
+        .await,
+        original
+    );
+    let mut changed = install.clone();
+    changed["alias"] = "other".into();
+    json(app.clone(), &owner, &catalog, changed, StatusCode::CONFLICT).await;
+    let id = original["installation"]["installation_id"]
+        .as_str()
+        .unwrap();
+    let detail_path = format!("{catalog}/{id}");
+    let observed = get(app.clone(), Some(&reader), &detail_path, StatusCode::OK).await;
+    assert_eq!(observed["last_operation"], original["operation_id"]);
+    get(
+        app.clone(),
+        Some(&reader),
+        &format!(
+            "{prefix}/plugin-operations/{}",
+            original["operation_id"].as_str().unwrap()
+        ),
+        StatusCode::OK,
+    )
+    .await;
+    get(
+        app.clone(),
+        Some(&owner),
+        &format!("/api/w/another/plugins/{id}"),
+        StatusCode::NOT_FOUND,
+    )
+    .await;
+    get(
+        app.clone(),
+        Some(&owner),
+        &format!(
+            "{prefix}/plugins/{}",
+            upload["artifact"]["artifact_digest"].as_str().unwrap()
+        ),
+        StatusCode::NOT_FOUND,
+    )
+    .await;
+    let next_upload = request(
+        app.clone(),
+        Some(&owner),
+        "POST",
+        &upload_path,
+        package_bytes("0.2.0"),
+        true,
+        StatusCode::OK,
+    )
+    .await;
+    let add = json(app.clone(), &owner, &format!("{detail_path}/artifacts"), serde_json::json!({"upload_id":next_upload["upload_id"],"expected_last_operation":observed["last_operation"],"operation_key":"add"}), StatusCode::OK).await;
+    json(app.clone(), &owner, &format!("{detail_path}/uninstall"), serde_json::json!({"expected_last_operation":observed["last_operation"],"operation_key":"stale-delete"}), StatusCode::CONFLICT).await;
+    let revoke = json(app.clone(), &owner, &format!("{detail_path}/revoke"), serde_json::json!({"expected_last_operation":add["operation_id"],"operation_key":"revoke","artifact_digest":upload["artifact"]["artifact_digest"],"reason":"withdrawn"}), StatusCode::OK).await;
+    assert!(
+        revoke["installation"]["artifacts"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|a| a["revoked"] == true)
+    );
+    let removed = json(app.clone(), &owner, &format!("{detail_path}/uninstall"), serde_json::json!({"expected_last_operation":revoke["operation_id"],"operation_key":"remove"}), StatusCode::OK).await;
+    assert_eq!(removed["installation"]["uninstalled"], true);
+    // Historical receipts remain historical after catalog movement.
+    assert_eq!(
+        json(
+            app.clone(),
+            &owner,
+            &catalog,
+            install.clone(),
+            StatusCode::OK
+        )
+        .await,
+        original
+    );
+    api.config_store
+        .with_conn(|conn| {
+            conn.execute(
+                "UPDATE api_tokens SET revoked_at='2026-01-01' WHERE token_hash=?1",
+                [crate::auth::token_hash(&owner)],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+    json(
+        app.clone(),
+        &owner,
+        &catalog,
+        install,
+        StatusCode::UNAUTHORIZED,
+    )
+    .await;
+    shutdown_workspace_feature_storage(&api).await.unwrap();
+}
+
+#[tokio::test]
+async fn catalog_rejects_invalid_uploads_self_reported_roles_and_runtime_proofs() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut api = test_api(dir.path()).await;
+    let token = seed_test_workspace_owner_token(api.store.as_ref(), "plugin-input").await;
+    let identity = RuntimeIdentityMaterial::generate("runtime-plugin-input").unwrap();
+    configure_runtime_request_auth(&mut api, &identity, "runtime-plugin-input");
+    let app = build_router(api.clone());
+    let prefix = format!("/api/w/{TEST_WORKSPACE_ID}");
+    request(
+        app.clone(),
+        Some(&token),
+        "POST",
+        &format!("{prefix}/plugin-uploads"),
+        b"partial zip".to_vec(),
+        true,
+        StatusCode::BAD_REQUEST,
+    )
+    .await;
+    let max = manifest::plugin::PluginPackageLimits::default().max_package_size_bytes as usize;
+    request(
+        app.clone(),
+        Some(&token),
+        "POST",
+        &format!("{prefix}/plugin-uploads"),
+        vec![0; max + 1],
+        true,
+        StatusCode::PAYLOAD_TOO_LARGE,
+    )
+    .await;
+    json(app.clone(), &token, &format!("{prefix}/plugins"), serde_json::json!({"upload_id":"missing","alias":"bad","operation_key":"bad","role":"owner"}), StatusCode::BAD_REQUEST).await;
+    let response = app
+        .clone()
+        .oneshot(runtime_source_request(
+            &identity,
+            None,
+            "GET",
+            &format!("{prefix}/plugins"),
+            vec![],
+        ))
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    assert_eq!(
+        get(
+            app,
+            Some(&token),
+            &format!("{prefix}/plugins"),
+            StatusCode::OK
+        )
+        .await["installations"],
+        serde_json::json!([])
+    );
+    shutdown_workspace_feature_storage(&api).await.unwrap();
+}
+
+#[tokio::test]
+async fn generated_client_and_interrupted_binary_transport_use_real_catalog_authority() {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    let token = seed_test_workspace_owner_token(api.store.as_ref(), "plugin-transport").await;
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let app = build_router(api.clone());
+    let task = tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    let mut stream = tokio::net::TcpStream::connect(address).await.unwrap();
+    let header = format!(
+        "POST /api/w/{TEST_WORKSPACE_ID}/plugin-uploads HTTP/1.1\r\nHost: {address}\r\nAuthorization: Bearer {token}\r\nContent-Type: application/octet-stream\r\nContent-Length: 2000\r\nConnection: close\r\n\r\npartial"
+    );
+    stream.write_all(header.as_bytes()).await.unwrap();
+    stream.shutdown().await.unwrap();
+    let mut response = Vec::new();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        stream.read_to_end(&mut response),
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    assert_eq!(
+        api.config_store
+            .with_conn(|conn| Ok(conn.query_row(
+                "SELECT count(*) FROM plugin_uploads",
+                [],
+                |r| r.get::<_, u64>(0)
+            )?))
+            .unwrap(),
+        0
+    );
+    let mut headers = reqwest::header::HeaderMap::new();
+    headers.insert(
+        reqwest::header::AUTHORIZATION,
+        format!("Bearer {token}").parse().unwrap(),
+    );
+    let client = server_api::ServerApiClient::builder(format!("http://{address}"))
+        .unwrap()
+        .client(
+            reqwest::Client::builder()
+                .default_headers(headers)
+                .redirect(reqwest::redirect::Policy::none())
+                .build()
+                .unwrap(),
+        )
+        .build()
+        .unwrap();
+    let upload = client
+        .plugin_upload(TEST_WORKSPACE_ID.into(), package_bytes("0.1.0").into())
+        .await
+        .unwrap();
+    let installed = client
+        .plugin_install(
+            TEST_WORKSPACE_ID.into(),
+            server_api::PluginInstallRequest {
+                upload_id: upload.upload_id,
+                alias: "transport".into(),
+                operation_key: "transport-install".into(),
+            },
+        )
+        .await
+        .unwrap();
+    let observed = client
+        .plugin_show(
+            TEST_WORKSPACE_ID.into(),
+            installed.installation.installation_id,
+        )
+        .await
+        .unwrap();
+    assert_eq!(observed.last_operation, installed.operation_id);
+    assert_eq!(observed.artifacts[0], upload.artifact);
+    task.abort();
+    let _ = task.await;
+    shutdown_workspace_feature_storage(&api).await.unwrap();
+}
+
+#[tokio::test]
+async fn api_restart_recovers_orphan_publication_and_preserves_committed_catalog() {
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    let token = seed_test_workspace_owner_token(api.store.as_ref(), "plugin-restart").await;
+    let app = build_router(api.clone());
+    let prefix = format!("/api/w/{TEST_WORKSPACE_ID}");
+    let upload = request(
+        app.clone(),
+        Some(&token),
+        "POST",
+        &format!("{prefix}/plugin-uploads"),
+        package_bytes("0.1.0"),
+        true,
+        StatusCode::OK,
+    )
+    .await;
+    let installed = json(app.clone(),&token,&format!("{prefix}/plugins"),serde_json::json!({"upload_id":upload["upload_id"],"alias":"restart","operation_key":"restart-install"}),StatusCode::OK).await;
+    let root = ServerConfig::server_data_root_for_data_dir(&api.config.data_root)
+        .join("plugin-artifacts/blobs");
+    let orphan = root.join("f".repeat(64));
+    let partial = root.join(".pending-interrupted-restart");
+    fs::write(&orphan, b"published without committed reference").unwrap();
+    fs::write(&partial, b"partial").unwrap();
+    drop(app);
+    shutdown_workspace_feature_storage(&api).await.unwrap();
+    drop(api);
+    let reopened = test_api(dir.path()).await;
+    assert!(!orphan.exists());
+    assert!(!partial.exists());
+    let observed = reopened
+        .config_store
+        .plugin_show(
+            TEST_WORKSPACE_ID,
+            installed["installation"]["installation_id"]
+                .as_str()
+                .unwrap(),
+        )
+        .unwrap();
+    assert_eq!(observed.last_operation, installed["operation_id"]);
+    let bytes = reopened
+        .config_store
+        .plugin_artifact_bytes(
+            root.parent().unwrap(),
+            TEST_WORKSPACE_ID,
+            &observed.installation_id,
+            &observed.artifacts[0].artifact_digest,
+        )
+        .unwrap();
+    assert_eq!(bytes, package_bytes("0.1.0"));
+    shutdown_workspace_feature_storage(&reopened).await.unwrap();
+}

crates/workspace-server/src/store.rs

@@ -127,7 +127,7 @@ pub use ticket::{
 };
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 105;
+const LATEST_SCHEMA_VERSION: i64 = 106;
 const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
 const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
 const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
@@ -460,6 +460,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "Durable Subject deletion snapshots and Job admission fences",
         apply: subject_deletion_store::migrate,
     },
+    Migration {
+        version: 106,
+        name: crate::plugin_catalog::MIGRATION_NAME,
+        apply: crate::plugin_catalog::migrate,
+    },
 ];
 
 // Both schema-103 identities shipped. Keep the alternate implementation and accept
@@ -1334,6 +1339,64 @@ pub struct HostingAdminInitialBearer {
 
 #[async_trait]
 pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
+    fn plugin_upload(
+        &self,
+        root: &Path,
+        workspace: &str,
+        actor: &str,
+        bytes: &[u8],
+    ) -> Result<server_api::PluginUploadResponse>;
+    fn plugin_install(
+        &self,
+        workspace: &str,
+        actor: &str,
+        request: &server_api::PluginInstallRequest,
+    ) -> Result<server_api::PluginOperationResponse>;
+    fn plugin_add_artifact(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &server_api::PluginAddArtifactRequest,
+    ) -> Result<server_api::PluginOperationResponse>;
+    fn plugin_revoke(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &server_api::PluginRevokeRequest,
+    ) -> Result<server_api::PluginOperationResponse>;
+    fn plugin_uninstall(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &server_api::PluginUninstallRequest,
+    ) -> Result<server_api::PluginOperationResponse>;
+    fn plugin_show(
+        &self,
+        workspace: &str,
+        installation: &str,
+    ) -> Result<server_api::PluginInstallation>;
+    fn plugin_list(
+        &self,
+        workspace: &str,
+        request: &server_api::PluginListRequest,
+    ) -> Result<server_api::PluginListResponse>;
+    fn plugin_operation(
+        &self,
+        workspace: &str,
+        operation: &str,
+    ) -> Result<server_api::PluginOperationResponse>;
+    fn plugin_artifact_bytes(
+        &self,
+        root: &Path,
+        workspace: &str,
+        installation: &str,
+        artifact_digest: &str,
+    ) -> Result<Vec<u8>>;
+    fn plugin_gc(&self, root: &Path) -> Result<u64>;
+
     /// Bounded owner for whole synchronous operations from async callers. Move owned inputs
     /// and services into one closure covering connection/mutex waits, source authorization,
     /// receipts, CAS, audit, commit and ordered publication. Do not split checks around awaits
@@ -4670,6 +4733,90 @@ fn insert_workspace_runtime_binding_audit(
 
 #[async_trait]
 impl ControlPlaneStore for SqliteWorkspaceStore {
+    fn plugin_upload(
+        &self,
+        root: &Path,
+        workspace: &str,
+        actor: &str,
+        bytes: &[u8],
+    ) -> Result<server_api::PluginUploadResponse> {
+        SqliteWorkspaceStore::plugin_upload(self, root, workspace, actor, bytes)
+    }
+    fn plugin_install(
+        &self,
+        workspace: &str,
+        actor: &str,
+        request: &server_api::PluginInstallRequest,
+    ) -> Result<server_api::PluginOperationResponse> {
+        SqliteWorkspaceStore::plugin_install(self, workspace, actor, request)
+    }
+    fn plugin_add_artifact(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &server_api::PluginAddArtifactRequest,
+    ) -> Result<server_api::PluginOperationResponse> {
+        SqliteWorkspaceStore::plugin_add_artifact(self, workspace, actor, installation, request)
+    }
+    fn plugin_revoke(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &server_api::PluginRevokeRequest,
+    ) -> Result<server_api::PluginOperationResponse> {
+        SqliteWorkspaceStore::plugin_revoke(self, workspace, actor, installation, request)
+    }
+    fn plugin_uninstall(
+        &self,
+        workspace: &str,
+        actor: &str,
+        installation: &str,
+        request: &server_api::PluginUninstallRequest,
+    ) -> Result<server_api::PluginOperationResponse> {
+        SqliteWorkspaceStore::plugin_uninstall(self, workspace, actor, installation, request)
+    }
+    fn plugin_show(
+        &self,
+        workspace: &str,
+        installation: &str,
+    ) -> Result<server_api::PluginInstallation> {
+        SqliteWorkspaceStore::plugin_show(self, workspace, installation)
+    }
+    fn plugin_list(
+        &self,
+        workspace: &str,
+        request: &server_api::PluginListRequest,
+    ) -> Result<server_api::PluginListResponse> {
+        SqliteWorkspaceStore::plugin_list(self, workspace, request)
+    }
+    fn plugin_operation(
+        &self,
+        workspace: &str,
+        operation: &str,
+    ) -> Result<server_api::PluginOperationResponse> {
+        SqliteWorkspaceStore::plugin_operation(self, workspace, operation)
+    }
+    fn plugin_artifact_bytes(
+        &self,
+        root: &Path,
+        workspace: &str,
+        installation: &str,
+        artifact_digest: &str,
+    ) -> Result<Vec<u8>> {
+        SqliteWorkspaceStore::plugin_artifact_bytes(
+            self,
+            root,
+            workspace,
+            installation,
+            artifact_digest,
+        )
+    }
+    fn plugin_gc(&self, root: &Path) -> Result<u64> {
+        SqliteWorkspaceStore::plugin_gc(self, root)
+    }
+
     fn db_executor(&self) -> crate::db_executor::DbExecutor {
         self.db_executor.clone()
     }
@@ -15639,6 +15786,10 @@ fn create_workspace_schema_structure_from_sql(
     conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?;
     conn.execute_batch(
         &schema_sql
+            .replace(
+                "-- Plugin catalog schema is composed by store.rs.",
+                include_str!("plugin_catalog_schema.sql"),
+            )
             .replace(
                 "-- Ticket-owned assignment schema is composed by store.rs.",
                 ticket::TICKET_ASSIGNMENT_SCHEMA_SQL,
@@ -16648,6 +16799,7 @@ fn apply_migrations_with_secret_source(
     }
 
     verify_schema_history(conn, LATEST_SCHEMA_VERSION)?;
+    crate::plugin_catalog::verify(conn)?;
     verify_workspace_runtime_binding_schema(conn)?;
     verify_workspace_runtime_verification_schema(conn)?;
     verify_workspace_deletion_schema(conn)?;
@@ -19092,7 +19244,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
         fixture.commit().unwrap();
     }
 
-    fn prepare_schema_v50(path: &Path, workspace_id: Option<&str>) {
+    pub(super) fn prepare_schema_v50(path: &Path, workspace_id: Option<&str>) {
         prepare_schema_v52(path);
         let conn = Connection::open(path).unwrap();
         configure_sqlite(&conn).unwrap();
@@ -19341,6 +19493,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                 WorkspaceSchemaMigrationStep {version:103, name:"Passkey metadata and atomic addition results".into()},
                 WorkspaceSchemaMigrationStep {version:104, name:"Converge Passkey results and Job input retention".into()},
                 WorkspaceSchemaMigrationStep {version:105, name:"Durable Subject deletion snapshots and Job admission fences".into()},
+                WorkspaceSchemaMigrationStep {version:106, name:crate::plugin_catalog::MIGRATION_NAME.into()},
             ]
         );
 
@@ -19456,6 +19609,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
 (103, "Passkey metadata and atomic addition results".to_string()),
 (104, "Converge Passkey results and Job input retention".to_string()),
 (105, "Durable Subject deletion snapshots and Job admission fences".to_string()),
+(106, crate::plugin_catalog::MIGRATION_NAME.to_string()),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -22742,7 +22896,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                          (102, "Durable failed Subject creation recovery".into()),
                          (103, "Passkey metadata and atomic addition results".into()),
                          (104, "Converge Passkey results and Job input retention".into()),
-                         (105, "Durable Subject deletion snapshots and Job admission fences".into())]);
+                         (105, "Durable Subject deletion snapshots and Job admission fences".into()),
+                         (106, crate::plugin_catalog::MIGRATION_NAME.into())]);
                 for (table, rows) in unchanged.iter().zip(before.iter()) {
                     assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
                 }
@@ -26936,6 +27091,7 @@ INSERT INTO worker_registry (
             .unwrap();
         store.with_conn(|conn| {
             // This fixture deliberately reconstructs schema 102 from fresh.
+            conn.execute_batch("DROP TABLE plugin_references; DROP TABLE plugin_artifacts; DROP TABLE plugin_installations; DROP TABLE plugin_uploads; DROP TABLE plugin_operations;")?;
             // Remove the newer Subject fence composition, not just its receipt.
             let triggers=conn.prepare("SELECT name,tbl_name FROM sqlite_schema WHERE type='trigger' AND tbl_name LIKE 'notification_%' AND name NOT LIKE 'subject_deleted_%' AND sql LIKE '%subject_deletion_jobs%'")?
                 .query_map([],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?)))?.collect::<rusqlite::Result<Vec<_>>>()?;
@@ -26947,7 +27103,7 @@ INSERT INTO worker_registry (
                 DROP TRIGGER passkey_metadata_insert;
                 DROP TABLE passkey_addition_receipts;
                 DROP TABLE passkey_metadata;
-                UPDATE __yoi_schema_migrations SET version=102 WHERE version=105;
+                UPDATE __yoi_schema_migrations SET version=102 WHERE version=106;
                 CREATE TRIGGER backend_job_input_retirement BEFORE UPDATE ON backend_job_attempts BEGIN SELECT RAISE(ABORT,'collision'); END;")?; Ok(())
         }).unwrap();
         drop(store);
@@ -28259,3 +28415,7 @@ mod repository_key_migration_tests;
 #[cfg(test)]
 #[path = "frozen_runtime_binding_tests.rs"]
 mod frozen_runtime_binding_tests;
+
+#[cfg(test)]
+#[path = "plugin_catalog/migration_tests.rs"]
+mod plugin_catalog_migration_tests;

crates/workspace-server/src/workspace_deletion.rs

@@ -16,6 +16,11 @@ use crate::{Error, Result};
 /// Explicit domain-owned purge inventory. The deletion operation tombstone is intentionally
 /// excluded so retries and audit remain available after the Workspace row is gone.
 const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
+    "plugin_references",
+    "plugin_operations",
+    "plugin_uploads",
+    "plugin_artifacts",
+    "plugin_installations",
     "artifacts",
     "audit_events",
     "backend_job_deliveries",
@@ -1283,6 +1288,84 @@ mod tests {
         }
     }
 
+    #[test]
+    fn workspace_purge_removes_plugin_references_and_catalog_without_cross_workspace_loss() {
+        let (store, owner, workspace_id) = setup();
+        for workspace in ["workspace-a", "workspace-b"] {
+            store.with_conn(|conn| {
+                let digest = format!("sha256:{}", "a".repeat(64));
+                conn.execute("INSERT INTO plugin_installations VALUES(?1,'installation','alias','package','operation',0)", [workspace])?;
+                conn.execute("INSERT INTO plugin_artifacts VALUES(?1,'installation',?2,?2,'1',1,'{}',0,NULL)", params![workspace,digest])?;
+                conn.execute("INSERT INTO plugin_uploads VALUES(?1,'upload','actor',9999999999,?2,1,'{}')", params![workspace,digest])?;
+                conn.execute("INSERT INTO plugin_operations VALUES(?1,'operation','actor','install','key',?2,'{}')", params![workspace,"f".repeat(64)])?;
+                for kind in ["config","pending","active","restorable"] {
+                    conn.execute("INSERT INTO plugin_references VALUES(?1,?2,'owner','installation',?3)", params![workspace,kind,digest])?;
+                }
+                Ok(())
+            }).unwrap();
+        }
+        let counts = |workspace: &str| {
+            store
+                .with_conn(|conn| {
+                    [
+                        "plugin_references",
+                        "plugin_operations",
+                        "plugin_uploads",
+                        "plugin_artifacts",
+                        "plugin_installations",
+                    ]
+                    .map(|table| {
+                        Ok(conn.query_row(
+                            &format!("SELECT count(*) FROM {table} WHERE workspace_id=?1"),
+                            [workspace],
+                            |r| r.get::<_, i64>(0),
+                        )?)
+                    })
+                    .into_iter()
+                    .collect::<Result<Vec<_>>>()
+                })
+                .unwrap()
+        };
+        let other = counts("workspace-b");
+        assert_eq!(counts("workspace-a"), vec![4, 1, 1, 1, 1]);
+        let preflight = store
+            .workspace_deletion_preflight(&owner, &workspace_id)
+            .unwrap();
+        let operation = store
+            .reserve_workspace_deletion(
+                &owner,
+                &workspace_id,
+                &WorkspaceDeletionRequest {
+                    operation_id: "purge-plugins".into(),
+                    expected_workspace_updated_at: preflight.expected_workspace_updated_at,
+                    confirmation: "Alpha".into(),
+                },
+            )
+            .unwrap()
+            .operation;
+        assert_eq!(
+            store
+                .finalize_workspace_deletion(&operation.operation_id)
+                .unwrap()
+                .state,
+            WorkspaceDeletionState::Succeeded
+        );
+        assert_eq!(counts("workspace-a"), vec![0; 5]);
+        assert_eq!(counts("workspace-b"), other);
+        store
+            .with_conn(|conn| {
+                assert_eq!(
+                    conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| r
+                        .get::<_, i64>(
+                        0
+                    ))?,
+                    0
+                );
+                Ok(())
+            })
+            .unwrap();
+    }
+
     #[test]
     fn workspace_purge_removes_subject_deletion_snapshots_and_preserves_other_workspace() {
         let (store, owner, workspace_id) = setup();