diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index e40577695..9661e6462 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -232,8 +232,9 @@ fn build_runtime(config: &ProcessConfig) -> Result { _ => 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 { .iter() .filter(|issuer| issuer.state == WorkspaceIssuerTrustState::Active) { - factory = factory.with_workspace_request_client( - RuntimeWorkspaceRequestClient::new( - issuer.workspace_id.clone(), - issuer.backend_url.clone(), - identity.identity_id.clone(), - ) - .with_runtime_request_source(identity, issuer.backend_url.clone()), - ); + 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()); + 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 = + 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, > = None; diff --git a/crates/worker-runtime/src/plugin.rs b/crates/worker-runtime/src/plugin.rs index 01d33db48..a2a4ef399 100644 --- a/crates/worker-runtime/src/plugin.rs +++ b/crates/worker-runtime/src/plugin.rs @@ -85,3 +85,208 @@ pub async fn materialize( .await .map_err(|error| error.to_string()) } + +pub struct PluginRuntimeServices { + pub artifacts: std::sync::Arc, + pub owner: std::sync::Arc, +} + +/// 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; +} + +pub struct PinnedPluginToolRuntime { + providers: std::sync::Arc, + cache: yoi_plugin_runtime::delivery::ArtifactCache, + limits: yoi_plugin_runtime::Limits, +} +impl PinnedPluginToolRuntime { + pub fn new( + providers: std::sync::Arc, + private_cache_root: &std::path::Path, + ) -> Result { + 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 { + 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 + + Send + + Sync; + pub struct RemotePluginRuntimeProviders { + clients: HashMap, + current: Arc, + } + impl RemotePluginRuntimeProviders { + pub fn new( + clients: HashMap, + current: Arc, + ) -> Self { + Self { clients, current } + } + } + struct RegisteredOwner { + source: Arc, + owner: Arc, + } + #[async_trait] + impl yoi_plugin_runtime::owner::PluginHostOwner for RegisteredOwner { + async fn call( + &self, + request: server_api::PluginHostRequest, + ) -> Result { + 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 { + 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 = 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, + }) + } + } +} diff --git a/crates/worker-runtime/src/workspace_request.rs b/crates/worker-runtime/src/workspace_request.rs index 4b55e457c..5373b1270 100644 --- a/crates/worker-runtime/src/workspace_request.rs +++ b/crates/worker-runtime/src/workspace_request.rs @@ -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 { + 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, diff --git a/crates/worker-runtime/src/workspace_request/plugin_delivery.rs b/crates/worker-runtime/src/workspace_request/plugin_delivery.rs index 29cc6cc36..bcf5530e7 100644 --- a/crates/worker-runtime/src/workspace_request/plugin_delivery.rs +++ b/crates/worker-runtime/src/workspace_request/plugin_delivery.rs @@ -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 diff --git a/crates/workspace-server/src/plugin_plan.rs b/crates/workspace-server/src/plugin_plan.rs index 35960ff30..92652234c 100644 --- a/crates/workspace-server/src/plugin_plan.rs +++ b/crates/workspace-server/src/plugin_plan.rs @@ -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 for Error { fn from(error: PlanError) -> Self { diff --git a/crates/workspace-server/src/plugin_plan/launch_migration_tests.rs b/crates/workspace-server/src/plugin_plan/launch_migration_tests.rs new file mode 100644 index 000000000..360c008d1 --- /dev/null +++ b/crates/workspace-server/src/plugin_plan/launch_migration_tests.rs @@ -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()); +} diff --git a/crates/workspace-server/src/plugin_plan/launch_schema.sql b/crates/workspace-server/src/plugin_plan/launch_schema.sql new file mode 100644 index 000000000..5f6f788d7 --- /dev/null +++ b/crates/workspace-server/src/plugin_plan/launch_schema.sql @@ -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) +); diff --git a/crates/workspace-server/src/plugin_plan/store.rs b/crates/workspace-server/src/plugin_plan/store.rs index 48276e48a..cefa70f86 100644 --- a/crates/workspace-server/src/plugin_plan/store.rs +++ b/crates/workspace-server/src/plugin_plan/store.rs @@ -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>> { + let id: Option> = 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> { + 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)?; diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 911b4b851..353db3fd6 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -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, journal_lifecycle: Arc>>>>, journal_dispatch_owner: Option>, + plugin_runtime_owner: Option>, observation_proxy: BackendObservationProxy, runtime_subscription_broker: RuntimeSubscriptionBroker, pub(crate) worker_projection: Arc, @@ -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::>(); 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, - worker_id, - create_fingerprint: reservation.create_fingerprint, + 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, + &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 { diff --git a/crates/workspace-server/src/server_plugin_delivery.rs b/crates/workspace-server/src/server_plugin_delivery.rs index b8622a214..2568ce11b 100644 --- a/crates/workspace-server/src/server_plugin_delivery.rs +++ b/crates/workspace-server/src/server_plugin_delivery.rs @@ -45,13 +45,40 @@ async fn worker_from_context( context: &server_api::ServerRequestContext, ) -> std::result::Result { 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 { + 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 { 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 { let execution = api - .plugin_execution_identity(worker) + .plugin_artifact_execution_identity(worker) .await .map_err(failure)?; let stage_api = api.clone(); diff --git a/crates/workspace-server/src/server_plugin_delivery_tests.rs b/crates/workspace-server/src/server_plugin_delivery_tests.rs index a522baaf4..d51030df7 100644 --- a/crates/workspace-server/src/server_plugin_delivery_tests.rs +++ b/crates/workspace-server/src/server_plugin_delivery_tests.rs @@ -82,14 +82,22 @@ fn artifact_package() -> Vec { } 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( diff --git a/crates/workspace-server/src/server_plugin_host.rs b/crates/workspace-server/src/server_plugin_host.rs index bbf8ce442..7ce444e5f 100644 --- a/crates/workspace-server/src/server_plugin_host.rs +++ b/crates/workspace-server/src/server_plugin_host.rs @@ -24,7 +24,7 @@ impl WorkspaceApi { ) } - fn plugin_reserved_identity( + pub(super) fn plugin_reserved_identity( &self, worker: &RuntimeWorkerRef, expected_fingerprint: Option<&str>, diff --git a/crates/workspace-server/src/server_plugin_host_tests.rs b/crates/workspace-server/src/server_plugin_host_tests.rs index 913f59aa3..7ba43c5d3 100644 --- a/crates/workspace-server/src/server_plugin_host_tests.rs +++ b/crates/workspace-server/src/server_plugin_host_tests.rs @@ -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, diff --git a/crates/workspace-server/src/server_plugin_launch_tests.rs b/crates/workspace-server/src/server_plugin_launch_tests.rs new file mode 100644 index 000000000..7de98d52c --- /dev/null +++ b/crates/workspace-server/src/server_plugin_launch_tests.rs @@ -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::,_>>()?) + }).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); +} diff --git a/crates/workspace-server/src/server_plugin_runtime.rs b/crates/workspace-server/src/server_plugin_runtime.rs new file mode 100644 index 000000000..af9a08c60 --- /dev/null +++ b/crates/workspace-server/src/server_plugin_runtime.rs @@ -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 { + 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>>, +} +impl EmbeddedPluginRuntimeProviders { + pub(super) fn bind(&self, api: &Arc) { + *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 { + 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), + }) + } +} diff --git a/crates/workspace-server/src/server_plugin_runtime_tests.rs b/crates/workspace-server/src/server_plugin_runtime_tests.rs new file mode 100644 index 000000000..a6f83feca --- /dev/null +++ b/crates/workspace-server/src/server_plugin_runtime_tests.rs @@ -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, + owner: Arc, + } + + #[async_trait::async_trait] + impl PluginRuntimeProviders for RuntimeProviders { + async fn connect( + &self, + workspace: &str, + worker: &WorkerRef, + ) -> std::result::Result { + 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, Option) { + 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>>, + } + + #[async_trait::async_trait] + impl LlmClient for Model { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + request: ModelRequest, + ) -> std::result::Result { + 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::>(); + 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::(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::>(); + 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::>(), + [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::>(), + [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::() + ); + let (consumer, state, status, body): (String, String, i64, Vec) = 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::(&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; + } +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index c617b82a6..5c3a9a1d7 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -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}"); }