feat: bind Plugin launch selections and authenticated Runtime providers

2026-10-30 05:28:51 +09:00
parent 0cd4811739
commit 479bf0872a
17 changed files with 1298 additions and 38 deletions
@@ -232,8 +232,9 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
        _ => fs_paths.runtime_dir.clone(),
    };
    let mut factory = ProfileRuntimeWorkerFactory::new(fs_paths.worker_dir.join("worker-root"))
-        .with_runtime_store_dir(runtime_store_dir);
+        .with_runtime_store_dir(runtime_store_dir.clone());
    let runtime_auth = read_runtime_auth_file(&runtime_auth_path(config))?;
+    let mut plugin_clients = std::collections::HashMap::new();
    if let Some(identity) = runtime_auth.identity.clone() {
        factory = factory.with_remote_worker_mutation_identity(identity);
    }
@@ -243,16 +244,42 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
            .iter()
            .filter(|issuer| issuer.state == WorkspaceIssuerTrustState::Active)
        {
-            factory = factory.with_workspace_request_client(
-                RuntimeWorkspaceRequestClient::new(
+            let client = RuntimeWorkspaceRequestClient::new(
                issuer.workspace_id.clone(),
                issuer.backend_url.clone(),
                identity.identity_id.clone(),
            )
-                .with_runtime_request_source(identity, issuer.backend_url.clone()),
-            );
+            .with_runtime_request_source(identity, issuer.backend_url.clone());
+            plugin_clients.insert(issuer.workspace_id.clone(), client.clone());
+            factory = factory.with_workspace_request_client(client);
        }
    }
+    if !plugin_clients.is_empty() {
+        let auth_path = runtime_auth_path(config);
+        let current: Arc<worker_runtime::plugin::remote::CurrentPluginServer> =
+            Arc::new(move |workspace| {
+                read_runtime_auth_file(&auth_path)
+                    .ok()
+                    .and_then(|auth| {
+                        auth.workspace_issuers
+                            .into_iter()
+                            .find(|issuer| issuer.workspace_id == workspace)
+                    })
+                    .ok_or(yoi_plugin_runtime::delivery::DeliveryError::Denied)
+            });
+        let providers = Arc::new(
+            worker_runtime::plugin::remote::RemotePluginRuntimeProviders::new(
+                plugin_clients,
+                current,
+            ),
+        );
+        let plugins = worker_runtime::plugin::PinnedPluginToolRuntime::new(
+            providers,
+            &runtime_store_dir.join("plugin-cache"),
+        )
+        .map_err(ProcessError::WorkerAdapter)?;
+        factory = factory.with_plugin_runtime(Arc::new(plugins));
+    }
    let mut backend_resource_client: Option<
        Arc<dyn worker_runtime::resource::BackendResourceClient>,
    > = None;
@@ -85,3 +85,208 @@ pub async fn materialize(
        .await
        .map_err(|error| error.to_string())
 }
+
+pub struct PluginRuntimeServices {
+    pub artifacts: std::sync::Arc<dyn yoi_plugin_runtime::delivery::ArtifactSource>,
+    pub owner: std::sync::Arc<dyn yoi_plugin_runtime::owner::PluginHostOwner>,
+}
+
+/// Supplies authenticated process-owned adapters, not a Plugin resolver. Handles
+/// are bound to the exact requested Worker; a connection failure has no fallback.
+#[async_trait]
+pub trait PluginRuntimeProviders: Send + Sync {
+    async fn connect(
+        &self,
+        workspace: &str,
+        worker: &WorkerRef,
+    ) -> Result<PluginRuntimeServices, String>;
+}
+
+pub struct PinnedPluginToolRuntime {
+    providers: std::sync::Arc<dyn PluginRuntimeProviders>,
+    cache: yoi_plugin_runtime::delivery::ArtifactCache,
+    limits: yoi_plugin_runtime::Limits,
+}
+impl PinnedPluginToolRuntime {
+    pub fn new(
+        providers: std::sync::Arc<dyn PluginRuntimeProviders>,
+        private_cache_root: &std::path::Path,
+    ) -> Result<Self, String> {
+        Ok(Self {
+            providers,
+            cache: yoi_plugin_runtime::delivery::ArtifactCache::new(
+                private_cache_root,
+                yoi_plugin_runtime::delivery::CacheLimits::default(),
+            )
+            .map_err(|_| "Plugin cache initialization failed")?,
+            limits: yoi_plugin_runtime::Limits::default(),
+        })
+    }
+}
+#[async_trait]
+impl PluginToolRuntime for PinnedPluginToolRuntime {
+    async fn prepare(
+        &self,
+        worker: &WorkerRef,
+        snapshot: &runtime_api::PluginExecutionSnapshot,
+        session_id: &str,
+    ) -> Result<worker::plugin::PluginToolsFeature, String> {
+        snapshot
+            .plan
+            .validate_canonical()
+            .map_err(|_| "invalid Plugin plan")?;
+        if snapshot
+            .plan
+            .fingerprint()
+            .map_err(|_| "invalid Plugin plan")?
+            != snapshot.fingerprint
+            || snapshot.plan.execution().worker_id != worker.worker_id.to_string()
+        {
+            return Err("Plugin launch identity mismatch".into());
+        }
+        let services = self
+            .providers
+            .connect(snapshot.plan.workspace_id(), worker)
+            .await?;
+        for pin in snapshot.plan.plugins() {
+            services
+                .owner
+                .prepare(
+                    snapshot.plan_id.clone(),
+                    pin.installation_id().into(),
+                    pin.projection().tools.clone(),
+                )
+                .await
+                .map_err(|_| "Plugin startup admission refused or unavailable")?;
+        }
+        let tools = materialize(
+            worker,
+            snapshot,
+            session_id,
+            services.artifacts.as_ref(),
+            services.owner.clone(),
+            &self.cache,
+            self.limits.clone(),
+        )
+        .await?;
+        // Materialization and compilation can wait on I/O. Recheck at publication
+        // preparation, including pure Tools and cache hits, without making a call.
+        for pin in snapshot.plan.plugins() {
+            services
+                .owner
+                .prepare(
+                    snapshot.plan_id.clone(),
+                    pin.installation_id().into(),
+                    pin.projection().tools.clone(),
+                )
+                .await
+                .map_err(|_| "Plugin startup publication refused or unavailable")?;
+        }
+        Ok(tools)
+    }
+}
+
+#[cfg(feature = "http-server")]
+pub mod remote {
+    use super::*;
+    use crate::{
+        workspace_issuer::WorkspaceIssuerTrustRecord,
+        workspace_request::{
+            RuntimeWorkspaceRequestClient,
+            plugin_delivery::{CurrentArtifactServer, RemotePluginArtifactSource},
+        },
+    };
+    use std::{collections::HashMap, sync::Arc, time::Duration};
+    pub type CurrentPluginServer = dyn Fn(&str) -> Result<WorkspaceIssuerTrustRecord, yoi_plugin_runtime::delivery::DeliveryError>
+        + Send
+        + Sync;
+    pub struct RemotePluginRuntimeProviders {
+        clients: HashMap<String, RuntimeWorkspaceRequestClient>,
+        current: Arc<CurrentPluginServer>,
+    }
+    impl RemotePluginRuntimeProviders {
+        pub fn new(
+            clients: HashMap<String, RuntimeWorkspaceRequestClient>,
+            current: Arc<CurrentPluginServer>,
+        ) -> Self {
+            Self { clients, current }
+        }
+    }
+    struct RegisteredOwner {
+        source: Arc<RemotePluginArtifactSource>,
+        owner: Arc<dyn yoi_plugin_runtime::owner::PluginHostOwner>,
+    }
+    #[async_trait]
+    impl yoi_plugin_runtime::owner::PluginHostOwner for RegisteredOwner {
+        async fn call(
+            &self,
+            request: server_api::PluginHostRequest,
+        ) -> Result<server_api::PluginHostResponse, server_api::PluginHostError> {
+            let denied = || server_api::PluginHostError {
+                code: server_api::PluginHostErrorCode::Denied,
+                effect: server_api::PluginEffectObservation::NotDispatched,
+                diagnostic_id: String::new(),
+            };
+            self.source.registration().map_err(|_| denied())?;
+            let dispatched = matches!(request, server_api::PluginHostRequest::Send { .. });
+            let response = self.owner.call(request).await?;
+            self.source
+                .registration()
+                .map_err(|_| server_api::PluginHostError {
+                    code: server_api::PluginHostErrorCode::Denied,
+                    effect: if dispatched {
+                        server_api::PluginEffectObservation::Indeterminate
+                    } else {
+                        server_api::PluginEffectObservation::NotDispatched
+                    },
+                    diagnostic_id: String::new(),
+                })?;
+            Ok(response)
+        }
+    }
+    #[async_trait]
+    impl PluginRuntimeProviders for RemotePluginRuntimeProviders {
+        async fn connect(
+            &self,
+            workspace: &str,
+            worker: &WorkerRef,
+        ) -> Result<PluginRuntimeServices, String> {
+            let client = self
+                .clients
+                .get(workspace)
+                .ok_or("Plugin Server is not registered")?
+                .clone();
+            let current = self.current.clone();
+            let workspace = workspace.to_owned();
+            let registered: Arc<CurrentArtifactServer> = Arc::new(move || current(&workspace));
+            let source = Arc::new(
+                RemotePluginArtifactSource::new(
+                    client.clone(),
+                    worker.worker_id.to_string(),
+                    registered,
+                )
+                .map_err(|_| "Plugin Server trust unavailable")?,
+            );
+            let generated = client
+                .server_api_client(
+                    crate::auth::WORKSPACE_REQUEST_PERMISSION,
+                    Some(worker.worker_id.to_string()),
+                    reqwest::header::HeaderMap::new(),
+                    Duration::from_secs(30),
+                    8 * 1024 * 1024,
+                )
+                .map_err(|_| "Plugin Host client unavailable")?;
+            let owner = Arc::new(RegisteredOwner {
+                source: source.clone(),
+                owner: Arc::new(yoi_plugin_runtime::owner::RemotePluginHostOwner::new(
+                    Arc::new(generated),
+                    client.workspace_id().into(),
+                )),
+            });
+            Ok(PluginRuntimeServices {
+                artifacts: source,
+                owner,
+            })
+        }
+    }
+}
@@ -174,6 +174,23 @@ impl RuntimeWorkspaceRequestClient {
            && self.base_url.trim_end_matches('/') == base_url.trim_end_matches('/')
    }

+    /// Expose the existing Runtime signer only as a request-bound authorizer for
+    /// the in-process Plugin Host adapter. This does not authenticate a request;
+    /// the Server still verifies its method/path/body proof on every operation.
+    pub fn plugin_host_authorizer(
+        &self,
+        worker_id: String,
+    ) -> std::sync::Arc<dyn server_api::client_support::RequestAuthorizer> {
+        std::sync::Arc::new(RuntimeWorkspaceRequestAuthorizer {
+            workspace_id: self.workspace_id.clone(),
+            runtime_id: self.runtime_id.clone(),
+            request_source: self.request_source.clone(),
+            permission: crate::auth::WORKSPACE_REQUEST_PERMISSION.into(),
+            worker_id: Some(worker_id),
+            extra_headers: HeaderMap::new(),
+        })
+    }
+
    pub(crate) fn server_api_client(
        &self,
        permission: impl Into<String>,
@@ -42,7 +42,7 @@ impl RemotePluginArtifactSource {
        Ok(source)
    }

-    fn registration(&self) -> Result<(), DeliveryError> {
+    pub(crate) fn registration(&self) -> Result<(), DeliveryError> {
        let registration = (self.current_server)()?;
        if registration.state != WorkspaceIssuerTrustState::Active
            || registration.workspace_id != self.client.workspace_id
@@ -24,6 +24,7 @@ pub use store::*;

 /// Included by the config owner's ordered database migration, never at runtime.
 pub const SCHEMA_SQL: &str = include_str!("plugin_plan/schema.sql");
+pub(crate) const LAUNCH_SCHEMA_SQL: &str = include_str!("plugin_plan/launch_schema.sql");

 impl From<PlanError> for Error {
    fn from(error: PlanError) -> Self {
@@ -0,0 +1,37 @@
+// Same store migration runner and retained-history fixture as the ordinary suite.
+#[test]
+fn plugin_launch_migration_preserves_reservations_and_rolls_back_failed_history_commit() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("workspace.db");
+    prepare_retained_schema(&path, 109);
+    let conn = Connection::open(&path).unwrap();
+    configure_sqlite(&conn).unwrap();
+    let before = stored_fixture_rows(&conn, "worker_create_reservations");
+    conn.execute_batch("CREATE TRIGGER reject_launch_migration BEFORE INSERT ON __yoi_schema_migrations WHEN NEW.version=110 BEGIN SELECT RAISE(ABORT,'injected history failure'); END;").unwrap();
+    assert!(migrate_plugin_launches_v109_to_v110(&conn).is_err());
+    assert_eq!(current_schema_version(&conn).unwrap(), 109);
+    assert!(!table_exists(&conn, "worker_plugin_launches").unwrap());
+    assert_eq!(
+        stored_fixture_rows(&conn, "worker_create_reservations"),
+        before
+    );
+    conn.execute_batch("DROP TRIGGER reject_launch_migration")
+        .unwrap();
+    migrate_plugin_launches_v109_to_v110(&conn).unwrap();
+    crate::plugin_plan::verify_launch_schema(&conn).unwrap();
+    assert_eq!(
+        stored_fixture_rows(&conn, "worker_create_reservations"),
+        before
+    );
+    assert!(conn.execute("INSERT INTO worker_plugin_launches(workspace_id,worker_id,runtime_id,plan_id) VALUES('missing','missing','runtime',NULL)", []).is_err());
+    let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap();
+    let ddl = |c: &Connection| {
+        c.query_row(
+            "SELECT sql FROM sqlite_schema WHERE name='worker_plugin_launches'",
+            [],
+            |r| r.get::<_, String>(0),
+        )
+        .unwrap()
+    };
+    assert_eq!(ddl(&conn), fresh.with_conn(|conn| Ok(ddl(conn))).unwrap());
+}
@@ -0,0 +1,9 @@
+CREATE TABLE worker_plugin_launches (
+    workspace_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    plan_id TEXT,
+    PRIMARY KEY(workspace_id, worker_id),
+    FOREIGN KEY(workspace_id, worker_id) REFERENCES worker_create_reservations(workspace_id, worker_id) ON DELETE CASCADE,
+    FOREIGN KEY(workspace_id, plan_id) REFERENCES plugin_execution_plans(workspace_id, plan_id)
+);
@@ -5,6 +5,49 @@ use crate::{
 };
 use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};

+pub(crate) const WORKER_LAUNCH_ACTOR: &str = "server-plugin-launch";
+pub(crate) fn worker_launch_key(runtime: &str, worker: &str) -> String {
+    config_source::digest_bytes(&serde_json::to_vec(&(runtime, worker)).expect("identity encoding"))
+}
+
+/// An existing row with a NULL plan seals the empty selection. Absence is an
+/// unresolved reservation, not permission to reselect after Runtime dispatch.
+pub(crate) fn worker_launch_selection(
+    conn: &Connection,
+    workspace: &str,
+    runtime: &str,
+    worker: &str,
+) -> Result<Option<Option<StoredPluginPlan>>> {
+    let id: Option<Option<String>> = conn.query_row("SELECT plan_id FROM worker_plugin_launches WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",
+        params![workspace, runtime, worker], |row| row.get(0)).optional()?;
+    id.map(|id| {
+        id.map(|id| plugin_plan_load(conn, workspace, &id))
+            .transpose()
+    })
+    .transpose()
+}
+pub(crate) fn worker_launch_plan(
+    conn: &Connection,
+    workspace: &str,
+    runtime: &str,
+    worker: &str,
+) -> Result<Option<StoredPluginPlan>> {
+    Ok(worker_launch_selection(conn, workspace, runtime, worker)?.flatten())
+}
+
+pub(crate) fn transition_worker_launch_plan(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    runtime: &str,
+    worker: &str,
+    target: PlanPinState,
+) -> Result<()> {
+    if let Some(plan) = worker_launch_plan(tx, workspace, runtime, worker)? {
+        plugin_plan_transition(tx, workspace, &plan.plan_id, plan.pin_state, target)?;
+    }
+    Ok(())
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum PlanPinState {
@@ -306,6 +349,25 @@ impl SqliteWorkspaceStore {
    }
 }

+pub(crate) fn verify_launch_schema(conn: &Connection) -> Result<()> {
+    let expected = Connection::open_in_memory()?;
+    expected.execute_batch(LAUNCH_SCHEMA_SQL)?;
+    let sql: String = expected.query_row(
+        "SELECT sql FROM sqlite_schema WHERE name='worker_plugin_launches'",
+        [],
+        |row| row.get(0),
+    )?;
+    let actual: String = conn.query_row(
+        "SELECT sql FROM sqlite_schema WHERE name='worker_plugin_launches'",
+        [],
+        |row| row.get(0),
+    )?;
+    if actual != sql {
+        return Err(Error::Store("Worker Plugin launch schema mismatch".into()));
+    }
+    Ok(())
+}
+
 pub(crate) fn verify_plan_schema(conn: &Connection) -> Result<()> {
    let expected = Connection::open_in_memory()?;
    expected.execute_batch(SCHEMA_SQL)?;
@@ -52,6 +52,8 @@ mod plugin;
 mod plugin_delivery;
 #[path = "server_plugin_host.rs"]
 mod plugin_host;
+#[path = "server_plugin_runtime.rs"]
+mod plugin_runtime;
 pub use plugin_delivery::InProcessPluginArtifactSource;
 #[path = "server_workspace_config.rs"]
 mod workspace_config;
@@ -1990,6 +1992,7 @@ pub struct WorkspaceApi {
    backend_job_drain_requests: Arc<AtomicU64>,
    journal_lifecycle: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
    journal_dispatch_owner: Option<Arc<WorkspaceApi>>,
+    plugin_runtime_owner: Option<Arc<WorkspaceApi>>,
    observation_proxy: BackendObservationProxy,
    runtime_subscription_broker: RuntimeSubscriptionBroker,
    pub(crate) worker_projection: Arc<crate::worker_projection::WorkerProjectionService>,
@@ -4157,8 +4160,15 @@ impl WorkspaceApi {
                store.clone(),
            ),
        );
+        let plugin_providers = Arc::new(plugin_runtime::EmbeddedPluginRuntimeProviders::default());
+        let plugins = worker_runtime::plugin::PinnedPluginToolRuntime::new(
+            plugin_providers.clone(),
+            &config.embedded_runtime_store_root.join("plugin-cache"),
+        )
+        .map_err(crate::Error::Config)?;
        let execution_backend = WorkerRuntimeExecutionBackend::new(
            ProfileRuntimeWorkerFactory::new(config.workspace_execution_root.clone())
+                .with_plugin_runtime(Arc::new(plugins))
                .with_embedded_worker_mutation_dispatcher(
                    EMBEDDED_RUNTIME_ID,
                    worker_remove_dispatcher.clone(),
@@ -4194,7 +4204,7 @@ impl WorkspaceApi {
            })
            .collect::<HashMap<_, _>>();
        let workspace_id = config.workspace_id.clone();
-        let api = Self::new_with_execution_backend_and_broker(
+        let mut api = Self::new_with_execution_backend_and_broker(
            config,
            store,
            Arc::new(execution_backend),
@@ -4245,6 +4255,9 @@ impl WorkspaceApi {
        {
            api.register_workspace_runtime_binding(binding)?;
        }
+        let plugin_owner = Arc::new(api.clone());
+        plugin_providers.bind(&plugin_owner);
+        api.plugin_runtime_owner = Some(plugin_owner);
        Ok(api)
    }

@@ -4503,6 +4516,7 @@ impl WorkspaceApi {
            backend_job_drain_requests: Arc::new(AtomicU64::new(0)),
            journal_lifecycle: Arc::new(Mutex::new(HashMap::new())),
            journal_dispatch_owner: None,
+            plugin_runtime_owner: None,
            observation_proxy,
            runtime_subscription_broker,
            worker_projection,
@@ -6063,13 +6077,26 @@ impl WorkspaceApi {
                &worker_id.to_string(),
            )?;
        }
-        request.resolved_memory_settings = Some(reservation.memory_settings);
-        let reservation_fingerprint = reservation.create_fingerprint.clone();
-        let create_binding = WorkerCreateBinding {
-            plugin_execution: None,
+        let create_binding = match self.plugin_launch_binding(runtime_id, &request, &reservation) {
+            Ok(binding) => binding,
+            Err(error) => {
+                let mut diagnostics = cleanup_spawn_created_workdirs(self, &spawned_workdir_ids);
+                if let Err(cleanup) = self.config_store.fail_worker_create_reservation(
+                    &self.config.workspace_id,
+                    runtime_id,
                    worker_id,
-            create_fingerprint: reservation.create_fingerprint,
+                    &reservation.create_fingerprint,
+                ) {
+                    diagnostics.push(spawn_compensation_diagnostic(
+                        "plugin_launch_reservation_cleanup_failed",
+                        cleanup.to_string(),
+                    ));
+                }
+                return Err(ApiError::with_diagnostics(error, diagnostics));
+            }
        };
+        request.resolved_memory_settings = Some(reservation.memory_settings);
+        let reservation_fingerprint = create_binding.create_fingerprint.clone();
        let lifecycle_assignment = request.ticket_assignment.clone();
        let compensation_context = WorkerSpawnCompensationContext {
            assignment: lifecycle_assignment.as_ref(),
@@ -38691,6 +38718,7 @@ mod tests {

    mod plugin_delivery_tests {
        include!("server_plugin_delivery_tests.rs");
+        include!("server_plugin_launch_tests.rs");
    }

    mod plugin_tests {
@@ -38698,6 +38726,7 @@ mod tests {
    }
    mod plugin_host_tests {
        include!("server_plugin_host_tests.rs");
+        include!("server_plugin_runtime_tests.rs");
    }

    mod ticket_tags_tests {
@@ -45,13 +45,40 @@ async fn worker_from_context(
    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)
+    // Runtime source proof has already been checked by the contract router.
+    // Byte preparation precedes controller publication, so a live Runtime
+    // summary is not required; the reservation/lineage owner below is required.
+    subject_authorization::source_from_context(context).map_err(failure)
 }

 impl WorkspaceApi {
+    async fn plugin_artifact_execution_identity(
+        &self,
+        worker: &RuntimeWorkerRef,
+    ) -> crate::Result<crate::plugin_plan::HostExecutionIdentity> {
+        let api = self.clone();
+        let worker_owned = worker.clone();
+        let preparing = self
+            .host_executors
+            .host()?
+            .run(move || {
+                if api
+                    .store
+                    .has_active_worker_create_reservation(api.workspace_id(), &worker_owned)?
+                {
+                    api.plugin_reserved_identity(&worker_owned, None, None)
+                        .map(Some)
+                } else {
+                    Ok(None)
+                }
+            })
+            .await?;
+        match preparing {
+            Some(identity) => Ok(identity),
+            None => self.plugin_execution_identity(worker).await,
+        }
+    }
+
    /// 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.
@@ -60,7 +87,7 @@ impl WorkspaceApi {
        worker: RuntimeWorkerRef,
    ) -> crate::Result<InProcessPluginArtifactSource> {
        let authority = authority_identity(self)?;
-        self.plugin_execution_identity(&worker).await?;
+        self.plugin_artifact_execution_identity(&worker).await?;
        Ok(InProcessPluginArtifactSource {
            api: self.clone(),
            worker,
@@ -91,7 +118,7 @@ async fn current_execution(
 ) -> std::result::Result<crate::plugin_plan::HostExecutionIdentity, server_api::RepositoryApiError>
 {
    let execution = api
-        .plugin_execution_identity(worker)
+        .plugin_artifact_execution_identity(worker)
        .await
        .map_err(failure)?;
    let stage_api = api.clone();
@@ -82,14 +82,22 @@ fn artifact_package() -> Vec<u8> {
 }

 async fn delivery_worker(api: &WorkspaceApi, subject: Option<&str>) -> RuntimeWorkerRef {
+    delivery_worker_profile(api, subject, "builtin:companion").await
+}
+
+async fn delivery_worker_profile(api: &WorkspaceApi, subject: Option<&str>, profile: &str) -> RuntimeWorkerRef {
    let Json(worker) = create_workspace_worker(
-        State(api.clone()),
-        HeaderMap::new(),
-        Json(CreateWorkspaceWorkerRequest {
+        State(api.clone()), HeaderMap::new(), Json(delivery_create_request(subject, profile)),
+    ).await.unwrap();
+    RuntimeWorkerRef::new(worker.runtime_id, worker.worker_id)
+}
+
+fn delivery_create_request(subject: Option<&str>, profile: &str) -> CreateWorkspaceWorkerRequest {
+    CreateWorkspaceWorkerRequest {
            runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.into(),
            display_name: "Artifact consumer".into(),
            singleton_key: None,
-            profile: Some("builtin:companion".into()),
+            profile: Some(profile.into()),
            ticket_assignment: None,
            initial_submit: vec![],
            workdir_attachments: vec![],
@@ -100,11 +108,7 @@ async fn delivery_worker(api: &WorkspaceApi, subject: Option<&str>) -> RuntimeWo
            },
            control_operation_id: None,
            execution_grant: None,
-        }),
-    )
-    .await
-    .unwrap();
-    RuntimeWorkerRef::new(worker.runtime_id, worker.worker_id)
+    }
 }

 fn configure_delivery_plan(
@@ -24,7 +24,7 @@ impl WorkspaceApi {
        )
    }

-    fn plugin_reserved_identity(
+    pub(super) fn plugin_reserved_identity(
        &self,
        worker: &RuntimeWorkerRef,
        expected_fingerprint: Option<&str>,
@@ -86,7 +86,7 @@ impl Fixture {
        let upload = api
            .config_store
            .plugin_upload(
-                &directory.path().join("blobs"),
+                &ServerConfig::server_data_root_for_data_dir(&api.config.data_root).join("plugin-artifacts"),
                TEST_WORKSPACE_ID,
                "fixture-owner",
                &package.bytes,
@@ -0,0 +1,193 @@
+// Included in the delivery suite; the Runtime execution boundary is scripted.
+// These cases prove Server reservation/pin atomicity, not Component execution.
+#[tokio::test]
+async fn worker_create_seals_plan_and_replay_retains_exact_snapshot_after_revoke() {
+    let f = DeliveryFixture::new(false).await;
+    let runtime = WorkdirlessFixtureRuntime::default();
+    f.api.runtime.register_or_replace(runtime.clone());
+    let worker = delivery_worker_profile(&f.api, None, "project:delivery").await;
+    let saved = f
+        .api
+        .config_store
+        .with_conn(|conn| {
+            crate::plugin_plan::worker_launch_plan(
+                conn,
+                TEST_WORKSPACE_ID,
+                &worker.runtime_id,
+                &worker.worker_id,
+            )
+        })
+        .unwrap()
+        .unwrap();
+    assert_eq!(saved.pin_state, PlanPinState::Active);
+    assert_eq!(saved.plan.plugins(), f.saved.plan.plugins());
+    assert_eq!(saved.plan.execution().worker_id, worker.worker_id);
+    let (allocation, request_fingerprint): (String, String) = f.api.config_store.with_conn(|conn| {
+        Ok(conn.query_row("SELECT allocation_key,request_fingerprint FROM worker_create_reservations WHERE workspace_id=?1 AND worker_id=?2", rusqlite::params![TEST_WORKSPACE_ID, worker.worker_id], |row| Ok((row.get(0)?,row.get(1)?)))?)
+    }).unwrap();
+    let settings = f
+        .api
+        .config_store
+        .get_workspace_memory_settings(TEST_WORKSPACE_ID)
+        .unwrap();
+    let reservation = f
+        .api
+        .config_store
+        .reserve_worker_create(
+            TEST_WORKSPACE_ID,
+            &worker.runtime_id,
+            &allocation,
+            &request_fingerprint,
+            None,
+            &settings,
+        )
+        .unwrap();
+    assert!(
+        f.api
+            .config_store
+            .reserve_worker_create(
+                TEST_WORKSPACE_ID,
+                &worker.runtime_id,
+                &allocation,
+                "different-plan",
+                None,
+                &settings
+            )
+            .is_err()
+    );
+    let request = runtime.spawn_requests().pop().unwrap();
+    f.api
+        .config_store
+        .plugin_revoke(
+            TEST_WORKSPACE_ID,
+            "delivery-host",
+            &f.request.installation_id,
+            &PluginRevokeRequest {
+                expected_last_operation: f.installed.operation_id.clone(),
+                operation_key: "revoke-launch".into(),
+                artifact_digest: f.request.artifact_digest.clone(),
+                reason: "withdraw".into(),
+            },
+        )
+        .unwrap();
+    let binding = f
+        .api
+        .plugin_launch_binding(&worker.runtime_id, &request, &reservation)
+        .unwrap();
+    let replay = binding.plugin_execution.unwrap();
+    assert_eq!(replay.plan_id, saved.plan_id);
+    assert_eq!(replay.fingerprint, saved.fingerprint);
+    assert_eq!(replay.plan, saved.plan);
+    let local = f.api.plugin_artifact_source(worker.clone()).await.unwrap();
+    let pin = &replay.plan.plugins()[0];
+    assert!(
+        local
+            .authorize(&PluginArtifactDeliveryRequest {
+                plan_id: replay.plan_id.clone(),
+                plan_fingerprint: replay.fingerprint,
+                installation_id: pin.installation_id().into(),
+                artifact_digest: pin.artifact_digest().into()
+            })
+            .await
+            .is_err()
+    );
+    f.api
+        .store
+        .delete_worker_registry(TEST_WORKSPACE_ID, &worker)
+        .unwrap();
+    assert_eq!(
+        f.api
+            .config_store
+            .plugin_plan_load(TEST_WORKSPACE_ID, &saved.plan_id)
+            .unwrap()
+            .pin_state,
+        PlanPinState::Released
+    );
+}
+
+#[tokio::test]
+async fn rejected_worker_create_releases_sealed_plugin_pins_without_registry_publication() {
+    let f = DeliveryFixture::new(false).await;
+    let runtime = WorkdirlessFixtureRuntime::default();
+    runtime.reject_next_spawn();
+    f.api.runtime.register_or_replace(runtime);
+    assert!(
+        create_workspace_worker(
+            State(f.api.clone()),
+            HeaderMap::new(),
+            Json(delivery_create_request(None, "project:delivery"))
+        )
+        .await
+        .is_err()
+    );
+    let plans: Vec<(String,String)> = f.api.config_store.with_conn(|conn| {
+        let mut statement = conn.prepare("SELECT plan_id,pin_state FROM plugin_execution_plans WHERE workspace_id=?1 AND actor=?2")?;
+        Ok(statement.query_map(rusqlite::params![TEST_WORKSPACE_ID, crate::plugin_plan::WORKER_LAUNCH_ACTOR], |row| Ok((row.get(0)?,row.get(1)?)))?.collect::<std::result::Result<Vec<_>,_>>()?)
+    }).unwrap();
+    assert_eq!(plans.len(), 1);
+    assert_eq!(plans[0].1, "released");
+    let saved = f
+        .api
+        .config_store
+        .plugin_plan_load(TEST_WORKSPACE_ID, &plans[0].0)
+        .unwrap();
+    let worker = RuntimeWorkerRef::new(
+        saved.plan.execution().runtime_id.clone(),
+        saved.plan.execution().worker_id.clone(),
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn empty_plugin_selection_is_sealed_and_retained_on_exact_replay() {
+    let f = DeliveryFixture::new(false).await;
+    let runtime = WorkdirlessFixtureRuntime::default();
+    f.api.runtime.register_or_replace(runtime.clone());
+    let worker = delivery_worker_profile(&f.api, None, "builtin:companion").await;
+    let selected = f
+        .api
+        .config_store
+        .with_conn(|conn| {
+            crate::plugin_plan::worker_launch_selection(
+                conn,
+                TEST_WORKSPACE_ID,
+                &worker.runtime_id,
+                &worker.worker_id,
+            )
+        })
+        .unwrap();
+    assert!(matches!(selected, Some(None)));
+    let (allocation, request_fingerprint): (String, String) = f.api.config_store.with_conn(|conn| {
+        Ok(conn.query_row("SELECT allocation_key,request_fingerprint FROM worker_create_reservations WHERE workspace_id=?1 AND worker_id=?2", rusqlite::params![TEST_WORKSPACE_ID, worker.worker_id], |row| Ok((row.get(0)?,row.get(1)?)))?)
+    }).unwrap();
+    let settings = f
+        .api
+        .config_store
+        .get_workspace_memory_settings(TEST_WORKSPACE_ID)
+        .unwrap();
+    let reservation = f
+        .api
+        .config_store
+        .reserve_worker_create(
+            TEST_WORKSPACE_ID,
+            &worker.runtime_id,
+            &allocation,
+            &request_fingerprint,
+            None,
+            &settings,
+        )
+        .unwrap();
+    let request = runtime.spawn_requests().pop().unwrap();
+    let replay = f
+        .api
+        .plugin_launch_binding(&worker.runtime_id, &request, &reservation)
+        .unwrap();
+    assert!(replay.plugin_execution.is_none());
+    assert_eq!(replay.create_fingerprint, reservation.create_fingerprint);
+}
@@ -0,0 +1,172 @@
+//! Embedded composition keeps only a weak Server view in the Runtime factory.
+//! The outer WorkspaceApi owns the strong view; the inner view does not retain
+//! itself. This avoids Runtime -> WorkspaceApi -> Runtime strong cycles
+//! while retaining the same proof verifier, plan owner and artifact source.
+use super::*;
+use worker_runtime::plugin::{PluginRuntimeProviders, PluginRuntimeServices};
+
+use crate::plugin_plan::{WORKER_LAUNCH_ACTOR as LAUNCH_ACTOR, worker_launch_key as launch_key};
+
+impl WorkspaceApi {
+    /// Seal the reserved launch and its Plugin pins in one write transaction.
+    /// No external dispatch precedes this commit. Replays load immutable input;
+    /// they do not resolve changed Workspace configuration or reacquire pins.
+    pub(super) fn plugin_launch_binding(
+        &self,
+        runtime: &str,
+        request: &WorkerSpawnRequest,
+        reservation: &crate::store::WorkerCreateReservation,
+    ) -> Result<WorkerCreateBinding> {
+        use rusqlite::params;
+        let prior = self.config_store.with_conn(|conn| {
+            crate::plugin_plan::worker_launch_selection(
+                conn,
+                self.workspace_id(),
+                runtime,
+                &reservation.worker_id.to_string(),
+            )
+        })?;
+        if let Some(prior) = prior {
+            let snapshot = prior
+                .map(|saved| {
+                    if saved.pin_state == crate::plugin_plan::PlanPinState::Released {
+                        return Err(Error::WorkspacePermissionDenied(
+                            "Plugin launch released".into(),
+                        ));
+                    }
+                    Ok(runtime_api::PluginExecutionSnapshot {
+                        plan_id: saved.plan_id,
+                        fingerprint: saved.fingerprint,
+                        plan: saved.plan,
+                    })
+                })
+                .transpose()?;
+            let fingerprint = self.config_store.with_conn(|conn| Ok(conn.query_row("SELECT create_fingerprint FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",
+                params![self.workspace_id(), runtime, reservation.worker_id.to_string()], |row| row.get::<_,String>(0))?))?;
+            return Ok(WorkerCreateBinding {
+                worker_id: reservation.worker_id,
+                create_fingerprint: fingerprint,
+                plugin_execution: snapshot,
+            });
+        }
+        let worker = RuntimeWorkerRef::new(runtime, reservation.worker_id.to_string());
+        if !self
+            .store
+            .has_active_worker_create_reservation(self.workspace_id(), &worker)?
+        {
+            return Ok(WorkerCreateBinding {
+                worker_id: reservation.worker_id,
+                create_fingerprint: reservation.create_fingerprint.clone(),
+                plugin_execution: None,
+            });
+        }
+        let selector = match &request.profile {
+            worker_runtime::catalog::ProfileSelector::Builtin(s)
+            | worker_runtime::catalog::ProfileSelector::Named(s) => s,
+        };
+        let manifest = match &request.resolved_config_bundle {
+            Some(bundle) => crate::profile_settings::resolve_profile_manifest_from_config_bundle(
+                bundle, selector,
+            )?,
+            None => crate::profile_settings::builtin_profile_source_archive(&request.profile)
+                .map_err(Error::Config)?
+                .verify()
+                .map_err(|_| Error::Config("Plugin Profile archive invalid".into()))?
+                .resolve_profile(selector, Path::new("/"), "plugin-launch")
+                .map_err(|_| Error::Config("Plugin Profile unavailable".into()))?,
+        };
+        if manifest.plugins.selections.is_empty() {
+            return self.config_store.with_immediate_transaction(|tx| {
+                tx.execute("INSERT INTO worker_plugin_launches(workspace_id,runtime_id,worker_id,plan_id) SELECT workspace_id,runtime_id,worker_id,NULL FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND create_fingerprint=?4 AND state='reserved' ON CONFLICT(workspace_id,worker_id) DO NOTHING",
+                    params![self.workspace_id(), runtime, reservation.worker_id.to_string(), reservation.create_fingerprint])?;
+                let selection = crate::plugin_plan::worker_launch_selection(tx, self.workspace_id(), runtime, &reservation.worker_id.to_string())?.ok_or_else(|| Error::WorkspaceConfigConflict("Plugin launch reservation changed".into()))?;
+                let plugin_execution = selection.map(|saved| runtime_api::PluginExecutionSnapshot { plan_id: saved.plan_id, fingerprint: saved.fingerprint, plan: saved.plan });
+                let fingerprint = tx.query_row("SELECT create_fingerprint FROM worker_create_reservations WHERE workspace_id=?1 AND worker_id=?2", params![self.workspace_id(), reservation.worker_id.to_string()], |row| row.get(0))?;
+                Ok(WorkerCreateBinding { worker_id: reservation.worker_id, create_fingerprint: fingerprint, plugin_execution })
+            });
+        }
+        let execution = self.plugin_reserved_execution_identity(
+            runtime,
+            reservation,
+            request.resolved_execution_grant.as_ref(),
+        )?;
+        let key = launch_key(runtime, &reservation.worker_id.to_string());
+        self.config_store.with_immediate_transaction(|tx| {
+            if let Some(saved) = crate::plugin_plan::worker_launch_selection(tx, self.workspace_id(), runtime, &reservation.worker_id.to_string())? {
+                let plugin_execution = saved.map(|saved| runtime_api::PluginExecutionSnapshot { plan_id: saved.plan_id, fingerprint: saved.fingerprint, plan: saved.plan });
+                let fingerprint = tx.query_row("SELECT create_fingerprint FROM worker_create_reservations WHERE workspace_id=?1 AND worker_id=?2", params![self.workspace_id(), reservation.worker_id.to_string()], |row| row.get(0))?;
+                return Ok(WorkerCreateBinding { worker_id: reservation.worker_id, create_fingerprint: fingerprint, plugin_execution });
+            }
+            let snapshot = {
+                let state = crate::config_source::load_state(tx, self.workspace_id())?.ok_or_else(|| Error::Config("Plugin config unavailable".into()))?;
+                let plan = crate::plugin_settings::resolve_worker_plugin_plan(tx, self.workspace_id(), &state.snapshot.digest, selector, execution)?;
+                let requested: std::collections::BTreeMap<_, std::collections::BTreeSet<_>> = manifest.plugins.selections.iter().map(|p| (p.installation_id.as_str(), p.tools.iter().map(String::as_str).collect())).collect();
+                let resolved: std::collections::BTreeMap<_, std::collections::BTreeSet<_>> = plan.plugins().iter().map(|p| (p.installation_id(), p.projection().tools.iter().map(String::as_str).collect())).collect();
+                if requested != resolved {
+                    return Err(Error::WorkspaceConfigConflict("Plugin Profile changed during launch".into()));
+                }
+                let saved = crate::plugin_plan::plugin_plan_create(tx, LAUNCH_ACTOR, &key, &plan)?;
+                let fingerprint = config_source::digest_bytes(&serde_json::to_vec(&(&reservation.create_fingerprint, &saved.fingerprint)).map_err(|_| Error::Config("Plugin launch encoding failed".into()))?);
+                let updated = tx.execute("UPDATE worker_create_reservations SET create_fingerprint=?1 WHERE workspace_id=?2 AND runtime_id=?3 AND worker_id=?4 AND create_fingerprint=?5 AND state='reserved'",
+                    params![fingerprint, self.workspace_id(), runtime, reservation.worker_id.to_string(), reservation.create_fingerprint])?;
+                if updated != 1 { return Err(Error::WorkspaceConfigConflict("Plugin launch reservation changed".into())); }
+                tx.execute("INSERT INTO worker_plugin_launches(workspace_id,runtime_id,worker_id,plan_id) VALUES(?1,?2,?3,?4)", params![self.workspace_id(), runtime, reservation.worker_id.to_string(), saved.plan_id])?;
+                runtime_api::PluginExecutionSnapshot { plan_id: saved.plan_id, fingerprint: saved.fingerprint, plan: saved.plan }
+            };
+            let fingerprint: String = tx.query_row("SELECT create_fingerprint FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state='reserved'",
+                params![self.workspace_id(), runtime, reservation.worker_id.to_string()], |row| row.get(0))?;
+            Ok(WorkerCreateBinding { worker_id: reservation.worker_id, create_fingerprint: fingerprint, plugin_execution: Some(snapshot) })
+        })
+    }
+}
+
+#[derive(Default)]
+pub(super) struct EmbeddedPluginRuntimeProviders {
+    api: std::sync::Mutex<Option<std::sync::Weak<WorkspaceApi>>>,
+}
+impl EmbeddedPluginRuntimeProviders {
+    pub(super) fn bind(&self, api: &Arc<WorkspaceApi>) {
+        *self.api.lock().expect("Plugin Host binding lock") = Some(Arc::downgrade(api));
+    }
+}
+#[async_trait::async_trait]
+impl PluginRuntimeProviders for EmbeddedPluginRuntimeProviders {
+    async fn connect(
+        &self,
+        workspace: &str,
+        worker: &worker_runtime::identity::WorkerRef,
+    ) -> std::result::Result<PluginRuntimeServices, String> {
+        let api = self
+            .api
+            .lock()
+            .map_err(|_| "Plugin Host binding unavailable")?
+            .as_ref()
+            .and_then(std::sync::Weak::upgrade)
+            .ok_or("Plugin Host binding unavailable")?;
+        if workspace != api.workspace_id() {
+            return Err("Plugin Workspace mismatch".into());
+        }
+        let worker = RuntimeWorkerRef::new(EMBEDDED_RUNTIME_ID, worker.worker_id.to_string());
+        let artifacts = api
+            .plugin_artifact_source(worker.clone())
+            .await
+            .map_err(|_| "Plugin artifact source unavailable")?;
+        let audience = embedded_runtime_request_audience(&api.config)
+            .map_err(|_| "Plugin Host audience unavailable")?;
+        let client = worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new(
+            workspace,
+            audience.clone(),
+            EMBEDDED_RUNTIME_ID,
+        )
+        .with_runtime_request_source(&EMBEDDED_RUNTIME_REQUEST_IDENTITY, audience);
+        let owner = plugin_host::EmbeddedPluginHostOwner::authenticated(
+            api,
+            workspace.into(),
+            client.plugin_host_authorizer(worker.worker_id),
+        );
+        Ok(PluginRuntimeServices {
+            artifacts: Arc::new(artifacts),
+            owner: Arc::new(owner),
+        })
+    }
+}
@@ -0,0 +1,444 @@
+// Included by server_plugin_host_tests.rs. These tests connect the production
+// pinned Runtime bridge, Feature planner/installer, ToolRegistry, Engine and
+// filesystem Session history to the real catalog/config/plan, signed Host owner,
+// artifact delivery, Wasmtime HTTPS Component, TLS peer and SQLite receipts.
+// Only the external model and the peer's DNS/socket/trust are substituted.
+// Fixture's Worker creation still uses WorkdirlessFixtureRuntime: this is NOT a
+// complete create-controller/restore vertical or a Runtime lifecycle test.
+// Run: cargo test -p yoi-workspace-server --lib plugin_runtime_tests
+// Prerequisite: openssl, as required by the shared Peer fixture.
+mod plugin_runtime_tests {
+    use super::*;
+    use agen::llm_client::event::{Event as ModelEvent, ResponseStatus, StatusEvent};
+    use agen::llm_client::{ClientError, LlmClient, Request as ModelRequest};
+    use agen::tool::ToolExecutionContext;
+    use session_store::Store;
+    use sha2::{Digest, Sha256};
+    use std::sync::Mutex;
+    use worker::feature::{FeatureId, FeatureInstallPlan, FeatureRegistryBuilder};
+    use worker_runtime::identity::WorkerRef;
+    use worker_runtime::plugin::{
+        PinnedPluginToolRuntime, PluginRuntimeProviders, PluginRuntimeServices, PluginToolRuntime,
+    };
+    use yoi_plugin_runtime::delivery::ArtifactSource;
+
+    // This factory binds actual adapters to the fixture's exact authenticated
+    // Worker; it supplies neither guest behavior nor admission responses.
+    struct RuntimeProviders {
+        worker: WorkerRef,
+        artifacts: Arc<dyn ArtifactSource>,
+        owner: Arc<dyn PluginHostOwner>,
+    }
+
+    #[async_trait::async_trait]
+    impl PluginRuntimeProviders for RuntimeProviders {
+        async fn connect(
+            &self,
+            workspace: &str,
+            worker: &WorkerRef,
+        ) -> std::result::Result<PluginRuntimeServices, String> {
+            assert_eq!(workspace, TEST_WORKSPACE_ID);
+            assert_eq!(worker, &self.worker);
+            Ok(PluginRuntimeServices {
+                artifacts: self.artifacts.clone(),
+                owner: self.owner.clone(),
+            })
+        }
+    }
+
+    // Artifact delivery needs its own registered HTTP audience. The Host suite's
+    // SignedAuthorizer deliberately uses "server-test", so keep its existing
+    // owner helper and use a separately URL-bound router for delivery. Both
+    // adapters reach the same real Server stores and execution identity owner.
+    async fn artifact_source(
+        f: &Fixture,
+        remote: bool,
+    ) -> (Arc<dyn ArtifactSource>, Option<ServerTask>) {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}", listener.local_addr().unwrap());
+        let mut api = f.api.clone();
+        api.config.backend_base_url = Some(url.clone());
+        let signing = api
+            .signing_identities
+            .provision_existing(TEST_WORKSPACE_ID, &test_owner_actor().account_id)
+            .unwrap();
+        let bundle = project_workspace_signing_identity(&api, signing)
+            .unwrap()
+            .public_bundle
+            .unwrap();
+        if !remote {
+            return (
+                Arc::new(api.plugin_artifact_source(f.source.clone()).await.unwrap()),
+                None,
+            );
+        }
+        let (_, registration) = worker_runtime::workspace_issuer::add_workspace_issuer_trust(
+            &mut Vec::new(),
+            bundle,
+            worker_runtime::auth::unix_now_seconds() as i64,
+        )
+        .unwrap();
+        let client = worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new(
+            TEST_WORKSPACE_ID,
+            &url,
+            &f.source.runtime_id,
+        )
+        .with_runtime_request_source(&f.identity, &url);
+        let source =
+            worker_runtime::workspace_request::plugin_delivery::RemotePluginArtifactSource::new(
+                client,
+                f.source.worker_id.clone(),
+                Arc::new(move || Ok(registration.clone())),
+            )
+            .unwrap();
+        let router = build_router(api);
+        let server = ServerTask(tokio::spawn(async move {
+            axum::serve(listener, router).await.unwrap();
+        }));
+        (Arc::new(source), Some(server))
+    }
+
+    fn allow_artifact_read(f: &Fixture) {
+        // Preserve the shared Host policy and grant the exact additional
+        // operation that real artifact delivery checks; no broad Feature grant.
+        let store = open_subjektiv_store(&f.api).unwrap();
+        let previous = store.subject_permissions(&f.subject).unwrap().unwrap();
+        let mut execution = previous.execution;
+        execution.push(ScopedPermission {
+            operation: "plugin.artifact.read".into(),
+            resource: PermissionResource::PluginInstallation {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+                installation_id: f.installed.installation.installation_id.clone(),
+            },
+        });
+        store
+            .set_subject_permissions(
+                &f.subject,
+                &SubjectPermissionsUpdate {
+                    operation_id: Uuid::now_v7().to_string(),
+                    expected_change_id: Some(previous.change_id),
+                    execution,
+                    delegation: previous.delegation,
+                },
+            )
+            .unwrap();
+    }
+
+    #[derive(Clone)]
+    struct Model {
+        tool_name: String,
+        requests: Arc<Mutex<Vec<ModelRequest>>>,
+    }
+
+    #[async_trait::async_trait]
+    impl LlmClient for Model {
+        fn clone_boxed(&self) -> Box<dyn LlmClient> {
+            Box::new(self.clone())
+        }
+
+        async fn stream(
+            &self,
+            request: ModelRequest,
+        ) -> std::result::Result<agen::llm_client::client::ResponseStream, ClientError> {
+            let turn = {
+                let mut requests = self.requests.lock().unwrap();
+                let turn = requests.len();
+                requests.push(request);
+                turn
+            };
+            let events = match turn {
+                0 => vec![
+                    ModelEvent::tool_use_start(0, "https-call", &self.tool_name),
+                    ModelEvent::tool_input_delta(0, "{}"),
+                    ModelEvent::tool_use_stop(0),
+                ],
+                1 => vec![
+                    ModelEvent::text_block_start(0),
+                    ModelEvent::text_delta(0, "Observed the HTTPS tool result."),
+                    ModelEvent::text_block_stop(0, None),
+                ],
+                _ => panic!("unexpected extra model turn"),
+            };
+            Ok(Box::pin(futures::stream::iter(
+                events
+                    .into_iter()
+                    .chain([ModelEvent::Status(StatusEvent {
+                        status: ResponseStatus::Completed,
+                    })])
+                    .map(Ok),
+            )))
+        }
+    }
+
+    fn assert_https_history(items: &[agen::Item], public_name: &str) {
+        let calls = items
+            .iter()
+            .filter_map(|item| match item {
+                agen::Item::ToolCall {
+                    call_id,
+                    name,
+                    arguments,
+                    ..
+                } => Some((call_id, name, arguments)),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(calls.len(), 1);
+        assert_eq!(calls[0].0.as_str(), "https-call");
+        assert_eq!(calls[0].1, public_name);
+        assert_eq!(
+            serde_json::from_str::<Value>(calls[0].2).unwrap(),
+            json!({})
+        );
+        let results = items
+            .iter()
+            .filter_map(|item| match item {
+                agen::Item::ToolResult {
+                    call_id,
+                    summary,
+                    content,
+                    disposition,
+                    is_error,
+                    attachments,
+                    ..
+                } => Some((
+                    call_id,
+                    summary,
+                    content,
+                    disposition,
+                    is_error,
+                    attachments,
+                )),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(results.len(), 1);
+        let (call, summary, content, disposition, is_error, attachments) = results[0];
+        assert_eq!(call, calls[0].0);
+        assert_eq!(*disposition, agen::ToolResultDisposition::Success);
+        assert!(!*is_error);
+        assert!(attachments.is_empty());
+        assert_eq!(summary, "HTTP response observed");
+        let output: Value = serde_json::from_str(content.as_deref().unwrap()).unwrap();
+        assert_eq!(output["structured"], json!({"status":503,"body_bytes":9}));
+    }
+
+    async fn https_through_runtime_and_history(remote: bool) {
+        let mut peer = Peer::new(HOST, vec![response(503, &[], b"peer data")]).await;
+        let mut f = Fixture::new(Some(peer_binding(&peer))).await;
+        f.api.plugin_https_transport = Some(Arc::new(peer.transport()));
+        allow_artifact_read(&f);
+        let (artifacts, mut delivery_server) = artifact_source(&f, remote).await;
+        let (owner, mut owner_server) = if remote {
+            let (owner, server) =
+                remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await;
+            (owner, Some(server))
+        } else {
+            // Fixture::owner constructs EmbeddedPluginHostOwner::authenticated
+            // with SignedAuthorizer; it never injects a trusted request context.
+            (f.owner(Some(peer.transport())).await, None)
+        };
+        let worker_ref = WorkerRef::try_from(&f.source).unwrap();
+        let providers = Arc::new(RuntimeProviders {
+            worker: worker_ref.clone(),
+            artifacts,
+            owner,
+        });
+        let bridge =
+            PinnedPluginToolRuntime::new(providers, &f._directory.path().join("runtime-cache"))
+                .unwrap();
+        let public_name = format!("{}__{}", f.installed.installation.installation_id, f.tool);
+        let model = Model {
+            tool_name: public_name.clone(),
+            requests: Arc::default(),
+        };
+        let manifest = worker::WorkerManifest::from_toml(&format!(
+            "[worker]\nname=\"{}\"\n[model]\nscheme=\"anthropic\"\nmodel_id=\"scripted\"\n[engine]\n[scope]\nallow=[]\n",
+            f.source.worker_id
+        ))
+        .unwrap();
+        let store_root = f._directory.path().join("tool-history");
+        let mut host = worker::Worker::new(
+            manifest,
+            agen::Engine::new_annotated(model.clone())
+                .system_prompt("Use the selected HTTPS tool."),
+            session_store::FsStore::new(&store_root).unwrap(),
+            worker::WorkerWorkspaceContext::unavailable(
+                Some(worker::WorkspaceId::new(TEST_WORKSPACE_ID).unwrap()),
+                "unrelated Workspace operations are not bound by this fixture",
+            ),
+            worker::WorkerFilesystemAuthority::None,
+            manifest::Scope::empty(),
+        )
+        .await
+        .unwrap();
+        let session_id = host.session_id();
+        let snapshot = runtime_api::PluginExecutionSnapshot {
+            plan_id: f.plan.plan_id.clone(),
+            fingerprint: f.plan.fingerprint.clone(),
+            plan: f.plan.plan.clone(),
+        };
+        let feature = bridge
+            .prepare(&worker_ref, &snapshot, &session_id.to_string())
+            .await
+            .unwrap();
+        assert_eq!(
+            f.count("plugin_invocations"),
+            0,
+            "preparation is not a Tool call"
+        );
+        assert_eq!(f.count("plugin_effect_receipts"), 0);
+        let registry = FeatureRegistryBuilder::new().with_module(feature);
+        let install_plan: FeatureInstallPlan = registry.plan().unwrap();
+        assert_eq!(
+            install_plan.ordered_features(),
+            &[FeatureId::new("plugin:tools").unwrap()]
+        );
+        let report = host.install_features(registry);
+        assert!(!report.has_errors(), "{}", report.error_message());
+        let tools = host.engine().tool_server_handle();
+        tools.flush_pending();
+        let definitions = tools.tool_definitions_sorted();
+        assert_eq!(
+            definitions
+                .iter()
+                .map(|tool| tool.name.as_str())
+                .collect::<Vec<_>>(),
+            [public_name.as_str()]
+        );
+        let (metadata, _) = tools.get_tool(&public_name).unwrap();
+        assert_eq!(
+            metadata.input_schema,
+            f.installed.installation.artifacts[0].tools[0].input_schema
+        );
+        assert_eq!(
+            metadata.origin.unwrap().plugin_ref,
+            f.installed.installation.installation_id
+        );
+        assert!(
+            tools.get_tool(f.tool).is_none(),
+            "bare guest names are not public tools"
+        );
+        host.wire_history_persistence();
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(30),
+            host.run(vec![protocol::Segment::Text {
+                content: "Check the selected service status.".into(),
+            }]),
+        )
+        .await
+        .expect("HTTPS Tool/Engine must finish within the fixture deadline")
+        .unwrap();
+        assert!(matches!(result, worker::WorkerRunResult::Finished));
+        assert_https_history(&host.history(), &public_name);
+        {
+            let requests = model.requests.lock().unwrap();
+            assert_eq!(requests.len(), 2);
+            assert_eq!(
+                requests[0]
+                    .tools
+                    .iter()
+                    .map(|tool| tool.name.as_str())
+                    .collect::<Vec<_>>(),
+                [public_name.as_str()]
+            );
+            assert_https_history(&requests[1].items, &public_name);
+        }
+        assert!(
+            String::from_utf8(peer.observed().await)
+                .unwrap()
+                .starts_with("GET /status HTTP/1.1\r\n")
+        );
+        assert_eq!(peer.connection_count(), 1);
+        assert_eq!(f.count("plugin_invocations"), 1);
+        assert_eq!(f.count("plugin_effect_receipts"), 1);
+
+        // Reopen the actual filesystem Store and reconstruct canonical history,
+        // rather than treating the model's captured request as persistence proof.
+        let reopened = session_store::FsStore::new(&store_root).unwrap();
+        let segments = reopened.list_segments(session_id).unwrap();
+        assert_eq!(segments.len(), 1);
+        let saved = session_store::restore(&reopened, session_id, segments[0]).unwrap();
+        assert_https_history(&saved.history, &public_name);
+        let identity_bytes = serde_json::to_vec(&(session_id.to_string(), "https-call")).unwrap();
+        let consumer_id = format!(
+            "tool-{}",
+            Sha256::digest(identity_bytes)
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect::<String>()
+        );
+        let (consumer, state, status, body): (String, String, i64, Vec<u8>) = f
+            .api
+            .config_store
+            .with_conn(|conn| {
+                Ok(conn.query_row(
+                    "SELECT i.consumer_json,r.state,r.response_status,r.response_body \
+                     FROM plugin_invocations i JOIN plugin_effect_receipts r \
+                     USING(workspace_id,invocation_id) \
+                     WHERE i.workspace_id=?1 AND i.plan_id=?2 AND i.worker_id=?3 \
+                     AND i.installation_id=?4 AND r.effect_key='status-v1'",
+                    rusqlite::params![
+                        TEST_WORKSPACE_ID,
+                        f.plan.plan_id,
+                        f.source.worker_id,
+                        f.installed.installation.installation_id
+                    ],
+                    |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
+                )?)
+            })
+            .unwrap();
+        assert_eq!(
+            serde_json::from_str::<PluginConsumer>(&consumer).unwrap(),
+            PluginConsumer::ToolCall {
+                tool_call_id: consumer_id,
+                tool_name: f.tool.into()
+            }
+        );
+        assert_eq!(state, "completed");
+        assert_eq!(status, 503);
+        assert_eq!(body, b"peer data");
+
+        // Current authority still governs a registered Tool and cached bytes.
+        // Denial must neither start a new invocation nor send another request.
+        f.deny();
+        assert!(
+            tools
+                .call_tool(
+                    &public_name,
+                    "{}",
+                    ToolExecutionContext::new("denied-call", "denied-batch", 0),
+                )
+                .await
+                .is_err()
+        );
+        assert!(
+            bridge
+                .prepare(&worker_ref, &snapshot, &session_id.to_string())
+                .await
+                .is_err()
+        );
+        assert_eq!(peer.connection_count(), 1);
+        assert_eq!(f.count("plugin_invocations"), 1);
+        assert_eq!(f.count("plugin_effect_receipts"), 1);
+        for server in [&mut owner_server, &mut delivery_server]
+            .into_iter()
+            .flatten()
+        {
+            server.0.abort();
+            let _ = (&mut server.0).await;
+        }
+        f.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn remote_pinned_https_tool_commits_normal_history_and_receipt_then_fences_live_denial() {
+        https_through_runtime_and_history(true).await;
+    }
+
+    #[tokio::test]
+    async fn embedded_pinned_https_tool_commits_normal_history_and_receipt_then_fences_live_denial()
+    {
+        https_through_runtime_and_history(false).await;
+    }
+}
@@ -129,7 +129,7 @@ pub use ticket::{
 };

 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 109;
+const LATEST_SCHEMA_VERSION: i64 = 110;
 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";
@@ -482,6 +482,11 @@ const MIGRATIONS: &[Migration] = &[
        name: "Durable Plugin Host invocations and HTTPS effect receipts",
        apply: migrate_plugin_receipts_v108_to_v109,
    },
+    Migration {
+        version: 110,
+        name: "Worker Plugin launch selection associations",
+        apply: migrate_plugin_launches_v109_to_v110,
+    },
 ];

 // Both schema-103 identities shipped. Keep the alternate implementation and accept
@@ -3729,8 +3734,10 @@ impl SqliteWorkspaceStore {
        workspace_id: &str,
        worker_id: WorkerId,
    ) -> Result<()> {
-        self.with_conn(|conn| {
-            let changed = conn.execute(
+        self.with_immediate_transaction(|tx| {
+            let runtime: String = tx.query_row("SELECT runtime_id FROM worker_create_reservations WHERE workspace_id=?1 AND worker_id=?2", params![workspace_id, worker_id.to_string()], |row| row.get(0))?;
+            crate::plugin_plan::transition_worker_launch_plan(tx, workspace_id, &runtime, &worker_id.to_string(), crate::plugin_plan::PlanPinState::Active)?;
+            let changed = tx.execute(
                "UPDATE worker_create_reservations \
                 SET state = 'created', updated_at = ?3 \
                 WHERE workspace_id = ?1 AND worker_id = ?2 AND state IN ('reserved', 'created')",
@@ -3822,6 +3829,7 @@ impl SqliteWorkspaceStore {
                    )));
                }
            }
+            crate::plugin_plan::transition_worker_launch_plan(&tx, workspace_id, &runtime_id, &worker_id.to_string(), crate::plugin_plan::PlanPinState::Released)?;
            let resource_id = worker_resource_id(&RuntimeWorkerRef::new(
                runtime_id,
                worker_id.to_string(),
@@ -3880,6 +3888,7 @@ impl SqliteWorkspaceStore {
                    worker_id
                )));
            }
+            crate::plugin_plan::transition_worker_launch_plan(&tx, workspace_id, &runtime_id, &worker_id.to_string(), crate::plugin_plan::PlanPinState::Released)?;
            let resource_id = worker_resource_id(&RuntimeWorkerRef::new(
                runtime_id,
                worker_id.to_string(),
@@ -9088,6 +9097,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                        chrono::Utc::now().to_rfc3339(),
                    ],
                )?;
+                crate::plugin_plan::transition_worker_launch_plan(&tx, workspace_id, &worker.runtime_id, &worker.worker_id, crate::plugin_plan::PlanPinState::Released)?;
                commit_worker_catalog_removal(&tx, workspace_id, worker)?
            } else if tx
                .query_row(
@@ -15856,6 +15866,23 @@ fn migrate_subject_assignments_v107_to_v108(conn: &Connection) -> Result<()> {
    Ok(())
 }

+fn migrate_plugin_launches_v109_to_v110(conn: &Connection) -> Result<()> {
+    if current_schema_version(conn)? != 109 {
+        return Err(Error::Store(
+            "Plugin launch migration requires schema 109".into(),
+        ));
+    }
+    verify_schema_history(conn, 109)?;
+    let tx = conn.unchecked_transaction()?;
+    tx.execute_batch(crate::plugin_plan::LAUNCH_SCHEMA_SQL)?;
+    tx.execute(
+        "INSERT INTO __yoi_schema_migrations(version,name) VALUES(110,?1)",
+        ["Worker Plugin launch selection associations"],
+    )?;
+    tx.commit()?;
+    Ok(())
+}
+
 fn migrate_plugin_receipts_v108_to_v109(conn: &Connection) -> Result<()> {
    if current_schema_version(conn)? != 108 {
        return Err(Error::Store(
@@ -15916,6 +15943,7 @@ fn create_workspace_schema_structure(conn: &Connection) -> Result<()> {
    conn.execute_batch(include_str!("passkey_results_schema.sql"))?;
    conn.execute_batch(crate::plugin_plan::SCHEMA_SQL)?;
    conn.execute_batch(crate::plugin_receipt::SCHEMA_SQL)?;
+    conn.execute_batch(crate::plugin_plan::LAUNCH_SCHEMA_SQL)?;
    Ok(())
 }

@@ -16959,6 +16987,7 @@ fn apply_migrations_with_secret_source(
    crate::plugin_catalog::verify(conn)?;
    crate::plugin_plan::verify_plan_schema(conn)?;
    crate::plugin_host::verify_receipt_schema(conn)?;
+    crate::plugin_plan::verify_launch_schema(conn)?;
    subject_assignment_migration::verify(conn)?;
    verify_workspace_runtime_binding_schema(conn)?;
    verify_workspace_runtime_verification_schema(conn)?;
@@ -17002,6 +17031,7 @@ mod tests {
    include!("job_notification_tests.rs");
    include!("implementation_result_tests.rs");
    include!("store_workspace_config_tests.rs");
+    include!("plugin_plan/launch_migration_tests.rs");

    fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {
        store.with_conn(|c| {
@@ -19659,6 +19689,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                WorkspaceSchemaMigrationStep {version:107, name:"Immutable Worker Plugin plans and retained pins".into()},
                WorkspaceSchemaMigrationStep {version:108, name:"Subject Ticket responsibility and current-body notification routing".into()},
                WorkspaceSchemaMigrationStep {version:109, name:"Durable Plugin Host invocations and HTTPS effect receipts".into()},
+                WorkspaceSchemaMigrationStep {version:110, name:"Worker Plugin launch selection associations".into()},
            ]
        );

@@ -19778,6 +19809,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
 (107, "Immutable Worker Plugin plans and retained pins".to_string()),
 (108, "Subject Ticket responsibility and current-body notification routing".to_string()),
 (109, "Durable Plugin Host invocations and HTTPS effect receipts".to_string()),
+(110, "Worker Plugin launch selection associations".to_string()),
                    ]
                );
                assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -23068,7 +23100,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                         (106, crate::plugin_catalog::MIGRATION_NAME.into()),
                         (107, "Immutable Worker Plugin plans and retained pins".into()),
                         (108, "Subject Ticket responsibility and current-body notification routing".into()),
-                         (109, "Durable Plugin Host invocations and HTTPS effect receipts".into())]);
+                         (109, "Durable Plugin Host invocations and HTTPS effect receipts".into()),
+                         (110, "Worker Plugin launch selection associations".into())]);
                for (table, rows) in unchanged.iter().zip(before.iter()) {
                    assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
                }