feat: complete Plugin controller lifecycle and owner diagnostics with real runtime coverage

crates/server-api/src/lib.rs

@@ -226,6 +226,7 @@ impl_openapi_schema!(
     PluginOperationResponse,
     PluginArtifactDeliveryRequest,
     PluginArtifactDeliveryMetadata,
+    PluginWorkerDiagnostics,
     SubjectPermissionsResponse,
     SubjectPermissionsMutationResponse,
     SubjectPermissionInterruption,
@@ -2973,6 +2974,15 @@ pub trait ServerApi {
         #[query] query: PluginListRequest,
     ) -> Result<PluginListResponse, RepositoryApiError>;
 
+    #[get("/api/w/{workspace_id}/plugins/worker-diagnostics/{runtime_id}/{worker_id}", status = 200, error_status = 400, additional_error_statuses = [401,403,404,503], bearer_auth = true, browser_auth = true)]
+    async fn plugin_worker_diagnostics(
+        &self,
+        #[extension] actor: RequestActor,
+        #[path] workspace_id: String,
+        #[path] runtime_id: String,
+        #[path] worker_id: String,
+    ) -> Result<PluginWorkerDiagnostics, RepositoryApiError>;
+
     #[get("/api/w/{workspace_id}/plugins/{installation_id}", status = 200, error_status = 400, additional_error_statuses = [401,403,404,409,413,429,503], bearer_auth = true, browser_auth = true)]
     async fn plugin_show(
         &self,

crates/server-api/src/plugin.rs

@@ -221,6 +221,126 @@ pub struct PluginOperationResponse {
     pub installation: PluginInstallation,
 }
 
+/// Point-in-time owner observation only, never a permission or execution lease.
+/// Deliberately excludes configuration, credentials and artifact locations.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct PluginWorkerDiagnostics {
+    pub workspace_id: String,
+    pub runtime_id: String,
+    pub worker_id: String,
+    pub runtime_limits: PluginRuntimeLimits,
+    pub selection: PluginWorkerDiagnosticSelection,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum PluginWorkerDiagnosticSelection {
+    /// No sealed launch selection exists; do not infer current Profile selection.
+    Unrecorded,
+    /// The owner sealed an explicitly empty selection (a NULL plan).
+    Empty,
+    Plan {
+        plan_id: String,
+        fingerprint: String,
+        pin_state: PluginDiagnosticPinState,
+        plugins: Vec<PluginWorkerDiagnostic>,
+    },
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(rename_all = "snake_case")]
+pub enum PluginDiagnosticPinState {
+    Pending,
+    Active,
+    Restorable,
+    Released,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(rename_all = "snake_case")]
+pub enum PluginDiagnosticSurface {
+    Tool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct PluginWorkerDiagnostic {
+    pub installation_id: String,
+    pub package_id: String,
+    pub package_version: String,
+    pub package_digest: String,
+    pub artifact_digest: String,
+    pub artifact_size: u32,
+    pub surfaces: Vec<PluginDiagnosticSurface>,
+    pub selected_tools: Vec<String>,
+    pub admission: PluginDiagnosticAdmission,
+    pub connections: Vec<PluginDiagnosticConnection>,
+}
+
+/// A successful observation is not retained admission authority.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum PluginDiagnosticAdmission {
+    Admissible,
+    Refused {
+        reason_code: crate::PluginHostErrorCode,
+    },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct PluginDiagnosticConnection {
+    pub slot: String,
+    pub effective_limits: PluginDiagnosticConnectionLimits,
+}
+
+/// Limits are absent when the current binding cannot be admitted; never fall
+/// back to a launch-time policy or a permissive default.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum PluginDiagnosticConnectionLimits {
+    Available {
+        timeout_ms: u32,
+        max_request_bytes: u32,
+        max_response_bytes: u32,
+        max_header_bytes: u32,
+    },
+    Unavailable {
+        reason_code: crate::PluginHostErrorCode,
+    },
+}
+
+/// Fixed Tool Runtime budgets, not usage counters or caller-selectable limits.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct PluginRuntimeLimits {
+    pub payload_bytes: u32,
+    pub config_bytes: u32,
+    pub memory_bytes: u32,
+    pub table_elements: u32,
+    pub instances: u32,
+    pub memories: u32,
+    pub tables: u32,
+    pub resources: u32,
+    pub fuel: u32,
+    pub wall_time_ms: u32,
+    pub host_operations: u32,
+    pub http_body_bytes: u32,
+    pub http_header_bytes: u32,
+    pub http_header_pairs: u32,
+    pub log_bytes: u32,
+}
+
 /// Namespace a byte cache by the authenticated Server origin and signing key.
 /// This is not a package hash or an authorization decision. Hosts derive the
 /// bundle from their current registered authority, never from a delivery reply.

crates/server-api/src/plugin_types.rs

@@ -48,6 +48,16 @@ pub fn plugin_api_typescript() -> String {
         PluginOperationResponse::decl(&config),
         PluginArtifactDeliveryRequest::decl(&config),
         PluginArtifactDeliveryMetadata::decl(&config),
+        PluginHostErrorCode::decl(&config),
+        PluginWorkerDiagnostics::decl(&config),
+        PluginWorkerDiagnosticSelection::decl(&config),
+        PluginDiagnosticPinState::decl(&config),
+        PluginDiagnosticSurface::decl(&config),
+        PluginWorkerDiagnostic::decl(&config),
+        PluginDiagnosticAdmission::decl(&config),
+        PluginDiagnosticConnection::decl(&config),
+        PluginDiagnosticConnectionLimits::decl(&config),
+        PluginRuntimeLimits::decl(&config),
     ];
     format!(
         "// Generated from server-api. Do not edit by hand.\n// Regenerate: cargo run -q -p server-api --features typescript --example generate_plugin_api_types > web/workspace/src/lib/generated/plugin-api.ts\n\n{}\n",

crates/worker-runtime/src/main.rs

@@ -275,7 +275,7 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
         );
         let plugins = worker_runtime::plugin::PinnedPluginToolRuntime::new(
             providers,
-            &runtime_store_dir.join("plugin-cache"),
+            &runtime_store_dir.with_extension("plugin-cache"),
         )
         .map_err(ProcessError::WorkerAdapter)?;
         factory = factory.with_plugin_runtime(Arc::new(plugins));

crates/worker-runtime/src/plugin.rs

@@ -196,7 +196,7 @@ pub mod remote {
             plugin_delivery::{CurrentArtifactServer, RemotePluginArtifactSource},
         },
     };
-    use std::{collections::HashMap, sync::Arc, time::Duration};
+    use std::{collections::HashMap, sync::Arc};
     pub type CurrentPluginServer = dyn Fn(&str) -> Result<WorkspaceIssuerTrustRecord, yoi_plugin_runtime::delivery::DeliveryError>
         + Send
         + Sync;
@@ -268,13 +268,7 @@ pub mod remote {
                 .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,
-                )
+                .plugin_host_client(worker.worker_id.to_string())
                 .map_err(|_| "Plugin Host client unavailable")?;
             let owner = Arc::new(RegisteredOwner {
                 source: source.clone(),

crates/worker-runtime/src/workspace_request.rs

@@ -191,6 +191,37 @@ impl RuntimeWorkspaceRequestClient {
         })
     }
 
+    pub(crate) fn plugin_host_client(
+        &self,
+        worker_id: String,
+    ) -> Result<
+        server_api::ServerApiClient<RuntimeWorkspaceRequestAuthorizer>,
+        RuntimeWorkspaceRequestError,
+    > {
+        let http = reqwest::Client::builder()
+            .redirect(reqwest::redirect::Policy::none())
+            .no_proxy()
+            .timeout(Duration::from_secs(30))
+            .build()
+            .map_err(RuntimeWorkspaceRequestError::transport)?;
+        let authorizer = 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(),
+        };
+        server_api::ServerApiClient::builder(&self.base_url)
+            .map_err(|error| RuntimeWorkspaceRequestError::InvalidRequest(error.to_string()))?
+            .client(http)
+            .authorizer(authorizer)
+            .request_timeout(Duration::from_secs(30))
+            .response_body_limit(yoi_plugin_runtime::owner::PLUGIN_HOST_RESPONSE_BODY_LIMIT)
+            .build()
+            .map_err(|error| RuntimeWorkspaceRequestError::InvalidRequest(error.to_string()))
+    }
+
     pub(crate) fn server_api_client(
         &self,
         permission: impl Into<String>,

crates/worker/src/plugin.rs

@@ -351,7 +351,24 @@ impl Tool for PluginTool {
                 &self.config_json,
             )
             .await
-            .map_err(|_| failure("Plugin invocation failed; no automatic retry"))?
+            .map_err(|error| {
+                let reason = match error {
+                    yoi_plugin_runtime::ExecutionError::Artifact(_) => "artifact",
+                    yoi_plugin_runtime::ExecutionError::UnsupportedComponent(_) => {
+                        "unsupported-component"
+                    }
+                    yoi_plugin_runtime::ExecutionError::InvalidInput(_) => "invalid-input",
+                    yoi_plugin_runtime::ExecutionError::InvalidOutput(_) => "invalid-output",
+                    yoi_plugin_runtime::ExecutionError::Cancelled => "cancelled",
+                    yoi_plugin_runtime::ExecutionError::Deadline => "deadline",
+                    yoi_plugin_runtime::ExecutionError::Host(_) => "host-refused-or-unavailable",
+                    yoi_plugin_runtime::ExecutionError::Guest { .. } => "guest-error",
+                    yoi_plugin_runtime::ExecutionError::Trap(_) => "trap-or-limit",
+                };
+                failure(&format!(
+                    "Plugin invocation failed ({reason}); no automatic retry"
+                ))
+            })?
             .output;
         // Fence again at the adapter publication boundary, not just on guest return.
         host.authorize(host.context(), Operation::Publish)

crates/workspace-server/src/assignment_tracking_tests.rs

@@ -517,7 +517,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
                         "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 retained.iter().zip(original.iter()) {

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

@@ -30,7 +30,7 @@ fn plugin_catalog_migration_matches_fresh_schema_from_oldest_retained_and_105()
             // Plugin catalog remains migration 106 even when startup applies later steps.
             assert_eq!(after.iter().filter(|(version, _)| *version == 106).cloned().collect::<Vec<_>>(),
                 vec![(106, crate::plugin_catalog::MIGRATION_NAME.into())]);
-            assert_eq!(after.last().unwrap(), &(109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
+            assert_eq!(after.last().unwrap(), &(110, "Worker Plugin launch selection associations".into()));
             assert_eq!(current_schema_version(c)?, LATEST_SCHEMA_VERSION);
             assert_eq!(c.query_row("SELECT display_name FROM workspaces WHERE workspace_id='plugin-workspace'",[],|r|r.get::<_,String>(0))?,"Plugins");
             assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
@@ -76,6 +76,7 @@ fn plugin_catalog_migration_marker_failure_rolls_back_then_retries_without_chang
                 109,
                 "Durable Plugin Host invocations and HTTPS effect receipts".into(),
             ));
+            expected.push((110, "Worker Plugin launch selection associations".into()));
             assert_eq!(workspace_schema_migration_history(c)?, expected);
             crate::plugin_catalog::verify(c)?;
             Ok(())

crates/workspace-server/src/plugin_host.rs

@@ -134,7 +134,7 @@ pub fn authorize(
     Ok(AdmittedPluginOperation { plan, pin, binding })
 }
 
-fn live_binding(
+pub(crate) fn live_binding(
     tx: &Transaction<'_>,
     workspace: &str,
     reference: &BindingPolicyReference,

crates/workspace-server/src/plugin_host_migration_tests.rs

@@ -12,16 +12,17 @@ fn plugin_receipt_tail_migrates_true_predecessor_reopens_and_matches_fresh_schem
     assert!(!table_exists(&conn, "plugin_invocations").unwrap());
     let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
     assert_eq!(plan.current_schema_version, 108);
-    assert_eq!(plan.migrations.iter().map(|step| step.version).collect::<Vec<_>>(), [109]);
+    assert_eq!(plan.migrations.iter().map(|step| step.version).collect::<Vec<_>>(), [109, 110]);
     assert_eq!(current_schema_version(&conn).unwrap(), 108);
     drop(conn);
     let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
     store.with_conn(|conn| {
         crate::plugin_host::verify_receipt_schema(conn)?;
-        assert_eq!(current_schema_version(conn)?, 109);
+        assert_eq!(current_schema_version(conn)?, 110);
         let upgraded = workspace_schema_migration_history(conn)?;
         assert_eq!(&upgraded[..history.len()], history.as_slice());
-        assert_eq!(upgraded.last().unwrap().1, "Durable Plugin Host invocations and HTTPS effect receipts");
+        assert_eq!(upgraded.iter().find(|(version, _)| *version == 109).unwrap().1, "Durable Plugin Host invocations and HTTPS effect receipts");
+        assert_eq!(upgraded.last().unwrap().1, "Worker Plugin launch selection associations");
         assert_eq!(conn.query_row("SELECT count(*) FROM workspaces WHERE workspace_id='space'", [], |row| row.get::<_, i64>(0))?, 1);
         assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| row.get::<_, i64>(0))?, 0);
         Ok(())

crates/workspace-server/src/plugin_https/test_support.rs

@@ -300,7 +300,11 @@ impl Peer {
     }
 
     pub(crate) async fn observed(&mut self) -> Vec<u8> {
-        tokio::time::timeout(Duration::from_secs(5), self.observations.recv())
+        self.observed_within(Duration::from_secs(5)).await
+    }
+
+    pub(crate) async fn observed_within(&mut self, deadline: Duration) -> Vec<u8> {
+        tokio::time::timeout(deadline, self.observations.recv())
             .await
             .unwrap()
             .unwrap()

crates/workspace-server/src/plugin_plan/launch_schema.sql

@@ -7,3 +7,11 @@ CREATE TABLE worker_plugin_launches (
     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)
 );
+CREATE TRIGGER worker_plugin_launch_immutable BEFORE UPDATE ON worker_plugin_launches
+BEGIN SELECT RAISE(ABORT,'Worker Plugin launch selection is immutable'); END;
+CREATE TRIGGER worker_plugin_launch_retained BEFORE DELETE ON worker_plugin_launches
+WHEN EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=OLD.workspace_id)
+BEGIN SELECT RAISE(ABORT,'Worker Plugin launch selection must be retained'); END;
+CREATE TRIGGER worker_plugin_launch_identity BEFORE INSERT ON worker_plugin_launches
+WHEN NOT EXISTS(SELECT 1 FROM worker_create_reservations WHERE workspace_id=NEW.workspace_id AND worker_id=NEW.worker_id AND runtime_id=NEW.runtime_id)
+BEGIN SELECT RAISE(ABORT,'Worker Plugin launch reservation identity mismatch'); END;

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

@@ -34,6 +34,7 @@ fn plugin_plan_migration_marker_failure_preserves_catalog_then_retries_and_match
                 109,
                 "Durable Plugin Host invocations and HTTPS effect receipts".into(),
             ));
+            expected.push((110, "Worker Plugin launch selection associations".into()));
             assert_eq!(workspace_schema_migration_history(conn)?, expected);
             subject_assignment_migration::verify(conn)?;
             crate::plugin_plan::verify_plan_schema(conn)?;

crates/workspace-server/src/plugin_plan/store.rs

@@ -352,18 +352,21 @@ 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()));
+    let mut statement = expected.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND tbl_name='worker_plugin_launches' ORDER BY name")?;
+    for row in statement.query_map([], |row| {
+        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+    })? {
+        let (name, sql) = row?;
+        let actual = conn
+            .query_row(
+                "SELECT sql FROM sqlite_schema WHERE name=?1",
+                [&name],
+                |row| row.get::<_, String>(0),
+            )
+            .optional()?;
+        if actual.as_deref() != Some(&sql) {
+            return Err(Error::Store("Worker Plugin launch schema mismatch".into()));
+        }
     }
     Ok(())
 }

crates/workspace-server/src/schema_convergence_tests.rs

@@ -116,6 +116,10 @@ mod schema_convergence {
                             version: 109,
                             name: "Durable Plugin Host invocations and HTTPS effect receipts".into(),
                         },
+                        WorkspaceSchemaMigrationStep {
+                            version: 110,
+                            name: "Worker Plugin launch selection associations".into(),
+                        },
                     ]
                 );
                 assert_eq!(
@@ -136,6 +140,7 @@ mod schema_convergence {
                     expected.push((107, "Immutable Worker Plugin plans and retained pins".into()));
                     expected.push((108, "Subject Ticket responsibility and current-body notification routing".into()));
                     expected.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
+                    expected.push((110, "Worker Plugin launch selection associations".into()));
                     assert_eq!(workspace_schema_migration_history(conn)?, expected);
                     for table in ["passkey_credentials", "passkey_metadata", "passkey_addition_receipts", "backend_jobs", "backend_job_attempts"] {
                         assert_eq!(state_contract_schema_signature(conn, table), fresh.with_conn(|c| Ok(state_contract_schema_signature(c, table)))?, "{table}");

crates/workspace-server/src/server.rs

@@ -50,6 +50,8 @@ mod drive_grants;
 mod plugin;
 #[path = "server_plugin_delivery.rs"]
 mod plugin_delivery;
+#[path = "server_plugin_diagnostics.rs"]
+mod plugin_diagnostics;
 #[path = "server_plugin_host.rs"]
 mod plugin_host;
 #[path = "server_plugin_runtime.rs"]
@@ -4153,6 +4155,23 @@ impl WorkspaceApi {
         config: ServerConfig,
         store: Arc<dyn ControlPlaneStore>,
         host_executors: HostExecutors,
+    ) -> Result<Self> {
+        Self::new_with_embedded_configurators(
+            config,
+            store,
+            host_executors,
+            |factory| factory,
+            |_| {},
+        )
+        .await
+    }
+
+    async fn new_with_embedded_configurators(
+        config: ServerConfig,
+        store: Arc<dyn ControlPlaneStore>,
+        host_executors: HostExecutors,
+        configure_factory: impl FnOnce(ProfileRuntimeWorkerFactory) -> ProfileRuntimeWorkerFactory,
+        configure_api: impl FnOnce(&mut WorkspaceApi),
     ) -> Result<Self> {
         let resource_broker = BackendResourceBroker::default();
         if store.get_workspace(&config.workspace_id).await?.is_none() {
@@ -4190,10 +4209,12 @@ impl WorkspaceApi {
         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"),
+            &config
+                .embedded_runtime_store_root
+                .with_extension("plugin-cache"),
         )
         .map_err(crate::Error::Config)?;
-        let execution_backend = WorkerRuntimeExecutionBackend::new(
+        let execution_backend = WorkerRuntimeExecutionBackend::new(configure_factory(
             ProfileRuntimeWorkerFactory::new(config.workspace_execution_root.clone())
                 .with_plugin_runtime(Arc::new(plugins))
                 .with_embedded_worker_mutation_dispatcher(
@@ -4210,7 +4231,7 @@ impl WorkspaceApi {
                 )
                 .with_runtime_store_dir(config.embedded_runtime_store_root.clone())
                 .with_resource_client(Arc::new(resource_broker.clone())),
-        )
+        ))
         .map_err(|err| {
             crate::Error::Store(format!(
                 "failed to initialize embedded Worker backend: {err}"
@@ -4282,6 +4303,7 @@ impl WorkspaceApi {
         {
             api.register_workspace_runtime_binding(binding)?;
         }
+        configure_api(&mut api);
         let plugin_owner = Arc::new(api.clone());
         plugin_providers.bind(&plugin_owner);
         api.plugin_runtime_owner = Some(plugin_owner);
@@ -13622,6 +13644,24 @@ impl server_api::ServerApi for ServerApiContractService {
         plugin::list(api, actor, workspace_id, query).await
     }
 
+    async fn plugin_worker_diagnostics(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        runtime_id: String,
+        worker_id: String,
+    ) -> std::result::Result<server_api::PluginWorkerDiagnostics, server_api::RepositoryApiError>
+    {
+        plugin_diagnostics::diagnostics(
+            self.workspace_api()?,
+            actor,
+            workspace_id,
+            runtime_id,
+            worker_id,
+        )
+        .await
+    }
+
     async fn plugin_show(
         &self,
         actor: RequestActor,
@@ -38755,6 +38795,8 @@ mod tests {
         include!("server_plugin_host_tests.rs");
         include!("server_plugin_runtime_tests.rs");
     }
+    include!("server_plugin_controller_tests.rs");
+    include!("server_plugin_remote_controller_tests.rs");
 
     mod ticket_tags_tests {
         include!("server_ticket_tags_tests.rs");

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

@@ -2662,17 +2662,19 @@ async fn delegated_subject_child_cannot_downgrade_ticket_start_to_manual_by_omit
     );
 }
 
-const CONNECTED_RUNTIME: &str = "connected-subject-authorization";
+pub(super) const CONNECTED_RUNTIME: &str = "connected-subject-authorization";
 
 /// Shared signed Backend/Runtime connection. Only the supplied model is replaced;
 /// temporary roots, request authentication and Controller/tool dispatch stay real.
-struct ConnectedWorkerHarness {
-    api: WorkspaceApi,
+pub(super) struct ConnectedWorkerHarness {
+    pub(super) api: WorkspaceApi,
+    // Expose only the persisted root needed to reopen ordinary Session history.
+    pub(super) runtime_root: PathBuf,
     runtime: worker_runtime::Runtime,
     factory: worker_runtime::worker_backend::ProfileRuntimeWorkerFactory,
     identity: RuntimeIdentityMaterial,
     backend_url: String,
-    owner: String,
+    pub(super) owner: String,
     materializer: worker_runtime::working_directory::RuntimeGitMaterializer,
     authorizations: Arc<Mutex<Vec<(Value, StatusCode)>>>,
     runtime_errors: Arc<Mutex<Vec<String>>>,
@@ -2681,7 +2683,12 @@ struct ConnectedWorkerHarness {
     _dir: tempfile::TempDir,
 }
 impl ConnectedWorkerHarness {
-    async fn human(&self, method: reqwest::Method, path: &str, body: Value) -> (StatusCode, Value) {
+    pub(super) async fn human(
+        &self,
+        method: reqwest::Method,
+        path: &str,
+        body: Value,
+    ) -> (StatusCode, Value) {
         let response = reqwest::Client::builder()
             .timeout(std::time::Duration::from_secs(30))
             .build()
@@ -2709,6 +2716,23 @@ impl ConnectedWorkerHarness {
     }
 
     async fn start<M: agen::llm_client::LlmClient + Clone + 'static>(model: M) -> Self {
+        Self::start_configured(model, None).await
+    }
+
+    /// Keep the existing signed RuntimeHTTP assembly, adding only the production
+    /// Plugin bridge and a TLS-peer transport on the Server before either router
+    /// can handle a request. No guest, delivery or owner implementation is replaced.
+    pub(super) async fn start_with_plugins<M: agen::llm_client::LlmClient + Clone + 'static>(
+        model: M,
+        transport: Arc<crate::plugin_https::HttpsTransport>,
+    ) -> Self {
+        Self::start_configured(model, Some(transport)).await
+    }
+
+    async fn start_configured<M: agen::llm_client::LlmClient + Clone + 'static>(
+        model: M,
+        plugin_https_transport: Option<Arc<crate::plugin_https::HttpsTransport>>,
+    ) -> Self {
         const RUNTIME: &str = CONNECTED_RUNTIME;
         let dir = tempfile::tempdir().unwrap();
         init_clean_git_workspace(dir.path());
@@ -2721,49 +2745,24 @@ impl ConnectedWorkerHarness {
         let materializer = worker_runtime::working_directory::RuntimeGitMaterializer::new(
             dir.path().join("connected-workdirs"),
         );
-        let factory = worker_runtime::worker_backend::ProfileRuntimeWorkerFactory::new(
-            dir.path().join("connected-profiles"),
-        )
-        .with_remote_worker_mutation_identity(identity.clone())
-        .with_workspace_request_client(
-            worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new(
-                TEST_WORKSPACE_ID,
-                &backend_url,
-                RUNTIME,
-            )
-            .with_runtime_request_source(&identity, &backend_url),
-        )
-        .with_runtime_store_dir(&runtime_root)
-        .with_spawn_model_client_factory(move || Box::new(model.clone()));
-        let execution =
-            worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(factory.clone())
-                .unwrap()
-                .with_working_directory_materializer(materializer.clone());
-        let runtime = worker_runtime::Runtime::with_fs_store_and_execution_backend(
-            worker_runtime::fs_store::FsRuntimeStoreOptions {
-                root: runtime_root,
-                runtime_id: RUNTIME.into(),
-                display_name: None,
-            },
-            Arc::new(execution),
-        )
-        .unwrap();
 
         let mut config = test_server_config(dir.path());
         config.backend_base_url = Some(backend_url.clone());
-        // The embedded Runtime is unused. Both source and child below execute on the
-        // connected real Runtime above, not on this existing Host fixture backend.
+        // The embedded Runtime is unused. Connected Workers execute on the real
+        // Runtime assembled below, not on this Host fixture's embedded backend.
         let control = Arc::new(test_control_store(&config));
         seed_test_registered_workspace(control.as_ref(), &config)
             .await
             .unwrap();
-        let api = WorkspaceApi::new_with_execution_backend(
+        let mut api = WorkspaceApi::new_with_execution_backend(
             config,
             control,
             Arc::new(DeterministicExecutionBackend::default()),
         )
         .await
         .unwrap();
+        let plugins_enabled = plugin_https_transport.is_some();
+        api.plugin_https_transport = plugin_https_transport;
         let now = now_registry_timestamp();
         api.store
             .upsert_workspace_runtime_binding_record(
@@ -2796,29 +2795,84 @@ impl ConnectedWorkerHarness {
             .signing_identities
             .provision_existing(TEST_WORKSPACE_ID, &test_owner_actor().account_id)
             .unwrap();
+        let registration = worker_runtime::workspace_issuer::WorkspaceIssuerTrustRecord {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            backend_url: backend_url.clone(),
+            key_id: workspace_identity.key_id.clone(),
+            algorithm: workspace_identity.algorithm.clone(),
+            public_key: workspace_identity.public_key.clone().unwrap(),
+            public_key_fingerprint: workspace_identity.public_key_fingerprint.clone().unwrap(),
+            trust_id: "connected-authorization-workspace-trust".into(),
+            state: worker_runtime::workspace_issuer::WorkspaceIssuerTrustState::Active,
+            registered_at_unix: 1,
+            updated_at_unix: 1,
+        };
         let verifier = worker_runtime::workspace_issuer::WorkspaceCapabilityVerifier::new(
-            vec![
-                worker_runtime::workspace_issuer::WorkspaceIssuerTrustRecord {
-                    workspace_id: TEST_WORKSPACE_ID.into(),
-                    backend_url: backend_url.clone(),
-                    key_id: workspace_identity.key_id.clone(),
-                    algorithm: workspace_identity.algorithm.clone(),
-                    public_key: workspace_identity.public_key.clone().unwrap(),
-                    public_key_fingerprint: workspace_identity
-                        .public_key_fingerprint
-                        .clone()
-                        .unwrap(),
-                    trust_id: "connected-authorization-workspace-trust".into(),
-                    state: worker_runtime::workspace_issuer::WorkspaceIssuerTrustState::Active,
-                    registered_at_unix: 1,
-                    updated_at_unix: 1,
-                },
-            ],
+            vec![registration.clone()],
             Arc::new(
                 worker_runtime::workspace_issuer::InMemoryWorkspaceClaimReplayProtection::default(),
             ),
         )
         .unwrap();
+        let workspace_client =
+            worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new(
+                TEST_WORKSPACE_ID,
+                &backend_url,
+                RUNTIME,
+            )
+            .with_runtime_request_source(&identity, &backend_url);
+        let mut factory = worker_runtime::worker_backend::ProfileRuntimeWorkerFactory::new(
+            dir.path().join("connected-profiles"),
+        )
+        .with_remote_worker_mutation_identity(identity.clone())
+        .with_workspace_request_client(workspace_client.clone())
+        .with_runtime_store_dir(&runtime_root)
+        .with_spawn_model_client_factory(move || Box::new(model.clone()));
+        if plugins_enabled {
+            // This is the Runtime's registered Server, also used by the HTTP
+            // verifier above, not a fabricated Plugin admission response. All
+            // admission/delivery/Host operations still traverse the signed client.
+            let current: Arc<worker_runtime::plugin::remote::CurrentPluginServer> =
+                Arc::new(move |workspace| {
+                    if workspace == registration.workspace_id {
+                        Ok(registration.clone())
+                    } else {
+                        Err(yoi_plugin_runtime::delivery::DeliveryError::Denied)
+                    }
+                });
+            // Remote providers build the standard signed Host client, retaining
+            // PLUGIN_HOST_RESPONSE_BODY_LIMIT (finite 4.5 MiB); do not substitute
+            // an unbounded test client or an in-process owner fallback here.
+            let providers = Arc::new(
+                worker_runtime::plugin::remote::RemotePluginRuntimeProviders::new(
+                    std::collections::HashMap::from([(
+                        TEST_WORKSPACE_ID.to_string(),
+                        workspace_client,
+                    )]),
+                    current,
+                ),
+            );
+            let plugins = worker_runtime::plugin::PinnedPluginToolRuntime::new(
+                providers,
+                &runtime_root.with_extension("plugin-cache"),
+            )
+            .unwrap();
+            factory = factory.with_plugin_runtime(Arc::new(plugins));
+        }
+        // Freeze the fully configured factory only after installing the bridge.
+        let execution =
+            worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(factory.clone())
+                .unwrap()
+                .with_working_directory_materializer(materializer.clone());
+        let runtime = worker_runtime::Runtime::with_fs_store_and_execution_backend(
+            worker_runtime::fs_store::FsRuntimeStoreOptions {
+                root: runtime_root.clone(),
+                runtime_id: RUNTIME.into(),
+                display_name: None,
+            },
+            Arc::new(execution),
+        )
+        .unwrap();
         let runtime_errors = Arc::new(Mutex::new(Vec::<String>::new()));
         let captured_errors = runtime_errors.clone();
         let runtime_server = ConnectedRuntimeServer::start(
@@ -2915,6 +2969,7 @@ impl ConnectedWorkerHarness {
 
         Self {
             api,
+            runtime_root,
             runtime,
             factory,
             identity,

crates/workspace-server/src/server_plugin.rs

@@ -16,6 +16,7 @@ pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
         )
         .merge(server_api_axum::plugin_install(service.clone()))
         .merge(server_api_axum::plugin_list(service.clone()))
+        .merge(server_api_axum::plugin_worker_diagnostics(service.clone()))
         .merge(server_api_axum::plugin_show(service.clone()))
         .merge(server_api_axum::plugin_artifact_add(service.clone()))
         .merge(server_api_axum::plugin_revoke(service.clone()))
@@ -60,7 +61,7 @@ pub(super) fn is_catalog_path(path: &str) -> bool {
         })
 }
 
-fn failure(error: Error) -> RepositoryApiError {
+pub(super) fn failure(error: Error) -> RepositoryApiError {
     let (status, code) = match error {
         Error::PluginCatalog(error) => (error.status(), error.code()),
         Error::WorkspacePermissionDenied(_) => (403, "access_denied"),
@@ -73,7 +74,7 @@ fn failure(error: Error) -> RepositoryApiError {
     RepositoryApiError::new(status, code, format!("Plugin catalog: {code}"), vec![])
 }
 
-async fn authorize(
+pub(super) async fn authorize(
     api: &WorkspaceApi,
     workspace: &str,
     actor: &RequestActor,

crates/workspace-server/src/server_plugin_diagnostics.rs

@@ -0,0 +1,259 @@
+//! Owner-only observations reuse live execution policy without issuing authority.
+//! No artifact bytes, guest, invocation or receipt is involved. The DB snapshot
+//! and live Worker checks are point-in-time observations, not an execution lease.
+use super::*;
+use crate::plugin_host as policy;
+use server_api::authorization::{OperationAuthorizationRequest, PermissionResource};
+use server_api::*;
+
+fn reason(error: Error) -> PluginHostErrorCode {
+    policy::map_error(error).code
+}
+
+fn wire_limit(value: impl TryInto<u32>) -> Result<u32> {
+    value
+        .try_into()
+        .map_err(|_| Error::Config("Plugin Runtime budget exceeds diagnostic wire range".into()))
+}
+
+fn runtime_limits(limits: &yoi_plugin_runtime::Limits) -> Result<PluginRuntimeLimits> {
+    Ok(PluginRuntimeLimits {
+        payload_bytes: wire_limit(limits.payload_bytes)?,
+        config_bytes: wire_limit(limits.config_bytes)?,
+        memory_bytes: wire_limit(limits.memory_bytes)?,
+        table_elements: wire_limit(limits.table_elements)?,
+        instances: wire_limit(limits.instances)?,
+        memories: wire_limit(limits.memories)?,
+        tables: wire_limit(limits.tables)?,
+        resources: wire_limit(limits.resources)?,
+        fuel: wire_limit(limits.fuel)?,
+        wall_time_ms: wire_limit(limits.wall_time.as_millis())?,
+        host_operations: wire_limit(limits.host_operations)?,
+        http_body_bytes: wire_limit(limits.http_body_bytes)?,
+        http_header_bytes: wire_limit(limits.http_header_bytes)?,
+        http_header_pairs: wire_limit(limits.http_header_pairs)?,
+        log_bytes: wire_limit(limits.log_bytes)?,
+    })
+}
+
+fn effective_limits(
+    runtime: &yoi_plugin_runtime::Limits,
+    binding: &crate::plugin_binding::PluginConnectionLimits,
+) -> PluginDiagnosticConnectionLimits {
+    PluginDiagnosticConnectionLimits::Available {
+        timeout_ms: u64::from(binding.timeout_ms).min(runtime.wall_time.as_millis() as u64) as u32,
+        max_request_bytes: u64::from(binding.max_request_bytes).min(runtime.http_body_bytes as u64)
+            as u32,
+        max_response_bytes: u64::from(binding.max_response_bytes)
+            .min(runtime.http_body_bytes as u64) as u32,
+        max_header_bytes: u64::from(binding.max_header_bytes).min(runtime.http_header_bytes as u64)
+            as u32,
+    }
+}
+
+pub(super) async fn diagnostics(
+    api: &WorkspaceApi,
+    actor: RequestActor,
+    workspace: String,
+    runtime: String,
+    worker: String,
+) -> std::result::Result<PluginWorkerDiagnostics, RepositoryApiError> {
+    // Management authority permits reading this projection only. Never turn the
+    // human actor into a Worker source or borrow its authority for live policy.
+    plugin::authorize(api, &workspace, &actor, true).await?;
+    let store = api.config_store.clone();
+    let w = workspace.clone();
+    let r = runtime.clone();
+    let k = worker.clone();
+    let selection = store
+        .db_executor()
+        .run(move || {
+            store.with_conn(|conn| crate::plugin_plan::worker_launch_selection(conn, &w, &r, &k))
+        })
+        .await
+        .map_err(plugin::failure)?;
+    let limits = yoi_plugin_runtime::Limits::default();
+    let selection = match selection {
+        None => PluginWorkerDiagnosticSelection::Unrecorded,
+        Some(None) => PluginWorkerDiagnosticSelection::Empty,
+        Some(Some(saved)) => {
+            let worker_ref = RuntimeWorkerRef::new(&runtime, &worker);
+            let execution = api
+                .plugin_execution_identity(&worker_ref)
+                .await
+                .map_err(reason);
+            let api = api.clone();
+            let store = api.config_store.clone();
+            let w = workspace.clone();
+            store
+                .db_executor()
+                .run(move || {
+                    // Keep the existing Subject/delegation owner outside the
+                    // config-store lock, as in the live Host boundary.
+                    let scopes: Vec<_> = saved
+                        .plan
+                        .plugins()
+                        .iter()
+                        .map(|pin| {
+                            subject_authorization::authorize_scope(
+                                &api,
+                                &worker_ref,
+                                &OperationAuthorizationRequest {
+                                    operation: "plugin.admit".into(),
+                                    resource: PermissionResource::PluginInstallation {
+                                        workspace_id: w.clone(),
+                                        installation_id: pin.installation_id().into(),
+                                    },
+                                },
+                            )
+                            .map(|_| ())
+                            .map_err(|error| reason(error.error))
+                        })
+                        .collect();
+                    store.with_conn_mut(|conn| {
+                        // Deferred read transaction: policy::authorize and
+                        // live_binding only SELECT; dropping it writes nothing.
+                        let tx = conn
+                            .transaction_with_behavior(rusqlite::TransactionBehavior::Deferred)?;
+                        let saved = crate::plugin_plan::plugin_plan_load(&tx, &w, &saved.plan_id)?;
+                        let runtime = yoi_plugin_runtime::Limits::default();
+                        let mut plugins = Vec::with_capacity(saved.plan.plugins().len());
+                        for (pin, scope) in saved.plan.plugins().iter().zip(scopes) {
+                            let observed =
+                                execution
+                                    .as_ref()
+                                    .map_err(|code| *code)
+                                    .and_then(|execution| {
+                                        scope?;
+                                        policy::authorize(
+                                            &tx,
+                                            &w,
+                                            execution,
+                                            &saved.plan_id,
+                                            pin.installation_id(),
+                                            &pin.projection().tools,
+                                            &PluginHostOperation::Admit,
+                                        )
+                                        .map(|_| ())
+                                        .map_err(reason)
+                                    });
+                            let admission = match observed {
+                                Ok(()) => PluginDiagnosticAdmission::Admissible,
+                                Err(reason_code) => {
+                                    PluginDiagnosticAdmission::Refused { reason_code }
+                                }
+                            };
+                            let connections = pin
+                                .projection()
+                                .bindings
+                                .iter()
+                                .map(|reference| {
+                                    let effective_limits =
+                                        match policy::live_binding(&tx, &w, reference) {
+                                            Ok(binding) => {
+                                                effective_limits(&runtime, &binding.limits)
+                                            }
+                                            Err(error) => {
+                                                PluginDiagnosticConnectionLimits::Unavailable {
+                                                    reason_code: reason(error),
+                                                }
+                                            }
+                                        };
+                                    PluginDiagnosticConnection {
+                                        slot: reference.slot.clone(),
+                                        effective_limits,
+                                    }
+                                })
+                                .collect();
+                            plugins.push(PluginWorkerDiagnostic {
+                                installation_id: pin.installation_id().into(),
+                                package_id: pin.package_id().into(),
+                                package_version: pin.package_version().into(),
+                                package_digest: pin.package_digest().into(),
+                                artifact_digest: pin.artifact_digest().into(),
+                                artifact_size: pin.artifact_size(),
+                                surfaces: pin
+                                    .projection()
+                                    .surfaces
+                                    .iter()
+                                    .map(|surface| match surface {
+                                        crate::plugin_plan::PluginSurface::Tool => {
+                                            PluginDiagnosticSurface::Tool
+                                        }
+                                    })
+                                    .collect(),
+                                selected_tools: pin.projection().tools.clone(),
+                                admission,
+                                connections,
+                            });
+                        }
+                        let pin_state = match saved.pin_state {
+                            crate::plugin_plan::PlanPinState::Pending => {
+                                PluginDiagnosticPinState::Pending
+                            }
+                            crate::plugin_plan::PlanPinState::Active => {
+                                PluginDiagnosticPinState::Active
+                            }
+                            crate::plugin_plan::PlanPinState::Restorable => {
+                                PluginDiagnosticPinState::Restorable
+                            }
+                            crate::plugin_plan::PlanPinState::Released => {
+                                PluginDiagnosticPinState::Released
+                            }
+                        };
+                        Ok(PluginWorkerDiagnosticSelection::Plan {
+                            plan_id: saved.plan_id,
+                            fingerprint: saved.fingerprint,
+                            pin_state,
+                            plugins,
+                        })
+                    })
+                })
+                .await
+                .map_err(plugin::failure)?
+        }
+    };
+    Ok(PluginWorkerDiagnostics {
+        workspace_id: workspace,
+        runtime_id: runtime,
+        worker_id: worker,
+        runtime_limits: runtime_limits(&limits).map_err(plugin::failure)?,
+        selection,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn connection_limits_intersect_each_runtime_and_binding_budget() {
+        let runtime = yoi_plugin_runtime::Limits {
+            wall_time: std::time::Duration::from_millis(200),
+            http_body_bytes: 300,
+            http_header_bytes: 100,
+            ..Default::default()
+        };
+        for (binding, expected) in [
+            ((150, 200, 250, 50), (150, 200, 250, 50)),
+            ((250, 400, 500, 150), (200, 300, 300, 100)),
+            ((150, 400, 250, 150), (150, 300, 250, 100)),
+        ] {
+            let binding = crate::plugin_binding::PluginConnectionLimits {
+                timeout_ms: binding.0,
+                max_request_bytes: binding.1,
+                max_response_bytes: binding.2,
+                max_header_bytes: binding.3,
+            };
+            assert_eq!(
+                effective_limits(&runtime, &binding),
+                PluginDiagnosticConnectionLimits::Available {
+                    timeout_ms: expected.0,
+                    max_request_bytes: expected.1,
+                    max_response_bytes: expected.2,
+                    max_header_bytes: expected.3,
+                }
+            );
+        }
+    }
+}

crates/workspace-server/src/server_plugin_host.rs

@@ -295,8 +295,11 @@ impl EmbeddedPluginHostOwner {
         }
     }
     #[cfg(test)]
-    pub(crate) fn with_transport(mut self, transport: https::HttpsTransport) -> Self {
-        self.transport_override = Some(Arc::new(transport));
+    pub(crate) fn with_transport(
+        mut self,
+        transport: impl Into<Arc<https::HttpsTransport>>,
+    ) -> Self {
+        self.transport_override = Some(transport.into());
         self
     }
 }

crates/workspace-server/src/server_plugin_host_tests.rs

@@ -86,7 +86,8 @@ impl Fixture {
         let upload = api
             .config_store
             .plugin_upload(
-                &ServerConfig::server_data_root_for_data_dir(&api.config.data_root).join("plugin-artifacts"),
+                &ServerConfig::server_data_root_for_data_dir(&api.config.data_root)
+                    .join("plugin-artifacts"),
                 TEST_WORKSPACE_ID,
                 "fixture-owner",
                 &package.bytes,
@@ -1372,7 +1373,7 @@ async fn prepare_unavailable_owner_foreign_workspace_and_unsigned_embedded_never
     f.api
         .config_store
         .with_conn(|conn| {
-            conn.execute_batch("DROP TABLE plugin_execution_plans")?;
+            conn.execute_batch("ALTER TABLE plugin_execution_plans RENAME COLUMN plan_json TO unavailable_plan_json")?;
             Ok(())
         })
         .unwrap();

crates/workspace-server/src/server_plugin_remote_controller_tests.rs

@@ -0,0 +1,403 @@
+// Include directly in server::tests (this file owns its module):
+//     include!("server_plugin_remote_controller_tests.rs");
+// Reuses subject_authorization_tests' signed RuntimeHTTP/Controller harness.
+// There is no second listener/scheduler/bootstrap implementation here. Only the
+// external model and TLS peer DNS/socket/trust are substituted, not the guest,
+// Remote Plugin providers, authorization, artifact delivery or effect receipts.
+// Run: cargo test -p yoi-workspace-server --lib plugin_remote_controller_tests
+// Prerequisites: openssl and the existing connected Runtime test environment.
+mod plugin_remote_controller_tests {
+    use super::plugin_host_tests::plugin_runtime_tests::{Model, assert_https_history};
+    use super::plugin_tests::request;
+    use super::subject_authorization_tests::{CONNECTED_RUNTIME, ConnectedWorkerHarness};
+    use super::*;
+    use crate::plugin_https::test_support::{HOST, Peer, response};
+    use crate::plugin_plan::PlanPinState;
+    use agen::llm_client::{ClientError, LlmClient, Request as ModelRequest};
+    use session_store::Store;
+    use sha2::Digest;
+    use std::sync::Mutex;
+    use std::time::Duration;
+
+    const PROFILE: &str = "project:remote-plugin-controller";
+    const TOOL: &str = "example_status";
+    const WATCHDOG: Duration = Duration::from_secs(60);
+
+    // Installation allocates the public namespace after the harness is started.
+    // Defer only that name, reusing the existing scripted model implementation.
+    #[derive(Clone)]
+    struct InstalledToolModel {
+        tool: Arc<Mutex<String>>,
+        requests: Arc<Mutex<Vec<ModelRequest>>>,
+    }
+
+    #[async_trait::async_trait]
+    impl LlmClient for InstalledToolModel {
+        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 model = Model {
+                tool_name: self.tool.lock().unwrap().clone(),
+                requests: self.requests.clone(),
+            };
+            model.stream(request).await
+        }
+    }
+
+    // Administrative setup only; all mutations pass the existing authenticated
+    // upload/install/config handlers, with no manually saved plan or Worker row.
+    async fn install_and_configure(
+        f: &ConnectedWorkerHarness,
+        peer: &Peer,
+    ) -> (server_api::PluginOperationResponse, Vec<u8>) {
+        use manifest::plugin::*;
+        let directory = tempfile::tempdir().unwrap();
+        fs::write(
+            directory.path().join("plugin.toml"),
+            include_str!("../../../resources/plugin/templates/rust-component-https/plugin.toml"),
+        )
+        .unwrap();
+        fs::write(
+            directory.path().join("plugin.component.wasm"),
+            include_bytes!("../../plugin-pdk/tests/fixtures/https-tool-v2.component.wasm"),
+        )
+        .unwrap();
+        let limits = PluginPackageLimits::default();
+        let package = pack_plugin_package(
+            &read_plugin_directory(directory.path(), PluginSourceKind::Explicit, &limits).unwrap(),
+            &limits,
+        )
+        .unwrap();
+        let prefix = format!("/api/w/{TEST_WORKSPACE_ID}");
+        let upload = request(
+            build_router(f.api.clone()),
+            Some(&f.owner),
+            "POST",
+            &format!("{prefix}/plugin-uploads"),
+            package.bytes.clone(),
+            true,
+            StatusCode::OK,
+        )
+        .await;
+        let (status, installed) = f
+            .human(
+                reqwest::Method::POST,
+                &format!("{prefix}/plugins"),
+                json!({"upload_id":upload["upload_id"], "alias":"remote_controller",
+                "operation_key":"remote-controller-install"}),
+            )
+            .await;
+        assert_eq!(status, StatusCode::OK, "{installed}");
+        let installed: server_api::PluginOperationResponse =
+            serde_json::from_value(installed).unwrap();
+        let artifact = &installed.installation.artifacts[0];
+        assert_eq!(artifact.tools[0].name, TOOL);
+        let mut binding = peer.binding();
+        binding.methods = vec![server_api::PluginHttpMethod::Get];
+        binding.path_prefixes = vec!["/status".into()];
+        let plugins = json!({"enabled":{"remote_controller":{
+            "installation_id":installed.installation.installation_id,
+            "expected_last_operation":installed.operation_id,
+            "package_digest":artifact.package_digest, "artifact_digest":artifact.artifact_digest,
+            "tools":[TOOL], "logging":false, "config":{}, "connections":{"api":"remote"}
+        }}, "bindings":{"remote":binding}});
+        let selection = json!({"selections":[{
+            "installation_id":installed.installation.installation_id,
+            "tools":[TOOL], "logging":false, "https_slots":["api"]
+        }]});
+        let content = format!(
+            r#"{{
+            plugins = {};
+            profile = {{ default_profile = "{PROFILE}"; entries = [{{
+                selector = "{PROFILE}"; label = "Remote controller"; description = "";
+                profile = {{ slug = "remote-plugin-controller"; scope = "workspace_read";
+                    delegation_scope = "workspace_read";
+                    model = {{ scheme = "anthropic"; model_id = "scripted"; auth = {{ kind = "none"; }}; }};
+                    plugins = {};
+                }};
+            }}]; }};
+        }}"#,
+            crate::plugin_settings::render_value(&plugins).unwrap(),
+            crate::plugin_settings::render_value(&selection).unwrap()
+        );
+        let state = f
+            .api
+            .config_store
+            .load_workspace_config(TEST_WORKSPACE_ID)
+            .unwrap()
+            .unwrap();
+        let path = config_source::VirtualPath::parse("main.dcdl").unwrap();
+        let commit = ConfigCommitRequest {
+            base_digest: state.snapshot.digest.clone(),
+            entrypoints: state.contract.entrypoints.clone(),
+            changes: vec![config_source::ConfigTreeChange::Update {
+                path: path.clone(),
+                expected_digest: state.snapshot.get(&path).unwrap().content_digest.clone(),
+                content,
+            }],
+        };
+        let (status, result) = f
+            .human(
+                reqwest::Method::POST,
+                &format!("{prefix}/config/source-tree/commit"),
+                serde_json::to_value(commit).unwrap(),
+            )
+            .await;
+        assert_eq!(status, StatusCode::CREATED, "{result}");
+        (installed, package.bytes)
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+    async fn remote_runtime_http_create_runs_wasmtime_https_and_persists_ordinary_tool_history() {
+        let mut peer = Peer::new(HOST, vec![response(503, &[], b"peer data")]).await;
+        let requests = Arc::new(Mutex::new(Vec::new()));
+        let tool = Arc::new(Mutex::new(String::new()));
+        let f = ConnectedWorkerHarness::start_with_plugins(
+            InstalledToolModel {
+                tool: tool.clone(),
+                requests: requests.clone(),
+            },
+            Arc::new(peer.transport()),
+        )
+        .await;
+        assert!(
+            !f.api
+                .store
+                .is_hosting_admin("user-connected-authorization-owner")
+                .unwrap(),
+            "Plugin setup must use a non-operator Workspace owner"
+        );
+        let (installed, package) = install_and_configure(&f, &peer).await;
+        let public_name = format!("{}__{TOOL}", installed.installation.installation_id);
+        *tool.lock().unwrap() = public_name.clone();
+        let (status, created) = tokio::time::timeout(
+            WATCHDOG,
+            f.human(
+                reqwest::Method::POST,
+                &format!("/api/w/{TEST_WORKSPACE_ID}/workers"),
+                json!({"runtime_id":CONNECTED_RUNTIME, "display_name":"Remote Plugin controller",
+                "profile":PROFILE, "initial_submit":[], "workdir_attachments":[],
+                "feature_connections":{}}),
+            ),
+        )
+        .await
+        .expect("signed RuntimeHTTP create timed out");
+        assert_eq!(status, StatusCode::OK, "{created}");
+        let worker = RuntimeWorkerRef::new(
+            created["runtime_id"].as_str().unwrap(),
+            created["worker_id"].as_str().unwrap(),
+        );
+        assert_eq!(worker.runtime_id, CONNECTED_RUNTIME);
+        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().len(), 1);
+        assert_eq!(
+            saved.plan.plugins()[0].installation_id(),
+            installed.installation.installation_id
+        );
+        let aggregate = f.runtime_root.join("workers").join(&worker.worker_id);
+        let persisted: Value =
+            serde_json::from_slice(&fs::read(aggregate.join("worker.json")).unwrap()).unwrap();
+        assert_eq!(persisted["execution_state"]["state"], "available");
+        assert_eq!(
+            persisted["execution_state"]["execution"]["request"]["plugin_execution"],
+            serde_json::to_value(runtime_api::PluginExecutionSnapshot {
+                plan_id: saved.plan_id.clone(),
+                fingerprint: saved.fingerprint.clone(),
+                plan: saved.plan.clone(),
+            })
+            .unwrap()
+        );
+        let cache_files = fs::read_dir(f.runtime_root.with_extension("plugin-cache"))
+            .unwrap()
+            .map(|entry| entry.unwrap().path())
+            .filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("artifact"))
+            .collect::<Vec<_>>();
+        assert_eq!(cache_files.len(), 1);
+        assert_eq!(
+            fs::read(&cache_files[0]).unwrap(),
+            package,
+            "Remote delivery must resolve the exact saved pin"
+        );
+        assert_eq!(
+            peer.connection_count(),
+            0,
+            "controller preparation must not invoke the guest Tool"
+        );
+        assert!(requests.lock().unwrap().is_empty());
+
+        let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None)
+            .await
+            .unwrap();
+        let greeting = tokio::time::timeout(WATCHDOG, async {
+            loop {
+                if let protocol::Event::Snapshot {
+                    greeting, state, ..
+                } = connection.events.recv().await.unwrap()
+                {
+                    assert_eq!(state.state, protocol::WorkerState::Idle);
+                    break greeting;
+                }
+            }
+        })
+        .await
+        .expect("Remote Controller did not publish its snapshot");
+        assert!(greeting.tools.contains(&public_name));
+        assert!(!greeting.tools.iter().any(|name| name == TOOL));
+        connection
+            .methods
+            .send(protocol::Method::Submit {
+                submission_request_id: "remote-controller-https-submit".into(),
+                input: vec![protocol::Segment::Text {
+                    content: "Check the service status.".into(),
+                }],
+            })
+            .await
+            .unwrap();
+        tokio::time::timeout(WATCHDOG, async {
+            let mut accepted = false;
+            let mut finished = false;
+            loop {
+                match connection.events.recv().await.unwrap() {
+                    protocol::Event::SubmissionAccepted {
+                        submission_request_id,
+                        ..
+                    } => {
+                        assert_eq!(submission_request_id, "remote-controller-https-submit");
+                        accepted = true;
+                    }
+                    protocol::Event::SubmissionRejected { message, .. } => {
+                        panic!("Submit rejected: {message}")
+                    }
+                    protocol::Event::RunEnd { result } => {
+                        assert!(accepted);
+                        assert_eq!(result, protocol::RunResult::Finished);
+                        finished = true;
+                    }
+                    protocol::Event::WorkerState { snapshot }
+                        if finished
+                            && snapshot.state == protocol::WorkerState::Idle
+                            && snapshot.last_finished_submission_request_id.as_deref()
+                                == Some("remote-controller-https-submit") =>
+                    {
+                        break;
+                    }
+                    _ => {}
+                }
+            }
+        })
+        .await
+        .expect("Remote Controller HTTPS submission did not finish");
+        {
+            let observed = requests.lock().unwrap();
+            assert_eq!(observed.len(), 2);
+            assert!(
+                observed[0]
+                    .tools
+                    .iter()
+                    .any(|definition| definition.name == public_name)
+            );
+            assert_https_history(&observed[1].items, &public_name);
+        }
+        assert!(
+            String::from_utf8(
+                tokio::time::timeout(WATCHDOG, peer.observed())
+                    .await
+                    .unwrap()
+            )
+            .unwrap()
+            .starts_with("GET /status HTTP/1.1\r\n")
+        );
+        assert_eq!(peer.connection_count(), 1);
+        drop(connection);
+        let api = f.api.clone();
+        let stopped = worker.clone();
+        tokio::time::timeout(
+            WATCHDOG,
+            tokio::task::spawn_blocking(move || {
+                api.runtime.stop_worker(
+                    &stopped,
+                    WorkerLifecycleRequest {
+                        reason: None,
+                        ticket_assignment: None,
+                    },
+                )
+            }),
+        )
+        .await
+        .expect("Remote Controller stop timed out")
+        .unwrap()
+        .unwrap();
+
+        // Reopen the Runtime's normal filesystem Session store, not a model-only
+        // observation or an injected Session entry, after Controller shutdown.
+        let reopened = session_store::WorkerSessionStore::new(aggregate.join("session")).unwrap();
+        let session = reopened.session_id().unwrap().unwrap();
+        let segments = reopened.list_segments(session).unwrap();
+        assert_eq!(segments.len(), 1);
+        let history = session_store::restore(&reopened, session, segments[0]).unwrap();
+        assert_https_history(&history.history, &public_name);
+        let (invocations, receipts): (i64, i64) = f
+            .api
+            .config_store
+            .with_conn(|conn| {
+                Ok((
+                    conn.query_row(
+                        "SELECT count(*) FROM plugin_invocations WHERE workspace_id=?1",
+                        [TEST_WORKSPACE_ID],
+                        |row| row.get(0),
+                    )?,
+                    conn.query_row(
+                        "SELECT count(*) FROM plugin_effect_receipts WHERE workspace_id=?1",
+                        [TEST_WORKSPACE_ID],
+                        |row| row.get(0),
+                    )?,
+                ))
+            })
+            .unwrap();
+        assert_eq!((invocations, receipts), (1, 1));
+        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 r.effect_key='status-v1'",
+                rusqlite::params![TEST_WORKSPACE_ID, saved.plan_id, worker.worker_id],
+                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))?)
+        }).unwrap();
+        let identity = serde_json::to_vec(&(session.to_string(), "https-call")).unwrap();
+        let consumer_id = format!(
+            "tool-{}",
+            sha2::Sha256::digest(identity)
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect::<String>()
+        );
+        assert_eq!(
+            serde_json::from_str::<server_api::PluginConsumer>(&consumer).unwrap(),
+            server_api::PluginConsumer::ToolCall {
+                tool_call_id: consumer_id,
+                tool_name: TOOL.into()
+            }
+        );
+        assert_eq!(state, "completed");
+        assert_eq!(status, 503);
+        assert_eq!(body, b"peer data");
+        shutdown_workspace_feature_storage(&f.api).await.unwrap();
+    }
+}

crates/workspace-server/src/server_plugin_runtime.rs

@@ -60,21 +60,36 @@ impl WorkspaceApi {
                 plugin_execution: None,
             });
         }
-        let selector = match &request.profile {
+        let selector_name = match &request.profile {
+            worker_runtime::catalog::ProfileSelector::Builtin(s) if !s.starts_with("builtin:") => {
+                format!("builtin:{s}")
+            }
             worker_runtime::catalog::ProfileSelector::Builtin(s)
-            | worker_runtime::catalog::ProfileSelector::Named(s) => s,
+            | worker_runtime::catalog::ProfileSelector::Named(s) => s.clone(),
         };
-        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()))?,
+        let selector = selector_name.as_str();
+        let archive = match request.resolved_config_bundle.as_ref() {
+            Some(bundle) if bundle.profile_source_archive.is_some() => bundle
+                .profile_source_archive
+                .clone()
+                .expect("checked archive"),
+            Some(bundle) if bundle.profile_source_archive_handle.is_some() => {
+                let handle = bundle
+                    .profile_source_archive_handle
+                    .as_ref()
+                    .expect("checked handle");
+                self.resource_broker
+                    .profile_source_archive(&handle.digest)
+                    .ok_or_else(|| Error::Config("Plugin Profile archive unavailable".into()))?
+            }
+            _ => crate::profile_settings::builtin_profile_source_archive(&request.profile)
+                .map_err(Error::Config)?,
         };
+        let manifest = archive
+            .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",
@@ -159,11 +174,18 @@ impl PluginRuntimeProviders for EmbeddedPluginRuntimeProviders {
             EMBEDDED_RUNTIME_ID,
         )
         .with_runtime_request_source(&EMBEDDED_RUNTIME_REQUEST_IDENTITY, audience);
+        #[cfg(test)]
+        let transport = api.plugin_https_transport.clone();
         let owner = plugin_host::EmbeddedPluginHostOwner::authenticated(
             api,
             workspace.into(),
             client.plugin_host_authorizer(worker.worker_id),
         );
+        #[cfg(test)]
+        let owner = match transport {
+            Some(transport) => owner.with_transport(transport),
+            None => owner,
+        };
         Ok(PluginRuntimeServices {
             artifacts: Arc::new(artifacts),
             owner: Arc::new(owner),

crates/workspace-server/src/server_plugin_runtime_tests.rs

@@ -7,7 +7,7 @@
 // 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 {
+pub(super) 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};
@@ -125,9 +125,9 @@ mod plugin_runtime_tests {
     }
 
     #[derive(Clone)]
-    struct Model {
-        tool_name: String,
-        requests: Arc<Mutex<Vec<ModelRequest>>>,
+    pub(in crate::server::tests) struct Model {
+        pub(in crate::server::tests) tool_name: String,
+        pub(in crate::server::tests) requests: Arc<Mutex<Vec<ModelRequest>>>,
     }
 
     #[async_trait::async_trait]
@@ -149,7 +149,14 @@ mod plugin_runtime_tests {
             let events = match turn {
                 0 => vec![
                     ModelEvent::tool_use_start(0, "https-call", &self.tool_name),
-                    ModelEvent::tool_input_delta(0, "{}"),
+                    ModelEvent::tool_input_delta(
+                        0,
+                        if self.tool_name.ends_with("__example_echo") {
+                            r#"{"text":"probe"}"#
+                        } else {
+                            "{}"
+                        },
+                    ),
                     ModelEvent::tool_use_stop(0),
                 ],
                 1 => vec![
@@ -170,7 +177,7 @@ mod plugin_runtime_tests {
         }
     }
 
-    fn assert_https_history(items: &[agen::Item], public_name: &str) {
+    pub(in crate::server::tests) fn assert_https_history(items: &[agen::Item], public_name: &str) {
         let calls = items
             .iter()
             .filter_map(|item| match item {
@@ -215,7 +222,11 @@ mod plugin_runtime_tests {
         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_eq!(
+            *disposition,
+            agen::ToolResultDisposition::Success,
+            "{summary}: {content:?}"
+        );
         assert!(!*is_error);
         assert!(attachments.is_empty());
         assert_eq!(summary, "HTTP response observed");

crates/workspace-server/src/server_plugin_tests.rs

@@ -25,7 +25,7 @@ fn package_bytes(version: &str) -> Vec<u8> {
     .bytes
 }
 
-async fn request(
+pub(super) async fn request(
     app: Router,
     token: Option<&str>,
     method: &str,
@@ -415,29 +415,101 @@ async fn api_restart_recovers_orphan_publication_and_preserves_committed_catalog
 
 #[tokio::test]
 async fn workspace_owner_without_hosting_admin_configures_plugin_through_existing_config_commit() {
-    let dir=tempfile::tempdir().unwrap();
-    let api=test_api(dir.path()).await;
-    let owner=seed_test_workspace_owner_token(api.store.as_ref(),"plugin-config-owner").await;
-    let reader=seed_test_api_token(api.store.as_ref(),"plugin-config-reader");
-    let app=build_router(api.clone());
-    let prefix=format!("/api/w/{TEST_WORKSPACE_ID}");
-    let upload=request(app.clone(),Some(&owner),"POST",&format!("{prefix}/plugin-uploads"),package_bytes("0.1.0"),true,StatusCode::OK).await;
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    let owner = seed_test_workspace_owner_token(api.store.as_ref(), "plugin-config-owner").await;
+    let reader = seed_test_api_token(api.store.as_ref(), "plugin-config-reader");
+    let app = build_router(api.clone());
+    let prefix = format!("/api/w/{TEST_WORKSPACE_ID}");
+    let upload = request(
+        app.clone(),
+        Some(&owner),
+        "POST",
+        &format!("{prefix}/plugin-uploads"),
+        package_bytes("0.1.0"),
+        true,
+        StatusCode::OK,
+    )
+    .await;
     let installed=json(app.clone(),&owner,&format!("{prefix}/plugins"),serde_json::json!({"upload_id":upload["upload_id"],"alias":"configured","operation_key":"config-install"}),StatusCode::OK).await;
-    let state=get(app.clone(),Some(&owner),&format!("{prefix}/config/source-tree"),StatusCode::OK).await;
-    let a=&installed["installation"]["artifacts"][0];
-    let source=format!("{{ plugins.enabled.echo = {{ installation_id = {}; expected_last_operation = {}; package_digest = {}; artifact_digest = {}; tools = [\"example_echo\"]; config = {{}}; }}; }} as WorkspaceConfigSchema",installed["installation"]["installation_id"],installed["operation_id"],a["package_digest"],a["artifact_digest"]);
-    let change=serde_json::json!({"base_digest":state["snapshot"]["digest"],"entrypoints":["main.dcdl"],"changes":[{"kind":"update","path":"main.dcdl","expected_digest":state["snapshot"]["entries"]["main.dcdl"]["content_digest"],"content":source}]});
-    json(app.clone(),&reader,&format!("{prefix}/config/source-tree/commit"),change.clone(),StatusCode::FORBIDDEN).await;
-    json(app.clone(),&reader,&format!("{prefix}/config/source-tree/preview"),change.clone(),StatusCode::FORBIDDEN).await;
-    let preview=json(app.clone(),&owner,&format!("{prefix}/config/source-tree/preview"),change.clone(),StatusCode::OK).await;
-    let unchanged=get(app.clone(),Some(&owner),&format!("{prefix}/config/source-tree"),StatusCode::OK).await;
-    assert_eq!(unchanged["snapshot"]["digest"],state["snapshot"]["digest"]);
-    let saved=json(app.clone(),&owner,&format!("{prefix}/config/source-tree/commit"),change.clone(),StatusCode::CREATED).await;
-    assert_eq!(saved,preview);
-    assert_ne!(saved["snapshot"]["digest"],state["snapshot"]["digest"]);
-    json(app.clone(),&owner,&format!("{prefix}/config/source-tree/commit"),change,StatusCode::CONFLICT).await;
+    let state = get(
+        app.clone(),
+        Some(&owner),
+        &format!("{prefix}/config/source-tree"),
+        StatusCode::OK,
+    )
+    .await;
+    let a = &installed["installation"]["artifacts"][0];
+    let source = format!(
+        "{{ plugins.enabled.echo = {{ installation_id = {}; expected_last_operation = {}; package_digest = {}; artifact_digest = {}; tools = [\"example_echo\"]; config = {{}}; }}; }} as WorkspaceConfigSchema",
+        installed["installation"]["installation_id"],
+        installed["operation_id"],
+        a["package_digest"],
+        a["artifact_digest"]
+    );
+    let change = serde_json::json!({"base_digest":state["snapshot"]["digest"],"entrypoints":["main.dcdl"],"changes":[{"kind":"update","path":"main.dcdl","expected_digest":state["snapshot"]["entries"]["main.dcdl"]["content_digest"],"content":source}]});
+    json(
+        app.clone(),
+        &reader,
+        &format!("{prefix}/config/source-tree/commit"),
+        change.clone(),
+        StatusCode::FORBIDDEN,
+    )
+    .await;
+    json(
+        app.clone(),
+        &reader,
+        &format!("{prefix}/config/source-tree/preview"),
+        change.clone(),
+        StatusCode::FORBIDDEN,
+    )
+    .await;
+    let preview = json(
+        app.clone(),
+        &owner,
+        &format!("{prefix}/config/source-tree/preview"),
+        change.clone(),
+        StatusCode::OK,
+    )
+    .await;
+    let unchanged = get(
+        app.clone(),
+        Some(&owner),
+        &format!("{prefix}/config/source-tree"),
+        StatusCode::OK,
+    )
+    .await;
+    assert_eq!(unchanged["snapshot"]["digest"], state["snapshot"]["digest"]);
+    let saved = json(
+        app.clone(),
+        &owner,
+        &format!("{prefix}/config/source-tree/commit"),
+        change.clone(),
+        StatusCode::CREATED,
+    )
+    .await;
+    assert_eq!(saved, preview);
+    assert_ne!(saved["snapshot"]["digest"], state["snapshot"]["digest"]);
+    json(
+        app.clone(),
+        &owner,
+        &format!("{prefix}/config/source-tree/commit"),
+        change,
+        StatusCode::CONFLICT,
+    )
+    .await;
     json(app.clone(),&owner,&format!("{prefix}/plugins/{}/uninstall",installed["installation"]["installation_id"].as_str().unwrap()),serde_json::json!({"expected_last_operation":installed["operation_id"],"operation_key":"config-blocks-uninstall"}),StatusCode::CONFLICT).await;
-    let current=api.config_store.load_workspace_config(TEST_WORKSPACE_ID).unwrap().unwrap();
-    let candidates=current.contract.schema_bundle.contributions.iter().filter(|c|c.provider_id=="builtin:plugins").count();
-    assert_eq!(candidates,1);
+    let current = api
+        .config_store
+        .load_workspace_config(TEST_WORKSPACE_ID)
+        .unwrap()
+        .unwrap();
+    let candidates = current
+        .contract
+        .schema_bundle
+        .contributions
+        .iter()
+        .filter(|c| c.provider_id == "builtin:plugins")
+        .count();
+    assert_eq!(candidates, 1);
 }

crates/workspace-server/src/subject_assignment_migration_tests.rs

@@ -1129,6 +1129,7 @@ mod subject_assignment_migration_tests {
                 _ => panic!("unsupported fixture origin"),
             };
             tail.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
+            tail.push((110, "Worker Plugin launch selection associations".into()));
             tail
         }
 

crates/workspace-server/src/workspace_deletion.rs

@@ -18,6 +18,7 @@ use crate::{Error, Result};
 const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "plugin_effect_receipts",
     "plugin_invocations",
+    "worker_plugin_launches",
     "plugin_execution_plans",
     "plugin_references",
     "plugin_operations",
@@ -513,6 +514,7 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
             for (table, trigger) in [
                 ("plugin_effect_receipts", "plugin_effect_retained"),
                 ("plugin_invocations", "plugin_invocation_retained"),
+                ("worker_plugin_launches", "worker_plugin_launch_retained"),
                 ("plugin_execution_plans", "plugin_plan_replay_retained"),
                 ("notification_source_preparation_failures", "notification_source_preparation_failures_no_delete"),
                 ("notification_job_targets", "notification_job_targets_no_delete"),