diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index f6580c3b9..27f74ac06 100644 --- a/crates/server-api/src/lib.rs +++ b/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; + #[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; + #[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, diff --git a/crates/server-api/src/plugin.rs b/crates/server-api/src/plugin.rs index 9d9693c55..b2afa63c9 100644 --- a/crates/server-api/src/plugin.rs +++ b/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, + }, +} + +#[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, + pub selected_tools: Vec, + pub admission: PluginDiagnosticAdmission, + pub connections: Vec, +} + +/// 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. diff --git a/crates/server-api/src/plugin_types.rs b/crates/server-api/src/plugin_types.rs index 775b4ac52..61a165f9f 100644 --- a/crates/server-api/src/plugin_types.rs +++ b/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", diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 9661e6462..3b016a448 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -275,7 +275,7 @@ fn build_runtime(config: &ProcessConfig) -> Result { ); 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)); diff --git a/crates/worker-runtime/src/plugin.rs b/crates/worker-runtime/src/plugin.rs index a2a4ef399..677c1768c 100644 --- a/crates/worker-runtime/src/plugin.rs +++ b/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 + 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(), diff --git a/crates/worker-runtime/src/workspace_request.rs b/crates/worker-runtime/src/workspace_request.rs index 5373b1270..8bb00f7ac 100644 --- a/crates/worker-runtime/src/workspace_request.rs +++ b/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, + 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, diff --git a/crates/worker/src/plugin.rs b/crates/worker/src/plugin.rs index 5924043e7..70757e38b 100644 --- a/crates/worker/src/plugin.rs +++ b/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) diff --git a/crates/workspace-server/src/assignment_tracking_tests.rs b/crates/workspace-server/src/assignment_tracking_tests.rs index 18807feea..b48a591e2 100644 --- a/crates/workspace-server/src/assignment_tracking_tests.rs +++ b/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()) { diff --git a/crates/workspace-server/src/plugin_catalog/migration_tests.rs b/crates/workspace-server/src/plugin_catalog/migration_tests.rs index 99c66caeb..58b06f735 100644 --- a/crates/workspace-server/src/plugin_catalog/migration_tests.rs +++ b/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![(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(()) diff --git a/crates/workspace-server/src/plugin_host.rs b/crates/workspace-server/src/plugin_host.rs index 84d1e80af..0b352720c 100644 --- a/crates/workspace-server/src/plugin_host.rs +++ b/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, diff --git a/crates/workspace-server/src/plugin_host_migration_tests.rs b/crates/workspace-server/src/plugin_host_migration_tests.rs index b8f57a6d6..9c527e33c 100644 --- a/crates/workspace-server/src/plugin_host_migration_tests.rs +++ b/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::>(), [109]); + assert_eq!(plan.migrations.iter().map(|step| step.version).collect::>(), [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(()) diff --git a/crates/workspace-server/src/plugin_https/test_support.rs b/crates/workspace-server/src/plugin_https/test_support.rs index 8f471b2ac..458a202fc 100644 --- a/crates/workspace-server/src/plugin_https/test_support.rs +++ b/crates/workspace-server/src/plugin_https/test_support.rs @@ -300,7 +300,11 @@ impl Peer { } pub(crate) async fn observed(&mut self) -> Vec { - 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 { + tokio::time::timeout(deadline, self.observations.recv()) .await .unwrap() .unwrap() diff --git a/crates/workspace-server/src/plugin_plan/launch_schema.sql b/crates/workspace-server/src/plugin_plan/launch_schema.sql index 5f6f788d7..4788f48dd 100644 --- a/crates/workspace-server/src/plugin_plan/launch_schema.sql +++ b/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; diff --git a/crates/workspace-server/src/plugin_plan/migration_tests.rs b/crates/workspace-server/src/plugin_plan/migration_tests.rs index 5db7cbd6b..b95443e82 100644 --- a/crates/workspace-server/src/plugin_plan/migration_tests.rs +++ b/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)?; diff --git a/crates/workspace-server/src/plugin_plan/store.rs b/crates/workspace-server/src/plugin_plan/store.rs index cefa70f86..4da08d5d3 100644 --- a/crates/workspace-server/src/plugin_plan/store.rs +++ b/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(()) } diff --git a/crates/workspace-server/src/schema_convergence_tests.rs b/crates/workspace-server/src/schema_convergence_tests.rs index 8e5fdc249..37a9827dc 100644 --- a/crates/workspace-server/src/schema_convergence_tests.rs +++ b/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}"); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index d5894b789..d42be1504 100644 --- a/crates/workspace-server/src/server.rs +++ b/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, host_executors: HostExecutors, + ) -> Result { + Self::new_with_embedded_configurators( + config, + store, + host_executors, + |factory| factory, + |_| {}, + ) + .await + } + + async fn new_with_embedded_configurators( + config: ServerConfig, + store: Arc, + host_executors: HostExecutors, + configure_factory: impl FnOnce(ProfileRuntimeWorkerFactory) -> ProfileRuntimeWorkerFactory, + configure_api: impl FnOnce(&mut WorkspaceApi), ) -> Result { 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 + { + 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"); diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index 6229a8f97..f24c11835 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/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>>, runtime_errors: Arc>>, @@ -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(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( + model: M, + transport: Arc, + ) -> Self { + Self::start_configured(model, Some(transport)).await + } + + async fn start_configured( + model: M, + plugin_https_transport: Option>, + ) -> 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 = + 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::::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, diff --git a/crates/workspace-server/src/server_plugin.rs b/crates/workspace-server/src/server_plugin.rs index e70a6ea6e..594aa8bcf 100644 --- a/crates/workspace-server/src/server_plugin.rs +++ b/crates/workspace-server/src/server_plugin.rs @@ -16,6 +16,7 @@ pub(super) fn router(service: Arc) -> 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, diff --git a/crates/workspace-server/src/server_plugin_controller_tests.rs b/crates/workspace-server/src/server_plugin_controller_tests.rs new file mode 100644 index 000000000..51e4ad172 --- /dev/null +++ b/crates/workspace-server/src/server_plugin_controller_tests.rs @@ -0,0 +1,1222 @@ +// Include directly in server::tests (this file owns its module). +// Production Embedded Server assembly -> ProfileRuntimeWorkerFactory -> actual +// Controller/Engine -> pinned Wasmtime Tool -> HTTPS -> SQLite and Session store. +// Only the external model and the TLS peer's DNS/socket/trust are substituted. +// No FixtureRuntime, manually prepared Plugin Feature, or fabricated Host owner. +// +// Shared-helper visibility required in the existing suites (no Fixture exposed): +// - plugin_tests::request: pub(super) +// - plugin_host_tests::plugin_runtime_tests: pub(super) +// - that module's Model, Model.tool_name, Model.requests and assert_https_history: +// pub(in crate::server::tests) +// TLS Peer is already crate-visible. WorkerSessionStore needs no new wrapper. +// Run: cargo test -p yoi-workspace-server --lib plugin_controller_tests +// Prerequisites: openssl and the normal connected-restore test environment. +mod plugin_controller_tests { + use super::plugin_host_tests::plugin_runtime_tests::{Model, assert_https_history}; + use super::plugin_tests::request; + use super::*; + use crate::plugin_https::test_support::{HOST, Peer, response}; + use crate::plugin_plan::PlanPinState; + use session_store::Store; + use sha2::Digest; + use std::sync::{ + Mutex, + atomic::{AtomicUsize, Ordering}, + }; + use std::time::Duration; + + const PROFILE: &str = "project:plugin-controller"; + const TOOL: &str = "example_status"; + const WATCHDOG: Duration = Duration::from_secs(60); + + struct Fixture { + api: WorkspaceApi, + owner: String, + model_tool: Arc>, + requests: Arc>>, + model_clients: Arc, + _directory: tempfile::TempDir, + } + + impl Fixture { + async fn new(peer: &Peer) -> Self { + // Use the established short-path/environment guard for real restore + // (not a mock controller or a replacement execution backend). + let directory = super::restored_parent_control_tests::connected_restore_tempdir(); + let mut config = test_server_config(directory.path()); + config.repositories.clear(); + let store = + Arc::new(SqliteWorkspaceStore::open_api_fixture(&config.database_path).unwrap()); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + let requests = Arc::new(Mutex::new(Vec::new())); + let model_tool = Arc::new(Mutex::new(String::new())); + let model_clients = Arc::new(AtomicUsize::new(0)); + let clients = model_clients.clone(); + let observed = requests.clone(); + let tool = model_tool.clone(); + let transport = Arc::new(peer.transport()); + let api = WorkspaceApi::new_with_embedded_configurators( + config, + store, + HostExecutors::default(), + move |factory| { + factory.with_spawn_model_client_factory(move || { + clients.fetch_add(1, Ordering::SeqCst); + Box::new(Model { + tool_name: tool.lock().unwrap().clone(), + requests: observed.clone(), + }) + }) + }, + move |api| api.plugin_https_transport = Some(transport), + ) + .await + .unwrap(); + let owner = + seed_test_workspace_owner_token(api.store.as_ref(), "plugin-controller-owner") + .await; + // Delivery/cache authority is the active Workspace signing identity, + // not the Embedded Runtime key or the human catalog credential. + request( + build_router(api.clone()), + Some(&owner), + "POST", + &format!("/api/w/{TEST_WORKSPACE_ID}/settings/signing-identity/provision"), + b"{}".to_vec(), + false, + StatusCode::OK, + ) + .await; + Self { + api, + owner, + model_tool, + requests, + model_clients, + _directory: directory, + } + } + + async fn install(&self, peer: &Peer) -> (server_api::PluginOperationResponse, Vec) { + self.install_component(peer, true).await + } + + async fn install_component( + &self, + peer: &Peer, + https: bool, + ) -> (server_api::PluginOperationResponse, Vec) { + use manifest::plugin::*; + let directory = tempfile::tempdir().unwrap(); + let (manifest, component): (&str, &[u8]) = if https { + ( + include_str!( + "../../../resources/plugin/templates/rust-component-https/plugin.toml" + ), + include_bytes!("../../plugin-pdk/tests/fixtures/https-tool-v2.component.wasm"), + ) + } else { + ( + include_str!( + "../../../resources/plugin/templates/rust-component-tool/plugin.toml" + ), + include_bytes!("../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"), + ) + }; + fs::write(directory.path().join("plugin.toml"), manifest).unwrap(); + fs::write(directory.path().join("plugin.component.wasm"), component).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 router = build_router(self.api.clone()); + request( + router.clone(), + Some(&self.owner), + "GET", + &format!("{prefix}/plugins"), + vec![], + false, + StatusCode::OK, + ) + .await; + let upload = request( + router.clone(), + Some(&self.owner), + "POST", + &format!("{prefix}/plugin-uploads"), + package.bytes.clone(), + true, + StatusCode::OK, + ) + .await; + let installed = request(router, Some(&self.owner), "POST", &format!("{prefix}/plugins"), + serde_json::to_vec(&json!({ + "upload_id": upload["upload_id"], "alias": "controller", "operation_key": "install" + })).unwrap(), false, StatusCode::OK).await; + let installed: server_api::PluginOperationResponse = + serde_json::from_value(installed).unwrap(); + *self.model_tool.lock().unwrap() = format!( + "{}__{}", + installed.installation.installation_id, + installed.installation.artifacts[0].tools[0].name + ); + self.configure(&installed, peer, true).await; + (installed, package.bytes) + } + + async fn configure( + &self, + installed: &server_api::PluginOperationResponse, + peer: &Peer, + selected: bool, + ) { + let artifact = &installed.installation.artifacts[0]; + let mut binding = peer.binding(); + binding.methods = vec![server_api::PluginHttpMethod::Get]; + binding.path_prefixes = vec!["/status".into()]; + let tool = &artifact.tools[0].name; + let https = tool == TOOL; + let plugins = json!({"enabled":{"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":if https { json!({"api":"remote"}) } else { json!({}) } + }},"bindings":if https { json!({"remote":binding}) } else { json!({}) }}); + let selection = if selected { + json!({"selections":[{"installation_id":installed.installation.installation_id, + "tools":[tool],"logging":false,"https_slots":if https { vec!["api"] } else { vec![] }}]}) + } else { + json!({"selections":[]}) + }; + // This deliberately minimal value Profile exercises project selection + // without adding unrelated Subject startup/Jobs or credential access. + let content = format!( + r#"{{ + plugins = {}; + profile = {{ default_profile = "{PROFILE}"; entries = [{{ + selector = "{PROFILE}"; label = "Controller"; description = ""; + profile = {{ slug = "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 = self + .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, + }], + }; + request( + build_router(self.api.clone()), + Some(&self.owner), + "POST", + &format!("/api/w/{TEST_WORKSPACE_ID}/config/source-tree/commit"), + serde_json::to_vec(&commit).unwrap(), + false, + StatusCode::CREATED, + ) + .await; + } + + async fn revoke(&self, installed: &server_api::PluginOperationResponse) { + let revoked = request( + build_router(self.api.clone()), + Some(&self.owner), + "POST", + &format!( + "/api/w/{TEST_WORKSPACE_ID}/plugins/{}/revoke", + installed.installation.installation_id + ), + serde_json::to_vec(&json!({ + "expected_last_operation": installed.operation_id, + "operation_key": "controller-revoke", + "artifact_digest": installed.installation.artifacts[0].artifact_digest, + "reason": "Withdraw controller artifact" + })) + .unwrap(), + false, + StatusCode::OK, + ) + .await; + assert_eq!(revoked["installation"]["artifacts"][0]["revoked"], true); + } + + fn cached_artifact(&self, package: &[u8]) -> PathBuf { + let cache = self + .api + .config + .embedded_runtime_store_root + .with_extension("plugin-cache"); + let files = fs::read_dir(cache) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("artifact")) + .collect::>(); + assert_eq!(files.len(), 1); + assert_eq!(fs::read(&files[0]).unwrap(), package); + files[0].clone() + } + + async fn restore( + &self, + worker: &RuntimeWorkerRef, + request_id: &str, + ) -> InternalWorkerRestoreResult { + let api = self.api.clone(); + let worker = worker.clone(); + let request_id = request_id.to_string(); + tokio::time::timeout( + WATCHDOG, + tokio::task::spawn_blocking(move || { + let observation = api + .runtime + .worker(&worker) + .unwrap() + .restore_observation_token + .unwrap(); + api.restore_workspace_worker( + &worker, + server_api::WorkerRestoreRequest { + expected_observation_token: observation, + request_id, + }, + ) + }), + ) + .await + .expect("Controller restore exceeded the fixture deadline") + .unwrap() + .unwrap() + } + + async fn create(&self, profile: &str) -> RuntimeWorkerRef { + let Json(created) = create_workspace_worker( + State(self.api.clone()), + HeaderMap::new(), + Json(create_request(profile)), + ) + .await + .unwrap(); + RuntimeWorkerRef::new(created.runtime_id, created.worker_id) + } + + fn aggregate(&self, worker: &RuntimeWorkerRef) -> PathBuf { + self.api + .config + .embedded_runtime_store_root + .join("workers") + .join(&worker.worker_id) + } + + fn execution_request(&self, worker: &RuntimeWorkerRef) -> Value { + let saved: Value = serde_json::from_slice( + &fs::read(self.aggregate(worker).join("worker.json")).unwrap(), + ) + .unwrap(); + assert_eq!(saved["execution_state"]["state"], "available"); + saved["execution_state"]["execution"]["request"].clone() + } + + fn saved_history( + &self, + worker: &RuntimeWorkerRef, + ) -> (session_store::SessionId, Vec) { + let reopened = + session_store::WorkerSessionStore::new(self.aggregate(worker).join("session")) + .unwrap(); + let session = reopened.session_id().unwrap().unwrap(); + let segments = reopened.list_segments(session).unwrap(); + assert_eq!(segments.len(), 1); + let saved = session_store::restore(&reopened, session, segments[0]).unwrap(); + (session, saved.history) + } + + fn count(&self, table: &str) -> i64 { + assert!( + [ + "plugin_invocations", + "plugin_effect_receipts", + "plugin_execution_plans" + ] + .contains(&table) + ); + self.api + .config_store + .with_conn(|conn| { + Ok(conn.query_row( + &format!("SELECT count(*) FROM {table} WHERE workspace_id=?1"), + [TEST_WORKSPACE_ID], + |row| row.get(0), + )?) + }) + .unwrap() + } + + fn effect_identity(&self, worker: &RuntimeWorkerRef) -> (String, String, String) { + self.api + .config_store + .with_conn(|conn| { + Ok(conn.query_row( + "SELECT r.invocation_id,r.effect_key,r.fingerprint FROM plugin_effect_receipts r \ + JOIN plugin_invocations i USING(workspace_id,invocation_id) \ + WHERE i.workspace_id=?1 AND i.worker_id=?2", + rusqlite::params![TEST_WORKSPACE_ID, worker.worker_id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?) + }) + .unwrap() + } + + async fn stop(&self, worker: &RuntimeWorkerRef) { + let api = self.api.clone(); + let worker = worker.clone(); + tokio::task::spawn_blocking(move || { + api.runtime.stop_worker( + &worker, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + }) + .await + .unwrap() + .unwrap(); + } + + async fn close(self) { + shutdown_workspace_feature_storage(&self.api).await.unwrap(); + } + } + + fn create_request(profile: &str) -> CreateWorkspaceWorkerRequest { + CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(), + display_name: "Plugin controller".into(), + singleton_key: None, + profile: Some(profile.into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: Default::default(), + control_operation_id: None, + execution_grant: None, + } + } + + async fn snapshot(connection: &mut WorkspaceWorkerProtocolConnection) -> protocol::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); + return greeting; + } + } + }) + .await + .expect("real Controller did not publish a snapshot") + } + + async fn submit(connection: &mut WorkspaceWorkerProtocolConnection) { + connection + .methods + .send(protocol::Method::Submit { + submission_request_id: "controller-https-submit".into(), + input: vec![protocol::Segment::Text { + content: "Check the service status.".into(), + }], + }) + .await + .unwrap(); + } + + async fn submit_and_finish(connection: &mut WorkspaceWorkerProtocolConnection) { + submit(connection).await; + finish(connection, protocol::RunResult::Finished).await; + } + + async fn finish( + connection: &mut WorkspaceWorkerProtocolConnection, + expected: protocol::RunResult, + ) { + 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, "controller-https-submit"); + accepted = true; + } + protocol::Event::SubmissionRejected { message, .. } => { + panic!("Submit rejected: {message}") + } + protocol::Event::RunEnd { result } => { + assert!(accepted); + assert_eq!(result, expected); + finished = true; + } + protocol::Event::WorkerState { snapshot } + if finished + && snapshot.state == protocol::WorkerState::Idle + && snapshot.last_finished_submission_request_id.as_deref() + == Some("controller-https-submit") => + { + return; + } + _ => {} + } + } + }) + .await + .expect("real Controller HTTPS submission did not finish"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn owner_diagnostics_observes_saved_pin_and_live_refusal_without_execution_or_paths() { + let peer = Peer::new(HOST, vec![]).await; + let f = Fixture::new(&peer).await; + let (installed, _) = f.install(&peer).await; + let worker = f.create(PROFILE).await; + let path = format!( + "/api/w/{TEST_WORKSPACE_ID}/plugins/worker-diagnostics/{}/{}", + worker.runtime_id, worker.worker_id + ); + let router = build_router(f.api.clone()); + let reader = seed_test_api_token(f.api.store.as_ref(), "plugin-diagnostics-reader"); + request( + router.clone(), + None, + "GET", + &path, + vec![], + false, + StatusCode::UNAUTHORIZED, + ) + .await; + request( + router.clone(), + Some(&reader), + "GET", + &path, + vec![], + false, + StatusCode::FORBIDDEN, + ) + .await; + for revoked in [false, true] { + if revoked { + f.revoke(&installed).await; + } + let value = request( + router.clone(), + Some(&f.owner), + "GET", + &path, + vec![], + false, + StatusCode::OK, + ) + .await; + let text = value.to_string(); + for private in [ + HOST, + f.owner.as_str(), + f.api.config.embedded_runtime_store_root.to_str().unwrap(), + "base_url", + "secret_ref", + "cache_path", + ] { + assert!( + !text.contains(private), + "diagnostic exposed private transport or credential information" + ); + } + let observed: server_api::PluginWorkerDiagnostics = + serde_json::from_value(value).unwrap(); + let server_api::PluginWorkerDiagnosticSelection::Plan { + pin_state, plugins, .. + } = observed.selection + else { + panic!("expected saved plan") + }; + assert_eq!(pin_state, server_api::PluginDiagnosticPinState::Active); + assert_eq!(plugins.len(), 1); + assert_eq!( + plugins[0].installation_id, + installed.installation.installation_id + ); + assert_eq!( + plugins[0].surfaces, + vec![server_api::PluginDiagnosticSurface::Tool] + ); + assert_eq!(plugins[0].selected_tools, vec![TOOL]); + assert_eq!(plugins[0].connections[0].slot, "api"); + assert!(matches!( + plugins[0].connections[0].effective_limits, + server_api::PluginDiagnosticConnectionLimits::Available { .. } + )); + if revoked { + assert!(matches!( + plugins[0].admission, + server_api::PluginDiagnosticAdmission::Refused { + reason_code: server_api::PluginHostErrorCode::Denied + } + )); + } else { + assert_eq!( + plugins[0].admission, + server_api::PluginDiagnosticAdmission::Admissible + ); + } + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); + assert_eq!(peer.connection_count(), 0); + assert!(f.requests.lock().unwrap().is_empty()); + } + f.stop(&worker).await; + f.close().await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_builtin_create_uses_profile_factory_and_seals_empty_plugin_selection() { + let peer = Peer::new(HOST, vec![]).await; + let f = Fixture::new(&peer).await; + let worker = f.create("builtin:companion").await; + assert_eq!(f.model_clients.load(Ordering::SeqCst), 1); + assert!(f.requests.lock().unwrap().is_empty()); + 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))); + assert!(f.execution_request(&worker)["plugin_execution"].is_null()); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!( + !snapshot(&mut connection) + .await + .tools + .iter() + .any(|tool| tool.ends_with("__example_status")) + ); + assert_eq!(f.count("plugin_execution_plans"), 0); + assert_eq!(peer.connection_count(), 0); + drop(connection); + f.stop(&worker).await; + f.close().await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_project_controller_https_history_and_same_plan_restore_refetch_cold_cache() { + let mut peer = Peer::new(HOST, vec![response(503, &[], b"peer data")]).await; + let f = Fixture::new(&peer).await; + let (installed, package) = f.install(&peer).await; + let public_name = f.model_tool.lock().unwrap().clone(); + let worker = f.create(PROFILE).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().len(), 1); + assert_eq!( + saved.plan.plugins()[0].installation_id(), + installed.installation.installation_id + ); + let expected = runtime_api::PluginExecutionSnapshot { + plan_id: saved.plan_id.clone(), + fingerprint: saved.fingerprint.clone(), + plan: saved.plan.clone(), + }; + assert_eq!( + f.execution_request(&worker)["plugin_execution"], + serde_json::to_value(&expected).unwrap() + ); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + let greeting = snapshot(&mut connection).await; + assert!(greeting.tools.contains(&public_name)); + assert!(!greeting.tools.iter().any(|tool| tool == TOOL)); + assert_eq!( + f.count("plugin_invocations"), + 0, + "preparation must not invoke a Tool" + ); + submit_and_finish(&mut connection).await; + let effect_states: Vec<(String,Option)> = f.api.config_store.with_conn(|conn| { + let mut statement = conn.prepare("SELECT state,json_extract(error_json,'$.code') FROM plugin_effect_receipts WHERE workspace_id=?1")?; + Ok(statement.query_map([TEST_WORKSPACE_ID], |row| Ok((row.get(0)?,row.get(1)?)))?.collect::,_>>()?) + }).unwrap(); + assert_eq!( + effect_states, + vec![("completed".into(), None)], + "HTTPS receipt observation before history assertions" + ); + { + let requests = f.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert!( + requests[0] + .tools + .iter() + .any(|tool| tool.name == public_name) + ); + assert_https_history(&requests[1].items, &public_name); + } + assert!( + String::from_utf8( + tokio::time::timeout(WATCHDOG, peer.observed_within(WATCHDOG)) + .await + .unwrap() + ) + .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); + drop(connection); + f.stop(&worker).await; + let (session, history) = f.saved_history(&worker); + assert_https_history(&history, &public_name); + 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 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::() + ); + assert_eq!( + serde_json::from_str::(&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"); + + // Current Profile now selects nothing. Normal restore must not resolve + // it again, nor treat previously cached bytes as a restored selection. + f.configure(&installed, &peer, false).await; + let current = f + .api + .config_store + .load_workspace_config(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let reselected = f + .api + .config_store + .with_immediate_transaction(|tx| { + crate::plugin_settings::resolve_worker_plugin_plan( + tx, + TEST_WORKSPACE_ID, + ¤t.snapshot.digest, + PROFILE, + saved.plan.execution().clone(), + ) + }) + .unwrap(); + assert!(reselected.plugins().is_empty()); + let cache_file = f.cached_artifact(&package); + fs::remove_file(&cache_file).unwrap(); + assert!(!cache_file.exists()); + let restored = f.restore(&worker, "same-plan-controller-restore").await; + assert_eq!( + restored.state, + server_api::WorkerRestoreState::Accepted, + "{restored:?}" + ); + assert_eq!(f.model_clients.load(Ordering::SeqCst), 2); + assert_eq!( + f.requests.lock().unwrap().len(), + 2, + "restore must not submit another model turn" + ); + assert_eq!( + fs::read(&cache_file).unwrap(), + package, + "cold restore must refetch and verify the exact package" + ); + assert_eq!( + f.execution_request(&worker)["plugin_execution"], + serde_json::to_value(&expected).unwrap() + ); + let retained = f + .api + .config_store + .plugin_plan_load(TEST_WORKSPACE_ID, &saved.plan_id) + .unwrap(); + assert_eq!(retained.pin_state, PlanPinState::Active); + assert_eq!(retained.fingerprint, saved.fingerprint); + assert_eq!(retained.plan, saved.plan); + assert_eq!( + f.count("plugin_execution_plans"), + 1, + "restore must not acquire a new plan" + ); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!( + snapshot(&mut connection).await.tools.contains(&public_name), + "restored Controller must publish the saved Tool even though current selection is empty" + ); + assert_eq!( + peer.connection_count(), + 1, + "restore preparation must not repeat the HTTP effect" + ); + assert_eq!(f.count("plugin_invocations"), 1); + assert_eq!(f.count("plugin_effect_receipts"), 1); + drop(connection); + f.stop(&worker).await; + let (restored_session, history) = f.saved_history(&worker); + assert_eq!(restored_session, session); + assert_https_history(&history, &public_name); + f.close().await; + } + + fn assert_failed_tool_history(items: &[agen::Item], public_name: &str) { + let calls = items + .iter() + .filter_map(|item| match item { + agen::Item::ToolCall { call_id, name, .. } => { + Some((call_id.as_str(), name.as_str())) + } + _ => None, + }) + .collect::>(); + assert_eq!(calls, vec![("https-call", public_name)]); + let results = items + .iter() + .filter_map(|item| match item { + agen::Item::ToolResult { + call_id, + disposition, + is_error, + content, + .. + } => Some((call_id.as_str(), disposition, is_error, content)), + _ => None, + }) + .collect::>(); + assert_eq!( + results.len(), + 1, + "denial must be persisted as the ordinary Tool result" + ); + assert_eq!(results[0].0, "https-call"); + assert_ne!(*results[0].1, agen::ToolResultDisposition::Success); + assert!(*results[0].2); + assert!( + !results[0].3.as_deref().unwrap_or("").contains("peer data"), + "a fenced response must not leak into ordinary Tool history" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_pure_and_https_registered_tool_rechecks_revoke_and_persists_denial() { + for https in [false, true] { + let peer = Peer::new(HOST, vec![]).await; + let f = Fixture::new(&peer).await; + let (installed, package) = f.install_component(&peer, https).await; + let worker = f.create(PROFILE).await; + let public_name = f.model_tool.lock().unwrap().clone(); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!(snapshot(&mut connection).await.tools.contains(&public_name)); + let cache_file = f.cached_artifact(&package); + f.revoke(&installed).await; + submit_and_finish(&mut connection).await; + { + let requests = f.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert!( + requests[0] + .tools + .iter() + .any(|tool| tool.name == public_name) + ); + assert_failed_tool_history(&requests[1].items, &public_name); + assert!( + requests[1].items.iter().any(|item| matches!(item, + agen::Item::ToolResult { summary, .. } + if summary.contains("Plugin live admission refused or unavailable"))), + "revoke must fail at live admission, not at schema validation or missing Tool lookup" + ); + } + assert_eq!( + f.count("plugin_invocations"), + 0, + "registered/cached must not authorize a fresh call" + ); + assert_eq!(f.count("plugin_effect_receipts"), 0); + assert_eq!(peer.connection_count(), 0); + assert_eq!( + fs::read(cache_file).unwrap(), + package, + "revoke denial must not rely on removing bytes" + ); + drop(connection); + f.stop(&worker).await; + assert_failed_tool_history(&f.saved_history(&worker).1, &public_name); + f.close().await; + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_dispatched_https_revoke_withholds_late_response_from_model_and_saved_history() + { + use crate::plugin_https::test_support::Reply; + let (release, response_ready) = tokio::sync::oneshot::channel(); + let mut peer = Peer::new(HOST, vec![Reply::Gated(response_ready)]).await; + let f = Fixture::new(&peer).await; + let (installed, _) = f.install(&peer).await; + let worker = f.create(PROFILE).await; + let public_name = f.model_tool.lock().unwrap().clone(); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!(snapshot(&mut connection).await.tools.contains(&public_name)); + submit(&mut connection).await; + let observed = tokio::time::timeout(WATCHDOG, peer.observed_within(WATCHDOG)) + .await + .unwrap(); + assert!( + String::from_utf8(observed) + .unwrap() + .starts_with("GET /status HTTP/1.1\r\n") + ); + assert_eq!(f.count("plugin_invocations"), 1); + assert_eq!( + f.count("plugin_effect_receipts"), + 1, + "dispatch must already have its durable effect key" + ); + f.revoke(&installed).await; + // The request has already reached the external peer. Release a response + // only after revocation commits: this asserts fencing, not rollback. + release + .send(match response(503, &[], b"peer data") { + Reply::Bytes(bytes) => bytes, + _ => unreachable!(), + }) + .unwrap(); + finish(&mut connection, protocol::RunResult::Finished).await; + { + let requests = f.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_failed_tool_history(&requests[1].items, &public_name); + } + assert_eq!( + peer.connection_count(), + 1, + "no fresh effect key may resend a revoked call" + ); + assert_eq!(f.count("plugin_invocations"), 1); + assert_eq!(f.count("plugin_effect_receipts"), 1); + drop(connection); + f.stop(&worker).await; + assert_failed_tool_history(&f.saved_history(&worker).1, &public_name); + f.close().await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_controller_cancel_closes_inflight_https_and_restore_does_not_resend() { + use crate::plugin_https::test_support::Reply; + let mut peer = Peer::new(HOST, vec![Reply::Stall]).await; + let f = Fixture::new(&peer).await; + f.install(&peer).await; + let worker = f.create(PROFILE).await; + let public_name = f.model_tool.lock().unwrap().clone(); + let persisted = f.execution_request(&worker)["plugin_execution"].clone(); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!(snapshot(&mut connection).await.tools.contains(&public_name)); + submit(&mut connection).await; + tokio::time::timeout(WATCHDOG, peer.observed_within(WATCHDOG)) + .await + .unwrap(); + assert_eq!(f.count("plugin_invocations"), 1); + assert_eq!(f.count("plugin_effect_receipts"), 1); + let effect = f.effect_identity(&worker); + assert_eq!(effect.1, "status-v1"); + connection + .methods + .send(protocol::Method::Cancel { + command: protocol::WorkerCommandEnvelope::new(901), + }) + .await + .unwrap(); + finish(&mut connection, protocol::RunResult::Cancelled).await; + peer.closed().await; + let cancelled: bool = f + .api + .config_store + .with_conn(|conn| { + Ok(conn.query_row( + "SELECT cancelled FROM plugin_invocations WHERE workspace_id=?1 AND invocation_id=?2", + rusqlite::params![TEST_WORKSPACE_ID, effect.0], |row| row.get(0))?) + }) + .unwrap(); + assert!( + cancelled, + "Controller cancellation must reach the durable Host owner" + ); + assert_eq!( + f.requests.lock().unwrap().len(), + 1, + "cancelled result must not start another model turn" + ); + drop(connection); + f.stop(&worker).await; + let (session, history) = f.saved_history(&worker); + assert_failed_tool_history(&history, &public_name); + let restored = f.restore(&worker, "cancelled-controller-restore").await; + assert_eq!( + restored.state, + server_api::WorkerRestoreState::Accepted, + "{restored:?}" + ); + assert_eq!(f.execution_request(&worker)["plugin_execution"], persisted); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!(snapshot(&mut connection).await.tools.contains(&public_name)); + assert_eq!( + f.requests.lock().unwrap().len(), + 1, + "restore must not implicitly resume a cancelled submission" + ); + assert_eq!( + peer.connection_count(), + 1, + "restore must not resend the indeterminate HTTP effect" + ); + assert_eq!( + f.effect_identity(&worker), + effect, + "restore must retain the original invocation/effect-key fingerprint" + ); + assert_eq!(f.count("plugin_invocations"), 1); + assert_eq!(f.count("plugin_effect_receipts"), 1); + drop(connection); + f.stop(&worker).await; + let (restored_session, history) = f.saved_history(&worker); + assert_eq!(restored_session, session); + assert_failed_tool_history(&history, &public_name); + f.close().await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_pure_and_https_revoked_restore_refuses_warm_and_cold_saved_plan_without_controller() + { + for https in [false, true] { + for cold in [false, true] { + let peer = Peer::new(HOST, vec![]).await; + let f = Fixture::new(&peer).await; + let (installed, package) = f.install_component(&peer, https).await; + let worker = f.create(PROFILE).await; + let public_name = f.model_tool.lock().unwrap().clone(); + let persisted = f.execution_request(&worker)["plugin_execution"].clone(); + let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .unwrap(); + assert!(snapshot(&mut connection).await.tools.contains(&public_name)); + drop(connection); + f.stop(&worker).await; + let plan_id = persisted["plan_id"].as_str().unwrap(); + let before = f + .api + .config_store + .plugin_plan_load(TEST_WORKSPACE_ID, plan_id) + .unwrap(); + assert_ne!( + before.pin_state, + PlanPinState::Released, + "stop must retain the restorable pin" + ); + let cache_file = f.cached_artifact(&package); + if cold { + fs::remove_file(&cache_file).unwrap(); + } + f.revoke(&installed).await; + let restored = f.restore(&worker, "revoked-controller-restore").await; + assert!( + matches!( + restored.state, + server_api::WorkerRestoreState::Rejected + | server_api::WorkerRestoreState::RolledBack + ), + "{https}/{cold}: {restored:?}" + ); + assert!( + restored.diagnostics.iter().any(|diagnostic| diagnostic + .message + .contains("Plugin restore preparation refused or unavailable")), + "must fail in the production Plugin restore preparation, not an unrelated prerequisite: {restored:?}" + ); + assert!( + connect_workspace_worker_protocol(&f.api, &worker, None) + .await + .is_err(), + "failed preparation must not publish a usable Controller" + ); + assert_eq!(f.execution_request(&worker)["plugin_execution"], persisted); + let after = f + .api + .config_store + .plugin_plan_load(TEST_WORKSPACE_ID, plan_id) + .unwrap(); + assert_eq!( + after.plan, before.plan, + "denial must not reselect current config" + ); + assert_eq!(after.fingerprint, before.fingerprint); + assert_eq!( + after.pin_state, before.pin_state, + "failed restore must not lose the pin" + ); + assert_eq!(f.count("plugin_execution_plans"), 1); + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); + assert!( + f.requests.lock().unwrap().is_empty(), + "prepare must not submit a model turn" + ); + assert_eq!(peer.connection_count(), 0); + if !cold { + assert_eq!( + fs::read(&cache_file).unwrap(), + package, + "warm cache must not bypass current admission" + ); + } + f.close().await; + } + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn embedded_project_create_refuses_missing_artifact_in_real_factory_and_releases_pin() { + let peer = Peer::new(HOST, vec![]).await; + let f = Fixture::new(&peer).await; + let (installed, _) = f.install(&peer).await; + // Keep catalog/config valid, but simulate unavailable authoritative bytes. + // This is deliberately a delivery refusal, not a claim of policy denial. + let digest = &installed.installation.artifacts[0].artifact_digest; + let blob = ServerConfig::server_data_root_for_data_dir(&f.api.config.data_root) + .join("plugin-artifacts/blobs") + .join(digest.strip_prefix("sha256:").unwrap()); + assert!(blob.is_file()); + fs::remove_file(blob).unwrap(); + assert!( + create_workspace_worker( + State(f.api.clone()), + HeaderMap::new(), + Json(create_request(PROFILE)) + ) + .await + .is_err() + ); + assert_eq!( + f.model_clients.load(Ordering::SeqCst), + 1, + "failure must reach the production factory, not fail in Profile selection before dispatch" + ); + assert!(f.requests.lock().unwrap().is_empty()); + 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, + &saved.plan.execution().worker_id, + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_none() + ); + assert!( + f.api.runtime.worker(&worker).is_err(), + "failed startup must not publish a Controller" + ); + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); + assert_eq!(peer.connection_count(), 0); + f.close().await; + } +} diff --git a/crates/workspace-server/src/server_plugin_diagnostics.rs b/crates/workspace-server/src/server_plugin_diagnostics.rs new file mode 100644 index 000000000..18acc4eef --- /dev/null +++ b/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) -> Result { + value + .try_into() + .map_err(|_| Error::Config("Plugin Runtime budget exceeds diagnostic wire range".into())) +} + +fn runtime_limits(limits: &yoi_plugin_runtime::Limits) -> Result { + 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 { + // 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, + } + ); + } + } +} diff --git a/crates/workspace-server/src/server_plugin_host.rs b/crates/workspace-server/src/server_plugin_host.rs index 627ede4e1..759d05952 100644 --- a/crates/workspace-server/src/server_plugin_host.rs +++ b/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>, + ) -> Self { + self.transport_override = Some(transport.into()); self } } diff --git a/crates/workspace-server/src/server_plugin_host_tests.rs b/crates/workspace-server/src/server_plugin_host_tests.rs index 5954c04f2..a9bd12736 100644 --- a/crates/workspace-server/src/server_plugin_host_tests.rs +++ b/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(); diff --git a/crates/workspace-server/src/server_plugin_remote_controller_tests.rs b/crates/workspace-server/src/server_plugin_remote_controller_tests.rs new file mode 100644 index 000000000..ea7bf4a1f --- /dev/null +++ b/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>, + requests: Arc>>, + } + + #[async_trait::async_trait] + impl LlmClient for InstalledToolModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + request: ModelRequest, + ) -> std::result::Result { + 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) { + 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::>(); + 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) = 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::() + ); + assert_eq!( + serde_json::from_str::(&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(); + } +} diff --git a/crates/workspace-server/src/server_plugin_runtime.rs b/crates/workspace-server/src/server_plugin_runtime.rs index af9a08c60..38ccac97d 100644 --- a/crates/workspace-server/src/server_plugin_runtime.rs +++ b/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), diff --git a/crates/workspace-server/src/server_plugin_runtime_tests.rs b/crates/workspace-server/src/server_plugin_runtime_tests.rs index a6f83feca..0069b21b5 100644 --- a/crates/workspace-server/src/server_plugin_runtime_tests.rs +++ b/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>>, + pub(in crate::server::tests) struct Model { + pub(in crate::server::tests) tool_name: String, + pub(in crate::server::tests) requests: Arc>>, } #[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"); diff --git a/crates/workspace-server/src/server_plugin_tests.rs b/crates/workspace-server/src/server_plugin_tests.rs index 529096761..c5eee0607 100644 --- a/crates/workspace-server/src/server_plugin_tests.rs +++ b/crates/workspace-server/src/server_plugin_tests.rs @@ -25,7 +25,7 @@ fn package_bytes(version: &str) -> Vec { .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); } diff --git a/crates/workspace-server/src/subject_assignment_migration_tests.rs b/crates/workspace-server/src/subject_assignment_migration_tests.rs index d52a0ffb2..ac933ebd4 100644 --- a/crates/workspace-server/src/subject_assignment_migration_tests.rs +++ b/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 } diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index d5f820775..928e45fb4 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/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"), diff --git a/openapi/server-api.json b/openapi/server-api.json index 914cb5a28..716035fd0 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -7587,6 +7587,131 @@ } ] }, + "PluginDiagnosticAdmission": { + "description": "A successful observation is not retained admission authority.", + "oneOf": [ + { + "additionalProperties": false, + "properties": { + "kind": { + "const": "admissible", + "type": "string" + } + }, + "required": [ + "kind" + ], + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "kind": { + "const": "refused", + "type": "string" + }, + "reason_code": { + "$ref": "#/components/schemas/PluginHostErrorCode" + } + }, + "required": [ + "kind", + "reason_code" + ], + "type": "object" + } + ] + }, + "PluginDiagnosticConnection": { + "additionalProperties": false, + "properties": { + "effective_limits": { + "$ref": "#/components/schemas/PluginDiagnosticConnectionLimits" + }, + "slot": { + "type": "string" + } + }, + "required": [ + "slot", + "effective_limits" + ], + "type": "object" + }, + "PluginDiagnosticConnectionLimits": { + "description": "Limits are absent when the current binding cannot be admitted; never fall\nback to a launch-time policy or a permissive default.", + "oneOf": [ + { + "additionalProperties": false, + "properties": { + "kind": { + "const": "available", + "type": "string" + }, + "max_header_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "max_request_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "max_response_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "timeout_ms": { + "format": "uint32", + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "kind", + "timeout_ms", + "max_request_bytes", + "max_response_bytes", + "max_header_bytes" + ], + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "kind": { + "const": "unavailable", + "type": "string" + }, + "reason_code": { + "$ref": "#/components/schemas/PluginHostErrorCode" + } + }, + "required": [ + "kind", + "reason_code" + ], + "type": "object" + } + ] + }, + "PluginDiagnosticPinState": { + "enum": [ + "pending", + "active", + "restorable", + "released" + ], + "type": "string" + }, + "PluginDiagnosticSurface": { + "enum": [ + "tool" + ], + "type": "string" + }, "PluginEffectObservation": { "enum": [ "not_dispatched", @@ -8472,6 +8597,105 @@ ], "type": "object" }, + "PluginRuntimeLimits": { + "additionalProperties": false, + "description": "Fixed Tool Runtime budgets, not usage counters or caller-selectable limits.", + "properties": { + "config_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "fuel": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "host_operations": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "http_body_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "http_header_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "http_header_pairs": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "instances": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "log_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "memories": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "memory_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "payload_bytes": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "resources": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "table_elements": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "tables": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "wall_time_ms": { + "format": "uint32", + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "payload_bytes", + "config_bytes", + "memory_bytes", + "table_elements", + "instances", + "memories", + "tables", + "resources", + "fuel", + "wall_time_ms", + "host_operations", + "http_body_bytes", + "http_header_bytes", + "http_header_pairs", + "log_bytes" + ], + "type": "object" + }, "PluginSchemaContribution": { "additionalProperties": false, "properties": { @@ -8567,6 +8791,158 @@ ], "type": "object" }, + "PluginWorkerDiagnostic": { + "additionalProperties": false, + "properties": { + "admission": { + "$ref": "#/components/schemas/PluginDiagnosticAdmission" + }, + "artifact_digest": { + "type": "string" + }, + "artifact_size": { + "format": "uint32", + "minimum": 0, + "type": "integer" + }, + "connections": { + "items": { + "$ref": "#/components/schemas/PluginDiagnosticConnection" + }, + "type": "array" + }, + "installation_id": { + "type": "string" + }, + "package_digest": { + "type": "string" + }, + "package_id": { + "type": "string" + }, + "package_version": { + "type": "string" + }, + "selected_tools": { + "items": { + "type": "string" + }, + "type": "array" + }, + "surfaces": { + "items": { + "$ref": "#/components/schemas/PluginDiagnosticSurface" + }, + "type": "array" + } + }, + "required": [ + "installation_id", + "package_id", + "package_version", + "package_digest", + "artifact_digest", + "artifact_size", + "surfaces", + "selected_tools", + "admission", + "connections" + ], + "type": "object" + }, + "PluginWorkerDiagnosticSelection": { + "oneOf": [ + { + "additionalProperties": false, + "description": "No sealed launch selection exists; do not infer current Profile selection.", + "properties": { + "kind": { + "const": "unrecorded", + "type": "string" + } + }, + "required": [ + "kind" + ], + "type": "object" + }, + { + "additionalProperties": false, + "description": "The owner sealed an explicitly empty selection (a NULL plan).", + "properties": { + "kind": { + "const": "empty", + "type": "string" + } + }, + "required": [ + "kind" + ], + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "fingerprint": { + "type": "string" + }, + "kind": { + "const": "plan", + "type": "string" + }, + "pin_state": { + "$ref": "#/components/schemas/PluginDiagnosticPinState" + }, + "plan_id": { + "type": "string" + }, + "plugins": { + "items": { + "$ref": "#/components/schemas/PluginWorkerDiagnostic" + }, + "type": "array" + } + }, + "required": [ + "kind", + "plan_id", + "fingerprint", + "pin_state", + "plugins" + ], + "type": "object" + } + ] + }, + "PluginWorkerDiagnostics": { + "additionalProperties": false, + "description": "Point-in-time owner observation only, never a permission or execution lease.\nDeliberately excludes configuration, credentials and artifact locations.", + "properties": { + "runtime_id": { + "type": "string" + }, + "runtime_limits": { + "$ref": "#/components/schemas/PluginRuntimeLimits" + }, + "selection": { + "$ref": "#/components/schemas/PluginWorkerDiagnosticSelection" + }, + "worker_id": { + "type": "string" + }, + "workspace_id": { + "type": "string" + } + }, + "required": [ + "workspace_id", + "runtime_id", + "worker_id", + "runtime_limits", + "selection" + ], + "type": "object" + }, "ProfileSettingsResponse": { "additionalProperties": false, "description": "Read-only Profile catalog projected from one active Workspace config tree.", @@ -19847,7 +20223,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80" + "x-yoi-source-digest": "sha256:3e616b1d2ec6b52f7504cf0f81f30c976cd6a18e510084ea50f4f9e340089581" }, "openapi": "3.1.0", "paths": { @@ -29977,6 +30353,113 @@ ] } }, + "/api/w/{workspace_id}/plugins/worker-diagnostics/{runtime_id}/{worker_id}": { + "get": { + "operationId": "plugin_worker_diagnostics", + "parameters": [ + { + "explode": false, + "in": "path", + "name": "workspace_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + }, + { + "explode": false, + "in": "path", + "name": "runtime_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + }, + { + "explode": false, + "in": "path", + "name": "worker_id", + "required": true, + "schema": { + "$ref": "#/components/schemas/string" + }, + "style": "simple" + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/PluginWorkerDiagnostics" + } + } + }, + "description": "Successful response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RepositoryApiError" + } + } + }, + "description": "Error response" + } + }, + "security": [ + { + "bearerAuth": [] + }, + { + "browserSession": [] + } + ] + } + }, "/api/w/{workspace_id}/plugins/{installation_id}": { "get": { "operationId": "plugin_show", diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index e9fec1102..4d69eb652 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) +// Input contract: 0.1.0 (sha256:3e616b1d2ec6b52f7504cf0f81f30c976cd6a18e510084ea50f4f9e340089581) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/plugin-api.ts b/web/workspace/src/lib/generated/plugin-api.ts index d1178a014..0844960f8 100644 --- a/web/workspace/src/lib/generated/plugin-api.ts +++ b/web/workspace/src/lib/generated/plugin-api.ts @@ -38,3 +38,23 @@ export type PluginOperationResponse = { operation_id: string, kind: PluginOperat export type PluginArtifactDeliveryRequest = { plan_id: string, plan_fingerprint: string, installation_id: string, artifact_digest: string, }; export type PluginArtifactDeliveryMetadata = { workspace_id: string, request: PluginArtifactDeliveryRequest, artifact: PluginArtifact, }; + +export type PluginHostErrorCode = "invalid_input" | "denied" | "not_found" | "conflict" | "unsupported" | "quota_exceeded" | "deadline_exceeded" | "cancelled" | "unavailable" | "indeterminate" | "receipt_expired"; + +export type PluginWorkerDiagnostics = { workspace_id: string, runtime_id: string, worker_id: string, runtime_limits: PluginRuntimeLimits, selection: PluginWorkerDiagnosticSelection, }; + +export type PluginWorkerDiagnosticSelection = { "kind": "unrecorded" } | { "kind": "empty" } | { "kind": "plan", plan_id: string, fingerprint: string, pin_state: PluginDiagnosticPinState, plugins: Array, }; + +export type PluginDiagnosticPinState = "pending" | "active" | "restorable" | "released"; + +export type PluginDiagnosticSurface = "tool"; + +export type PluginWorkerDiagnostic = { installation_id: string, package_id: string, package_version: string, package_digest: string, artifact_digest: string, artifact_size: number, surfaces: Array, selected_tools: Array, admission: PluginDiagnosticAdmission, connections: Array, }; + +export type PluginDiagnosticAdmission = { "kind": "admissible" } | { "kind": "refused", reason_code: PluginHostErrorCode, }; + +export type PluginDiagnosticConnection = { slot: string, effective_limits: PluginDiagnosticConnectionLimits, }; + +export type PluginDiagnosticConnectionLimits = { "kind": "available", timeout_ms: number, max_request_bytes: number, max_response_bytes: number, max_header_bytes: number, } | { "kind": "unavailable", reason_code: PluginHostErrorCode, }; + +export type PluginRuntimeLimits = { payload_bytes: number, config_bytes: number, memory_bytes: number, table_elements: number, instances: number, memories: number, tables: number, resources: number, fuel: number, wall_time_ms: number, host_operations: number, http_body_bytes: number, http_header_bytes: number, http_header_pairs: number, log_bytes: number, }; diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 37d4c5a9a..d4c2b02c0 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) +// Input contract: 0.1.0 (sha256:3e616b1d2ec6b52f7504cf0f81f30c976cd6a18e510084ea50f4f9e340089581) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index 3a5a024cb..6ff883206 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) +// Input contract: 0.1.0 (sha256:3e616b1d2ec6b52f7504cf0f81f30c976cd6a18e510084ea50f4f9e340089581) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index 3b3f30934..9a01db415 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) +// Input contract: 0.1.0 (sha256:3e616b1d2ec6b52f7504cf0f81f30c976cd6a18e510084ea50f4f9e340089581) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index d17adc8e8..a2f76f22e 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) +// Input contract: 0.1.0 (sha256:3e616b1d2ec6b52f7504cf0f81f30c976cd6a18e510084ea50f4f9e340089581) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check