feat: integrate authorized Plugin delivery and verified cache

crates/workspace-server/Cargo.toml

@@ -26,6 +26,7 @@ workspace-drive.workspace = true
 subjektiv.workspace = true
 flow = { path = "../flow" }
 manifest.workspace = true
+yoi-plugin-runtime = { path = "../plugin-runtime" }
 job.workspace = true
 protocol = { workspace = true }
 runtime-api.workspace = true

crates/workspace-server/src/lib.rs

@@ -14,6 +14,7 @@ pub mod memory_staging;
 pub mod observation;
 pub mod plugin_binding;
 pub mod plugin_catalog;
+pub mod plugin_delivery;
 pub mod plugin_plan;
 pub mod plugin_settings;
 pub mod profile_settings;

crates/workspace-server/src/plugin_catalog.rs

@@ -170,7 +170,7 @@ impl Drop for PendingFile {
         let _ = fs::remove_file(&self.0);
     }
 }
-fn read_blob(root: &Path, artifact: &str, size: u64) -> Result<Vec<u8>> {
+pub(crate) 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()
@@ -222,7 +222,7 @@ fn artifact_metadata(
     canonical: &plugin::CanonicalPluginArtifact,
 ) -> Result<PluginArtifact> {
     let m = &package.package.manifest;
-    let schema_digest = format!("sha256:{}", hash(m.config_schema.as_bytes()));
+    let schema = plugin::plugin_config_schema_identity(m);
     Ok(PluginArtifact {
         package_id: m.id.clone(),
         name: m.name.clone(),
@@ -250,10 +250,10 @@ fn artifact_metadata(
         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,
+            identity: schema.identity,
+            version: schema.version,
+            namespace: schema.namespace,
+            digest: schema.digest,
             dcdl: m.config_schema.clone(),
         },
         revoked: false,

crates/workspace-server/src/plugin_delivery.rs

@@ -0,0 +1,146 @@
+//! Delivery admission is bound to an authenticated Host identity and a persisted
+//! exact plan pin, not management permission or possession of a digest. Service
+//! adapters must derive the identity and recheck live Host authority on every
+//! call, including cache hits; these database checks are additional constraints.
+//!
+//! Reads never change plan retention. Revoked and released plans remain audit
+//! evidence, but cannot authorize metadata or bytes. Bytes are read while holding
+//! the catalog's SQLite write fence, preventing revoke/uninstall/GC interleaving.
+use crate::{
+    Result,
+    plugin_catalog::{self, PluginCatalogError},
+    plugin_plan::{self, HostExecutionIdentity, PlanPinState},
+    store::SqliteWorkspaceStore,
+};
+use rusqlite::{Transaction, TransactionBehavior, params};
+use server_api::{PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest};
+use std::path::Path;
+
+/// Promote fresh descriptor metadata to a cache expectation only after binding
+/// every immutable identity to the saved Host pin. This does not admit use; the
+/// cache must still ask the current owner on each resolution.
+pub fn artifact_expectation(
+    pin: &crate::plugin_plan::WorkerPluginPin,
+    artifact: &server_api::PluginArtifact,
+) -> Result<yoi_plugin_runtime::delivery::ArtifactExpectation> {
+    if !pin.matches_artifact(artifact)? {
+        return Err(PluginCatalogError::Conflict.into());
+    }
+    Ok(yoi_plugin_runtime::delivery::ArtifactExpectation::new(
+        artifact.clone(),
+    ))
+}
+
+/// The caller must hold an immediate transaction in the plan/catalog owner DB.
+/// The current catalog operation is read inside this snapshot: unrelated version
+/// additions must not invalidate the immutable pin's creation-time observation.
+/// Current config is deliberately not resolved again, avoiding latest selection.
+pub fn plugin_delivery_authorize(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    execution: &HostExecutionIdentity,
+    request: &PluginArtifactDeliveryRequest,
+) -> Result<PluginArtifactDeliveryMetadata> {
+    let saved = plugin_plan::plugin_plan_load(tx, workspace, &request.plan_id)?;
+    if saved.fingerprint != request.plan_fingerprint || saved.plan.execution() != execution {
+        return Err(PluginCatalogError::Conflict.into());
+    }
+    let kind = match saved.pin_state {
+        PlanPinState::Pending => "pending",
+        PlanPinState::Active => "active",
+        PlanPinState::Restorable => "restorable",
+        PlanPinState::Released => return Err(PluginCatalogError::Conflict.into()),
+    };
+    let pin = saved
+        .plan
+        .plugins()
+        .iter()
+        .find(|pin| {
+            pin.installation_id() == request.installation_id
+                && pin.artifact_digest() == request.artifact_digest
+        })
+        .ok_or(PluginCatalogError::NotFound)?;
+    let retained: bool = tx.query_row(
+        "SELECT EXISTS(SELECT 1 FROM plugin_references WHERE workspace_id=?1 AND kind=?2 AND owner=?3 AND installation_id=?4 AND artifact_digest=?5)",
+        params![workspace, kind, request.plan_id, request.installation_id, request.artifact_digest],
+        |row| row.get(0),
+    )?;
+    if !retained {
+        return Err(PluginCatalogError::Conflict.into());
+    }
+    let operation: String = tx.query_row(
+        "SELECT last_operation FROM plugin_installations WHERE workspace_id=?1 AND installation_id=?2",
+        params![workspace, request.installation_id],
+        |row| row.get(0),
+    )?;
+    let artifact = plugin_catalog::plugin_artifact_admit(
+        tx,
+        workspace,
+        &request.installation_id,
+        &operation,
+        pin.package_digest(),
+        &request.artifact_digest,
+    )?;
+    // Reconstruct using the *saved* projection and operation observation. This
+    // compares all immutable identities, size, Component world/imports and schema
+    // identity while retaining the original config/grants rather than live config.
+    if !pin.matches_artifact(&artifact)? {
+        return Err(PluginCatalogError::Conflict.into());
+    }
+    Ok(PluginArtifactDeliveryMetadata {
+        workspace_id: workspace.into(),
+        request: request.clone(),
+        artifact,
+    })
+}
+
+/// Reauthorizes in the same owner transaction as the bounded immutable blob read.
+/// A previous metadata response (including a cache descriptor) is not authority.
+pub fn plugin_delivery_bytes(
+    tx: &Transaction<'_>,
+    root: &Path,
+    workspace: &str,
+    execution: &HostExecutionIdentity,
+    request: &PluginArtifactDeliveryRequest,
+) -> Result<Vec<u8>> {
+    let admitted = plugin_delivery_authorize(tx, workspace, execution, request)?;
+    plugin_catalog::read_blob(
+        root,
+        &admitted.artifact.artifact_digest,
+        u64::from(admitted.artifact.artifact_size),
+    )
+}
+
+impl SqliteWorkspaceStore {
+    pub fn plugin_delivery_authorize(
+        &self,
+        workspace: &str,
+        execution: &HostExecutionIdentity,
+        request: &PluginArtifactDeliveryRequest,
+    ) -> Result<PluginArtifactDeliveryMetadata> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let metadata = plugin_delivery_authorize(&tx, workspace, execution, request)?;
+            tx.commit()?;
+            Ok(metadata)
+        })
+    }
+
+    pub fn plugin_delivery_bytes(
+        &self,
+        root: &Path,
+        workspace: &str,
+        execution: &HostExecutionIdentity,
+        request: &PluginArtifactDeliveryRequest,
+    ) -> Result<Vec<u8>> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let bytes = plugin_delivery_bytes(&tx, root, workspace, execution, request)?;
+            tx.commit()?;
+            Ok(bytes)
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests;

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

@@ -0,0 +1,591 @@
+//! Owner-DB integration: production upload/install, schema-checked config commit,
+//! plan resolution/creation and file-backed SQLite. Host authentication/transport
+//! are outside this suite; the supplied Host identity is the adapter's boundary.
+use super::*;
+use crate::{
+    Error,
+    plugin_plan::{PluginPlanCreateReceipt, WorkerPluginExecutionPlan},
+    store::{ControlPlaneStore, WorkspaceRecord},
+};
+use manifest::plugin;
+use rusqlite::Connection;
+use server_api::*;
+use std::fs;
+use tempfile::{TempDir, tempdir};
+
+struct Fixture {
+    dir: TempDir,
+    store: SqliteWorkspaceStore,
+}
+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-28T00:00:00Z".into(),
+                    updated_at: "2026-10-28T00:00:00Z".into(),
+                })
+                .await
+                .unwrap();
+        }
+        Self { dir, store }
+    }
+    fn root(&self) -> std::path::PathBuf {
+        self.dir.path().join("artifacts")
+    }
+    fn install(&self, alias: &str) -> PluginOperationResponse {
+        let upload = self
+            .store
+            .plugin_upload(&self.root(), "a", "actor", &package("0.1.0"))
+            .unwrap();
+        self.store
+            .plugin_install(
+                "a",
+                "actor",
+                &PluginInstallRequest {
+                    upload_id: upload.upload_id,
+                    alias: alias.into(),
+                    operation_key: alias.into(),
+                },
+            )
+            .unwrap()
+    }
+    fn create(&self, installed: &PluginOperationResponse, key: &str) -> PluginPlanCreateReceipt {
+        let plan = self.configure(&[installed]);
+        self.store.plugin_plan_create("actor", key, &plan).unwrap()
+    }
+    fn authorize(
+        &self,
+        request: &PluginArtifactDeliveryRequest,
+    ) -> Result<PluginArtifactDeliveryMetadata> {
+        self.store
+            .plugin_delivery_authorize("a", &execution(), request)
+    }
+    fn bytes(&self, request: &PluginArtifactDeliveryRequest) -> Result<Vec<u8>> {
+        self.store
+            .plugin_delivery_bytes(&self.root(), "a", &execution(), request)
+    }
+    fn retained(&self, plan_id: &str) -> Vec<String> {
+        self.store.with_conn(|conn| Ok(conn.prepare("SELECT kind FROM plugin_references WHERE workspace_id='a' AND owner=?1 ORDER BY kind")?
+            .query_map([plan_id], |r| r.get(0))?.collect::<rusqlite::Result<_>>()?)).unwrap()
+    }
+    fn revoke(&self, installed: &PluginOperationResponse) {
+        self.store
+            .plugin_revoke(
+                "a",
+                "actor",
+                &installed.installation.installation_id,
+                &PluginRevokeRequest {
+                    expected_last_operation: installed.operation_id.clone(),
+                    operation_key: "revoke".into(),
+                    artifact_digest: installed.installation.artifacts[0].artifact_digest.clone(),
+                    reason: "owner revoked".into(),
+                },
+            )
+            .unwrap();
+    }
+}
+fn execution() -> HostExecutionIdentity {
+    HostExecutionIdentity {
+        runtime_id: "runtime".into(),
+        worker_id: "worker".into(),
+        principal_id: "worker-principal".into(),
+        delegation_id: Some("delegation".into()),
+    }
+}
+fn request(saved: &PluginPlanCreateReceipt) -> PluginArtifactDeliveryRequest {
+    let pin = &saved.plan.plugins()[0];
+    PluginArtifactDeliveryRequest {
+        plan_id: saved.plan_id.clone(),
+        plan_fingerprint: saved.fingerprint.clone(),
+        installation_id: pin.installation_id().into(),
+        artifact_digest: pin.artifact_digest().into(),
+    }
+}
+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 package(version: &str) -> Vec<u8> {
+    let dir = tempdir().unwrap();
+    fs::write(
+        dir.path().join("plugin.toml"),
+        include_str!("../../../../resources/plugin/templates/rust-component-tool/plugin.toml")
+            .replace("version = \"0.1.0\"", &format!("version = \"{version}\"")),
+    )
+    .unwrap();
+    fs::write(
+        dir.path().join("plugin.component.wasm"),
+        include_bytes!("../../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"),
+    )
+    .unwrap();
+    let package = plugin::read_plugin_directory(
+        dir.path(),
+        plugin::PluginSourceKind::Explicit,
+        &plugin::PluginPackageLimits::default(),
+    )
+    .unwrap();
+    plugin::pack_plugin_package(&package, &plugin::PluginPackageLimits::default())
+        .unwrap()
+        .bytes
+}
+impl Fixture {
+    fn configure(&self, installed: &[&PluginOperationResponse]) -> WorkerPluginExecutionPlan {
+        use crate::config_source::{ConfigCommitRequest, WorkspaceConfigSchemaProvider};
+        use config_source::{ConfigTreeChange, VirtualPath, WorkspaceConfigSchemaBundle};
+        let bundle = WorkspaceConfigSchemaBundle::compose([
+            crate::plugin_settings::PluginConfigSchemaProvider
+                .contribution()
+                .unwrap(),
+            crate::profile_settings::ProfileConfigSchemaProvider
+                .contribution()
+                .unwrap(),
+        ])
+        .unwrap();
+        let state = self
+            .store
+            .ensure_workspace_config_materialized_with_schema("a", "1", bundle.clone())
+            .unwrap();
+        let mut enabled = serde_json::Map::new();
+        let mut selections = Vec::new();
+        for (index, item) in installed.iter().enumerate() {
+            let a = &item.installation.artifacts[0];
+            enabled.insert(
+                format!("plugin_{index}"),
+                serde_json::json!({
+                    "installation_id": item.installation.installation_id,
+                    "expected_last_operation": item.operation_id,
+                    "package_digest": a.package_digest, "artifact_digest": a.artifact_digest,
+                    "tools": ["example_echo"], "config": {},
+                }),
+            );
+            selections.push(serde_json::json!({"installation_id":item.installation.installation_id,"tools":["example_echo"]}));
+        }
+        let value = serde_json::json!({"plugins":{"enabled": enabled}, "profile":{"entries":[{"selector":"project:tool","profile":{"plugins":{"selections":selections}}}]}});
+        let path = VirtualPath::parse("main.dcdl").unwrap();
+        let candidate = self
+            .store
+            .evaluate_workspace_config_candidate_with_schema(
+                "a",
+                &ConfigCommitRequest {
+                    base_digest: state.snapshot.digest.clone(),
+                    entrypoints: vec![path.clone()],
+                    changes: vec![ConfigTreeChange::Update {
+                        path: path.clone(),
+                        expected_digest: state.snapshot.get(&path).unwrap().content_digest.clone(),
+                        content: format!(
+                            "{} as WorkspaceConfigSchema",
+                            crate::plugin_settings::render_value(&value).unwrap()
+                        ),
+                    }],
+                },
+                bundle,
+            )
+            .unwrap();
+        let state = self
+            .store
+            .commit_evaluated_workspace_config("a", &candidate)
+            .unwrap();
+        self.store
+            .with_conn(|conn| {
+                let tx = conn.unchecked_transaction()?;
+                crate::plugin_settings::resolve_worker_plugin_plan(
+                    &tx,
+                    "a",
+                    &state.snapshot.digest,
+                    "project:tool",
+                    execution(),
+                )
+            })
+            .unwrap()
+    }
+    fn disable(&self) {
+        self.configure(&[]);
+    }
+}
+
+#[tokio::test]
+async fn pending_active_restorable_reads_preserve_exact_pin_across_restart() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    let original = package("0.1.0");
+    for (state, kind) in [
+        (PlanPinState::Pending, "pending"),
+        (PlanPinState::Active, "active"),
+        (PlanPinState::Restorable, "restorable"),
+    ] {
+        let before = f.retained(&saved.plan_id);
+        assert_eq!(before, [kind]);
+        assert_eq!(
+            f.authorize(&request).unwrap(),
+            PluginArtifactDeliveryMetadata {
+                workspace_id: "a".into(),
+                request: request.clone(),
+                artifact: installed.installation.artifacts[0].clone(),
+            }
+        );
+        assert_eq!(f.bytes(&request).unwrap(), original);
+        assert_eq!(f.retained(&saved.plan_id), before);
+        assert_eq!(
+            f.store
+                .plugin_plan_load("a", &saved.plan_id)
+                .unwrap()
+                .pin_state,
+            state
+        );
+        let reopened = SqliteWorkspaceStore::open_for_test(f.dir.path().join("server.db")).unwrap();
+        assert_eq!(
+            reopened
+                .plugin_delivery_bytes(&f.root(), "a", &execution(), &request)
+                .unwrap(),
+            original
+        );
+        if state != PlanPinState::Restorable {
+            f.store
+                .plugin_plan_transition(
+                    "a",
+                    &saved.plan_id,
+                    state,
+                    if state == PlanPinState::Pending {
+                        PlanPinState::Active
+                    } else {
+                        PlanPinState::Restorable
+                    },
+                )
+                .unwrap();
+        }
+    }
+}
+
+#[tokio::test]
+async fn foreign_workspace_host_identity_and_plan_coordinates_cannot_fetch_bytes() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let other = f.install("other");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    error(
+        f.store
+            .plugin_delivery_authorize("b", &execution(), &request),
+        PluginCatalogError::NotFound,
+    );
+    error(
+        f.store
+            .plugin_delivery_bytes(&f.root(), "b", &execution(), &request),
+        PluginCatalogError::NotFound,
+    );
+    let mut identities = Vec::new();
+    let mut wrong = execution();
+    wrong.runtime_id = "other-runtime".into();
+    identities.push(wrong);
+    let mut wrong = execution();
+    wrong.worker_id = "other-worker".into();
+    identities.push(wrong);
+    let mut wrong = execution();
+    wrong.principal_id = "other-principal".into();
+    identities.push(wrong);
+    let mut wrong = execution();
+    wrong.delegation_id = Some("other-delegation".into());
+    identities.push(wrong);
+    let mut wrong = execution();
+    wrong.delegation_id = None;
+    identities.push(wrong);
+    for wrong in identities {
+        error(
+            f.store.plugin_delivery_authorize("a", &wrong, &request),
+            PluginCatalogError::Conflict,
+        );
+        error(
+            f.store
+                .plugin_delivery_bytes(&f.root(), "a", &wrong, &request),
+            PluginCatalogError::Conflict,
+        );
+    }
+    let mut cases = Vec::new();
+    let mut wrong = request.clone();
+    wrong.plan_id = "missing-plan".into();
+    cases.push((wrong, PluginCatalogError::NotFound));
+    let mut wrong = request.clone();
+    wrong.plan_fingerprint = format!("sha256:{}", "0".repeat(64));
+    cases.push((wrong, PluginCatalogError::Conflict));
+    let mut wrong = request.clone();
+    wrong.installation_id = other.installation.installation_id.clone();
+    cases.push((wrong, PluginCatalogError::NotFound));
+    let mut wrong = request.clone();
+    wrong.artifact_digest = format!("sha256:{}", "0".repeat(64));
+    cases.push((wrong, PluginCatalogError::NotFound));
+    // Another real persisted plan cannot lend its fingerprint to this plan.
+    let other_plan = f.create(&other, "other-launch");
+    let mut wrong = request.clone();
+    wrong.plan_id = other_plan.plan_id;
+    cases.push((wrong, PluginCatalogError::Conflict));
+    for (wrong, expected) in cases {
+        error(f.authorize(&wrong), expected);
+        error(f.bytes(&wrong), expected);
+    }
+    assert_eq!(f.retained(&saved.plan_id), ["pending"]);
+}
+
+#[tokio::test]
+async fn revoked_artifact_denies_new_metadata_and_bytes_without_releasing_audit_pin() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    // Previously obtained metadata cannot stand in for current admission.
+    let metadata = f.authorize(&request).unwrap();
+    f.revoke(&installed);
+    error(f.authorize(&metadata.request), PluginCatalogError::Conflict);
+    error(f.bytes(&metadata.request), PluginCatalogError::Conflict);
+    let stored = f.store.plugin_plan_load("a", &saved.plan_id).unwrap();
+    assert_eq!(stored.plan, saved.plan);
+    assert_eq!(stored.pin_state, PlanPinState::Pending);
+    assert_eq!(f.retained(&saved.plan_id), ["pending"]);
+}
+
+#[tokio::test]
+async fn released_plan_and_missing_or_wrong_kind_reference_never_authorize_delivery() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    // Retention corruption is injected after valid owner creation, not by
+    // constructing or inserting a fake serialized plan.
+    f.store
+        .with_conn(|conn| {
+            conn.execute(
+                "UPDATE plugin_references SET kind='active' WHERE workspace_id='a' AND owner=?1",
+                [&saved.plan_id],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+    error(f.authorize(&request), PluginCatalogError::Conflict);
+    error(f.bytes(&request), PluginCatalogError::Conflict);
+    f.store
+        .with_conn(|conn| {
+            conn.execute(
+                "DELETE FROM plugin_references WHERE workspace_id='a' AND owner=?1",
+                [&saved.plan_id],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+    error(f.authorize(&request), PluginCatalogError::Conflict);
+    error(f.bytes(&request), PluginCatalogError::Conflict);
+    let another = f.create(&installed, "second-launch");
+    let another_request = self::request(&another);
+    f.store
+        .plugin_plan_transition(
+            "a",
+            &another.plan_id,
+            PlanPinState::Pending,
+            PlanPinState::Released,
+        )
+        .unwrap();
+    error(f.authorize(&another_request), PluginCatalogError::Conflict);
+    error(f.bytes(&another_request), PluginCatalogError::Conflict);
+    assert!(f.retained(&another.plan_id).is_empty());
+    assert_eq!(
+        f.store
+            .plugin_plan_load("a", &another.plan_id)
+            .unwrap()
+            .plan,
+        another.plan
+    );
+}
+
+#[tokio::test]
+async fn new_catalog_version_and_current_config_change_do_not_reselect_saved_pin() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    let original = f.bytes(&request).unwrap();
+    let upload = f
+        .store
+        .plugin_upload(&f.root(), "a", "actor", &package("0.2.0"))
+        .unwrap();
+    let added = f
+        .store
+        .plugin_add_artifact(
+            "a",
+            "actor",
+            &installed.installation.installation_id,
+            &PluginAddArtifactRequest {
+                upload_id: upload.upload_id,
+                expected_last_operation: installed.operation_id.clone(),
+                operation_key: "add-version".into(),
+            },
+        )
+        .unwrap();
+    assert_ne!(added.operation_id, installed.operation_id);
+    f.disable();
+    assert_eq!(
+        f.authorize(&request).unwrap().artifact,
+        installed.installation.artifacts[0]
+    );
+    assert_eq!(f.bytes(&request).unwrap(), original);
+    assert_eq!(f.retained(&saved.plan_id), ["pending"]);
+    let newer = added
+        .installation
+        .artifacts
+        .iter()
+        .find(|a| a.version == "0.2.0")
+        .unwrap();
+    let mut outside = request.clone();
+    outside.artifact_digest = newer.artifact_digest.clone();
+    error(f.authorize(&outside), PluginCatalogError::NotFound);
+    error(f.bytes(&outside), PluginCatalogError::NotFound);
+}
+
+#[tokio::test]
+async fn release_then_uninstall_keeps_plan_for_audit_but_denies_delivery() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    f.store
+        .plugin_plan_transition(
+            "a",
+            &saved.plan_id,
+            PlanPinState::Pending,
+            PlanPinState::Released,
+        )
+        .unwrap();
+    f.disable();
+    f.store
+        .plugin_uninstall(
+            "a",
+            "actor",
+            &installed.installation.installation_id,
+            &PluginUninstallRequest {
+                expected_last_operation: installed.operation_id.clone(),
+                operation_key: "uninstall".into(),
+            },
+        )
+        .unwrap();
+    error(f.authorize(&request), PluginCatalogError::Conflict);
+    error(f.bytes(&request), PluginCatalogError::Conflict);
+    assert_eq!(
+        f.store.plugin_plan_load("a", &saved.plan_id).unwrap().plan,
+        saved.plan
+    );
+}
+
+#[tokio::test]
+async fn artifact_metadata_identity_corruption_cannot_override_persisted_pin() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    let original = serde_json::to_value(&installed.installation.artifacts[0]).unwrap();
+    let mut corruptions = Vec::new();
+    for (name, value) in [
+        (
+            "artifact_digest",
+            serde_json::json!(format!("sha256:{}", "0".repeat(64))),
+        ),
+        (
+            "package_digest",
+            serde_json::json!(format!("sha256:{}", "0".repeat(64))),
+        ),
+        ("package_id", serde_json::json!("other-package")),
+        ("version", serde_json::json!("9.0.0")),
+        ("artifact_size", serde_json::json!(1)),
+        ("schema_version", serde_json::json!(999)),
+        ("world", serde_json::json!("unsupported-world")),
+        ("required_imports", serde_json::json!(["other-import"])),
+    ] {
+        let mut value_ = original.clone();
+        value_[name] = value;
+        corruptions.push(value_);
+    }
+    for name in ["identity", "version", "namespace", "digest"] {
+        let mut changed = original.clone();
+        changed["config_schema"][name] = if name == "digest" {
+            serde_json::json!(format!("sha256:{}", "0".repeat(64)))
+        } else {
+            serde_json::json!("other-schema")
+        };
+        corruptions.push(changed);
+    }
+    for metadata in corruptions {
+        f.store.with_conn(|conn| {
+            conn.execute("UPDATE plugin_artifacts SET metadata_json=?1 WHERE workspace_id='a' AND installation_id=?2 AND artifact_digest=?3", params![metadata.to_string(), request.installation_id, request.artifact_digest])?;
+            Ok(())
+        }).unwrap();
+        assert!(
+            f.authorize(&request).is_err(),
+            "admitted changed {metadata}"
+        );
+        assert!(f.bytes(&request).is_err(), "fetched changed {metadata}");
+    }
+    assert_eq!(f.retained(&saved.plan_id), ["pending"]);
+}
+
+#[tokio::test]
+async fn bounded_blob_read_rejects_missing_truncated_oversize_and_digest_mismatch() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    let path = f
+        .root()
+        .join("blobs")
+        .join(request.artifact_digest.strip_prefix("sha256:").unwrap());
+    let original = fs::read(&path).unwrap();
+    fs::remove_file(&path).unwrap();
+    error(f.bytes(&request), PluginCatalogError::Unavailable);
+    for bad in [vec![], vec![0; original.len() + 1], vec![0; original.len()]] {
+        fs::write(&path, bad).unwrap();
+        error(f.bytes(&request), PluginCatalogError::Unavailable);
+    }
+    fs::write(&path, &original).unwrap();
+    assert_eq!(f.bytes(&request).unwrap(), original);
+    assert_eq!(f.retained(&saved.plan_id), ["pending"]);
+}
+
+#[tokio::test]
+async fn owner_transaction_fences_concurrent_catalog_mutation_until_bytes_are_read() {
+    let f = Fixture::new().await;
+    let installed = f.install("one");
+    let saved = f.create(&installed, "launch");
+    let request = request(&saved);
+    let mut other = Connection::open(f.dir.path().join("server.db")).unwrap();
+    other.busy_timeout(std::time::Duration::ZERO).unwrap();
+    f.store
+        .with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            plugin_delivery_authorize(&tx, "a", &execution(), &request)?;
+            assert!(
+                other
+                    .transaction_with_behavior(TransactionBehavior::Immediate)
+                    .is_err()
+            );
+            assert_eq!(
+                plugin_delivery_bytes(&tx, &f.root(), "a", &execution(), &request)?,
+                package("0.1.0")
+            );
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+    other
+        .transaction_with_behavior(TransactionBehavior::Immediate)
+        .unwrap()
+        .rollback()
+        .unwrap();
+    f.revoke(&installed);
+    error(f.bytes(&request), PluginCatalogError::Conflict);
+}

crates/workspace-server/src/server.rs

@@ -49,6 +49,9 @@ mod drive;
 mod drive_grants;
 #[path = "server_plugin.rs"]
 mod plugin;
+#[path = "server_plugin_delivery.rs"]
+mod plugin_delivery;
+pub use plugin_delivery::InProcessPluginArtifactSource;
 #[path = "server_workspace_config.rs"]
 mod workspace_config;
 use std::collections::{BTreeMap, HashMap, HashSet};
@@ -7656,6 +7659,7 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
             service.clone(),
         ))
         .merge(plugin::router(service.clone()))
+        .merge(plugin_delivery::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(
@@ -13586,6 +13590,27 @@ impl server_api::ServerApi for ServerApiContractService {
         .map_err(ApiError::into_repository_api_error)
     }
 
+    async fn plugin_artifact_authorize(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        request: server_api::PluginArtifactDeliveryRequest,
+    ) -> std::result::Result<
+        server_api::PluginArtifactDeliveryMetadata,
+        server_api::RepositoryApiError,
+    > {
+        plugin_delivery::authorize(self.workspace_api()?, context, workspace_id, request).await
+    }
+
+    async fn plugin_artifact_pull(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        request: server_api::PluginArtifactDeliveryRequest,
+    ) -> std::result::Result<server_api::BinaryBody, server_api::RepositoryApiError> {
+        plugin_delivery::pull(self.workspace_api()?, context, workspace_id, request).await
+    }
+
     async fn plugin_upload(
         &self,
         actor: RequestActor,
@@ -38741,6 +38766,10 @@ mod tests {
         include!("server_ticket_case_signed_tests.rs");
     }
 
+    mod plugin_delivery_tests {
+        include!("server_plugin_delivery_tests.rs");
+    }
+
     mod plugin_tests {
         include!("server_plugin_tests.rs");
     }

crates/workspace-server/src/server/subject_authorization.rs

@@ -458,6 +458,37 @@ pub(super) fn worker_permissions(
     authority.subject_permissions(worker)
 }
 
+/// Host-only identity shared by plan creation and live Plugin operations. The
+/// identity is an immutable ceiling, not a permission: recheck the live lineage
+/// on every use, including byte delivery and cache hits.
+pub(super) fn plugin_execution_identity(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+) -> ApiResult<crate::plugin_plan::HostExecutionIdentity> {
+    worker_permissions(api, worker)?;
+    let (principal_id, delegation_id) = if let Some(grant) = live_child_grant(api, worker)? {
+        (
+            format!("subject:{}", grant.subject_id),
+            Some(grant.source_change_id),
+        )
+    } else if let Some(subject) = current_body_subject(api, worker)? {
+        (format!("subject:{subject}"), None)
+    } else if let Some(subject) = job_subject(api, worker)? {
+        (format!("subject:{subject}"), None)
+    } else {
+        (
+            format!("worker:{}/{}", worker.runtime_id, worker.worker_id),
+            None,
+        )
+    };
+    Ok(crate::plugin_plan::HostExecutionIdentity {
+        runtime_id: worker.runtime_id.clone(),
+        worker_id: worker.worker_id.clone(),
+        principal_id,
+        delegation_id,
+    })
+}
+
 pub(super) fn authorize(
     api: &WorkspaceApi,
     worker: &RuntimeWorkerRef,
@@ -886,6 +917,50 @@ pub(super) async fn interrupt_subject_workers(
     results
 }
 
+impl WorkspaceApi {
+    /// Trusted Host plan creation uses the same identity as live Plugin operations.
+    /// Principal and delegation are derived here, never accepted from public JSON.
+    pub async fn plugin_execution_identity(
+        &self,
+        worker: &RuntimeWorkerRef,
+    ) -> crate::Result<crate::plugin_plan::HostExecutionIdentity> {
+        self.store
+            .get_workspace_runtime_binding(self.workspace_id(), &worker.runtime_id)
+            .await?
+            .filter(|binding| binding.revoked_at.is_none())
+            .ok_or_else(|| {
+                Error::WorkspacePermissionDenied("Plugin Runtime trust revoked".into())
+            })?;
+        let api = self.clone();
+        let worker = worker.clone();
+        self.host_executors
+            .host()?
+            .run(move || {
+                let member = api.store.get_worker_registry(api.workspace_id(), &worker)?;
+                let reserved = api
+                    .store
+                    .has_active_worker_create_reservation(api.workspace_id(), &worker)?;
+                if member.is_none() && !reserved {
+                    return Err(Error::WorkspacePermissionDenied(
+                        "Plugin Worker is not registered".into(),
+                    ));
+                }
+                api.store
+                    .require_current_worker_singleton_owner(api.workspace_id(), &worker)?;
+                let summary = api.runtime.worker(&worker).map_err(|_| {
+                    Error::WorkspacePermissionDenied("Plugin Worker unavailable".into())
+                })?;
+                if summary.workspace.workspace_id.as_deref() != Some(api.workspace_id()) {
+                    return Err(Error::WorkspacePermissionDenied(
+                        "foreign Plugin Worker".into(),
+                    ));
+                }
+                plugin_execution_identity(&api, &worker).map_err(|error| error.error)
+            })
+            .await
+    }
+}
+
 pub(super) fn administration_actor(
     context: &server_api::ServerRequestContext,
 ) -> ApiResult<&RequestActor> {

crates/workspace-server/src/server_plugin.rs

@@ -24,7 +24,7 @@ pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
         .layer(middleware::from_fn(normalize_rejections))
 }
 
-async fn normalize_rejections(request: Request, next: Next) -> Response {
+pub(super) 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
@@ -40,7 +40,7 @@ async fn normalize_rejections(request: Request, next: Next) -> Response {
             Json(RepositoryApiError::new(
                 400,
                 "plugin_invalid",
-                "Invalid Plugin catalog request",
+                "Invalid Plugin request",
                 vec![],
             )),
         )

crates/workspace-server/src/server_plugin_delivery.rs

@@ -0,0 +1,215 @@
+//! Authenticated byte delivery and the in-process adapter share current Worker
+//! authority and the persisted plan owner. Neither a cache hit nor an adapter
+//! handle retains permission. No destination policy or credential reaches a guest.
+use super::*;
+use server_api::{PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest};
+use yoi_plugin_runtime::delivery::{ArtifactSource, ArtifactStream, DeliveryError};
+
+pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
+    Router::new()
+        .merge(server_api::server_api_axum::plugin_artifact_authorize(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::plugin_artifact_pull(service))
+        .layer(middleware::from_fn(plugin::normalize_rejections))
+}
+
+fn failure(error: impl Into<ApiError>) -> server_api::RepositoryApiError {
+    let error = error.into();
+    let status = match &error.error {
+        Error::PluginCatalog(error) => error.status(),
+        _ => api_error_status(&error.error).as_u16(),
+    };
+    let (status, code) = match status {
+        400 => (400, "plugin_delivery_invalid"),
+        401 | 403 => (403, "plugin_delivery_denied"),
+        404 => (404, "plugin_delivery_not_found"),
+        409 => (409, "plugin_delivery_conflict"),
+        413 => (413, "plugin_delivery_limit"),
+        _ => (503, "plugin_delivery_unavailable"),
+    };
+    server_api::RepositoryApiError::new(status, code, "Plugin artifact delivery refused", vec![])
+}
+
+fn delivery_error(error: server_api::RepositoryApiError) -> DeliveryError {
+    if api_macros::HttpError::status_code(&error) >= 500 {
+        DeliveryError::Unavailable("owner unavailable".into())
+    } else {
+        DeliveryError::Denied
+    }
+}
+
+async fn worker_from_context(
+    api: &WorkspaceApi,
+    workspace: &str,
+    context: &server_api::ServerRequestContext,
+) -> std::result::Result<RuntimeWorkerRef, server_api::RepositoryApiError> {
+    validate_workspace_scope(api, workspace).map_err(failure)?;
+    let headers = current_worker_contract_headers(context)?;
+    authenticate_worker_mutation_source_owned(api, workspace, &headers)
+        .await
+        .map_err(failure)
+}
+
+impl WorkspaceApi {
+    /// A Host-only adapter bound to this Server and Worker, not a loopback client.
+    /// Construction resolves the active Server key; every operation checks it
+    /// again, together with Worker membership, live grants, lineage and pin state.
+    pub async fn plugin_artifact_source(
+        &self,
+        worker: RuntimeWorkerRef,
+    ) -> crate::Result<InProcessPluginArtifactSource> {
+        let authority = authority_identity(self)?;
+        self.plugin_execution_identity(&worker).await?;
+        Ok(InProcessPluginArtifactSource {
+            api: self.clone(),
+            worker,
+            authority,
+        })
+    }
+}
+
+fn authority_identity(api: &WorkspaceApi) -> crate::Result<String> {
+    let identity = api
+        .store
+        .get_workspace_signing_identity(api.workspace_id())?
+        .ok_or_else(|| {
+            Error::WorkspacePermissionDenied("Plugin Server identity unavailable".into())
+        })?;
+    let bundle = project_workspace_signing_identity(api, identity)?
+        .public_bundle
+        .ok_or_else(|| {
+            Error::WorkspacePermissionDenied("Plugin Server identity inactive".into())
+        })?;
+    Ok(server_api::plugin_server_authority_identity(&bundle))
+}
+
+async fn current_execution(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+    request: &PluginArtifactDeliveryRequest,
+) -> std::result::Result<crate::plugin_plan::HostExecutionIdentity, server_api::RepositoryApiError>
+{
+    let execution = api
+        .plugin_execution_identity(worker)
+        .await
+        .map_err(failure)?;
+    let stage_api = api.clone();
+    let stage_worker = worker.clone();
+    let installation_id = request.installation_id.clone();
+    api.host_executors
+        .host()
+        .map_err(failure)?
+        .run(move || {
+            subject_authorization::authorize(
+                &stage_api,
+                &stage_worker,
+                &server_api::authorization::OperationAuthorizationRequest {
+                    operation: "plugin.artifact.read".into(),
+                    resource: server_api::authorization::PermissionResource::PluginInstallation {
+                        workspace_id: stage_api.workspace_id().into(),
+                        installation_id,
+                    },
+                },
+            )
+            .map_err(|error| error.error)
+        })
+        .await
+        .map_err(failure)?;
+    Ok(execution)
+}
+
+async fn metadata_for_worker(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+    request: PluginArtifactDeliveryRequest,
+) -> std::result::Result<PluginArtifactDeliveryMetadata, server_api::RepositoryApiError> {
+    let execution = current_execution(api, worker, &request).await?;
+    let store = api.config_store.clone();
+    let workspace = api.workspace_id().to_owned();
+    store
+        .db_executor()
+        .run(move || store.plugin_delivery_authorize(&workspace, &execution, &request))
+        .await
+        .map_err(failure)
+}
+
+async fn bytes_for_worker(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+    request: PluginArtifactDeliveryRequest,
+) -> std::result::Result<Vec<u8>, server_api::RepositoryApiError> {
+    let execution = current_execution(api, worker, &request).await?;
+    let store = api.config_store.clone();
+    let workspace = api.workspace_id().to_owned();
+    let root =
+        ServerConfig::server_data_root_for_data_dir(&api.config.data_root).join("plugin-artifacts");
+    let before_read = request.clone();
+    let bytes = store
+        .db_executor()
+        .run(move || store.plugin_delivery_bytes(&root, &workspace, &execution, &request))
+        .await
+        .map_err(failure)?;
+    // The storage read is not publication authority. Recheck security narrowing
+    // and the exact persisted pin after I/O; Runtime also checks before returning.
+    metadata_for_worker(api, worker, before_read).await?;
+    Ok(bytes)
+}
+
+pub(super) async fn authorize(
+    api: &WorkspaceApi,
+    context: server_api::ServerRequestContext,
+    workspace: String,
+    request: PluginArtifactDeliveryRequest,
+) -> std::result::Result<PluginArtifactDeliveryMetadata, server_api::RepositoryApiError> {
+    let worker = worker_from_context(api, &workspace, &context).await?;
+    metadata_for_worker(api, &worker, request).await
+}
+
+pub(super) async fn pull(
+    api: &WorkspaceApi,
+    context: server_api::ServerRequestContext,
+    workspace: String,
+    request: PluginArtifactDeliveryRequest,
+) -> std::result::Result<server_api::BinaryBody, server_api::RepositoryApiError> {
+    let worker = worker_from_context(api, &workspace, &context).await?;
+    bytes_for_worker(api, &worker, request)
+        .await
+        .map(server_api::BinaryBody::from)
+}
+
+pub struct InProcessPluginArtifactSource {
+    api: WorkspaceApi,
+    worker: RuntimeWorkerRef,
+    authority: String,
+}
+
+#[async_trait::async_trait]
+impl ArtifactSource for InProcessPluginArtifactSource {
+    fn server_authority_identity(&self) -> &str {
+        &self.authority
+    }
+
+    async fn authorize(
+        &self,
+        request: &PluginArtifactDeliveryRequest,
+    ) -> std::result::Result<PluginArtifactDeliveryMetadata, DeliveryError> {
+        if authority_identity(&self.api).map_err(|_| DeliveryError::Denied)? != self.authority {
+            return Err(DeliveryError::Denied);
+        }
+        metadata_for_worker(&self.api, &self.worker, request.clone())
+            .await
+            .map_err(delivery_error)
+    }
+
+    async fn fetch(
+        &self,
+        request: &PluginArtifactDeliveryRequest,
+    ) -> std::result::Result<ArtifactStream, DeliveryError> {
+        self.authorize(request).await?;
+        let bytes = bytes_for_worker(&self.api, &self.worker, request.clone())
+            .await
+            .map_err(delivery_error)?;
+        Ok(Box::pin(std::io::Cursor::new(bytes)))
+    }
+}