From 165153c63cde58f3a0fb5cbffa3dee83466493c3 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 04:42:53 +0900 Subject: [PATCH] feat: stage exact Plugin Tools through shared Worker composition Extract the unchanged canonical plan model for typed Runtime snapshots without a Server dependency cycle. Bind Tool consumer identities to persisted Session/call IDs, use the owner-issued context and current publication authorization, and register through FeatureInstallPlan with cooperative shutdown. Validation so far: root cargo check --locked, portable model 6 tests/doctest, Worker actual Component adapter 7 focused tests, Server plan 9 tests via shared isolation. Full product launch owner/pin lifecycle and Remote/Embedded vertical integration remain a subsequent coherent change. --- Cargo.lock | 17 + Cargo.toml | 3 + crates/plugin-plan/Cargo.toml | 15 + crates/plugin-plan/README.md | 5 + .../src/fixtures/canonical-plan.json | 1 + crates/plugin-plan/src/lib.rs | 446 ++++++++++++ crates/plugin-plan/src/tests.rs | 204 ++++++ crates/plugin-runtime/src/executor.rs | 5 + crates/runtime-api/Cargo.toml | 1 + crates/runtime-api/src/lib.rs | 13 + crates/server-api/Cargo.toml | 1 + crates/server-api/src/plugin.rs | 38 + crates/worker-runtime/Cargo.toml | 1 + crates/worker-runtime/src/catalog.rs | 2 + crates/worker-runtime/src/http_server.rs | 2 + crates/worker-runtime/src/lib.rs | 1 + crates/worker-runtime/src/plugin.rs | 87 +++ crates/worker-runtime/src/runtime.rs | 18 + .../worker-runtime/src/test_support_tests.rs | 1 + crates/worker-runtime/src/worker_backend.rs | 51 ++ crates/worker/Cargo.toml | 1 + crates/worker/src/controller.rs | 12 + crates/worker/src/lib.rs | 1 + crates/worker/src/plugin.rs | 389 ++++++++++ crates/worker/src/plugin/tests.rs | 676 ++++++++++++++++++ crates/worker/src/worker.rs | 30 + crates/workspace-server/Cargo.toml | 2 +- crates/workspace-server/src/hosts.rs | 3 + crates/workspace-server/src/lib.rs | 2 +- crates/workspace-server/src/plugin_plan.rs | 392 +--------- .../workspace-server/src/plugin_plan/store.rs | 36 +- .../workspace-server/src/plugin_plan/tests.rs | 67 +- .../workspace-server/src/plugin_settings.rs | 1 + .../src/runtime_subscription_tests.rs | 1 + crates/workspace-server/src/server.rs | 8 +- openapi/server-api.json | 2 +- .../src/lib/generated/companion-api.ts | 2 +- .../src/lib/generated/repository-api.ts | 2 +- .../src/lib/generated/runtime-api.ts | 2 +- web/workspace/src/lib/generated/ticket-api.ts | 2 +- .../src/lib/generated/workdir-api.ts | 2 +- .../src/lib/generated/worker-launch-api.ts | 2 +- 42 files changed, 2123 insertions(+), 424 deletions(-) create mode 100644 crates/plugin-plan/Cargo.toml create mode 100644 crates/plugin-plan/README.md create mode 100644 crates/plugin-plan/src/fixtures/canonical-plan.json create mode 100644 crates/plugin-plan/src/lib.rs create mode 100644 crates/plugin-plan/src/tests.rs create mode 100644 crates/worker-runtime/src/plugin.rs create mode 100644 crates/worker/src/plugin.rs create mode 100644 crates/worker/src/plugin/tests.rs diff --git a/Cargo.lock b/Cargo.lock index 27472dcb8..6affd33b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3648,6 +3648,18 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "plugin-plan" +version = "0.1.0" +dependencies = [ + "manifest", + "schemars", + "serde", + "serde_json", + "sha2 0.11.0", + "thiserror 2.0.18", +] + [[package]] name = "poly1305" version = "0.8.0" @@ -4260,6 +4272,7 @@ version = "0.1.0" dependencies = [ "api-macros", "manifest", + "plugin-plan", "protocol", "serde", "serde_json", @@ -4699,6 +4712,7 @@ dependencies = [ "api-macros", "memory", "merge-request", + "plugin-plan", "protocol", "runtime-api", "schemars", @@ -7287,6 +7301,7 @@ dependencies = [ "wip-protocol", "wip-text-view", "workdir", + "yoi-plugin-runtime", ] [[package]] @@ -7303,6 +7318,7 @@ dependencies = [ "futures", "libc", "manifest", + "plugin-plan", "protocol", "reqwest", "ring", @@ -7508,6 +7524,7 @@ dependencies = [ "memory", "merge-request", "native-tls", + "plugin-plan", "project-record", "protocol", "reqwest", diff --git a/Cargo.toml b/Cargo.toml index 38c7516a2..400d2e489 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ members = [ "crates/runtime-api", "crates/server-api", "crates/plugin-pdk", + "crates/plugin-plan", "crates/plugin-runtime", "crates/yoi", "crates/protocol", @@ -58,6 +59,7 @@ default-members = [ "crates/runtime-api", "crates/server-api", "crates/plugin-pdk", + "crates/plugin-plan", "crates/plugin-runtime", "crates/yoi", "crates/protocol", @@ -122,6 +124,7 @@ project-record = { path = "crates/project-record" } worker = { path = "crates/worker" } worker-runtime = { path = "crates/worker-runtime" } yoi-plugin-pdk = { path = "crates/plugin-pdk" } +plugin-plan = { path = "crates/plugin-plan" } yoi = { path = "crates/yoi" } protocol = { path = "crates/protocol" } session-metrics = { path = "crates/session-metrics" } diff --git a/crates/plugin-plan/Cargo.toml b/crates/plugin-plan/Cargo.toml new file mode 100644 index 000000000..6b88ed29d --- /dev/null +++ b/crates/plugin-plan/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "plugin-plan" +version = "0.1.0" +edition.workspace = true +license.workspace = true +rust-version.workspace = true +publish = false + +[dependencies] +manifest.workspace = true +schemars.workspace = true +serde = { workspace = true, features = ["derive"] } +serde_json.workspace = true +sha2.workspace = true +thiserror.workspace = true diff --git a/crates/plugin-plan/README.md b/crates/plugin-plan/README.md new file mode 100644 index 000000000..1c765d268 --- /dev/null +++ b/crates/plugin-plan/README.md @@ -0,0 +1,5 @@ +# Portable Plugin execution plan + +This crate owns the immutable [`WorkerPluginExecutionPlan`] model, validation and canonical fingerprint shared by Server persistence and Runtime launch snapshots. [`ArtifactDescriptor`] projects only the artifact metadata needed to construct or compare exact pins without depending on Server API types. + +It does not own catalog admission, persistence, configuration resolution, artifact delivery or authorization. A deserialized plan or owner-supplied artifact projection is not evidence of trust or live execution authority; callers must enforce those boundaries with their owners. diff --git a/crates/plugin-plan/src/fixtures/canonical-plan.json b/crates/plugin-plan/src/fixtures/canonical-plan.json new file mode 100644 index 000000000..f8866d427 --- /dev/null +++ b/crates/plugin-plan/src/fixtures/canonical-plan.json @@ -0,0 +1 @@ +{"format_version":1,"input":{"config":{"content_digest":"sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd","toolchain_fingerprint":"toolchain"},"execution":{"delegation_id":"delegation","principal_id":"principal","runtime_id":"runtime","worker_id":"worker"},"plugins":[{"artifact_digest":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","artifact_size":42,"component_world":"yoi:plugin/https-logged-tool-plugin@2.0.0","config_digest":"sha256:4bf0a2da94b0c881343726db532ddbfeeed37790af7da44e790cc0bba708654e","config_schema":{"digest":"sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","identity":"config","namespace":"example","version":"1"},"expected_last_operation":"operation","installation_id":"installation","package_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","package_id":"package","package_schema_version":2,"package_version":"1.0.0","projection":{"bindings":[{"binding_id":"binding","policy_digest":"sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff","slot":"api"}],"config":{"a":[{"a":false,"z":0}],"z":{"a":1,"b":2}},"grants":[{"kind":"logging"},{"kind":"https","slot":"api"}],"surfaces":["tool"],"tools":["echo","other"]},"required_imports":["https","logging"]}],"profile":{"identity":"profile:coder","projection_digest":"sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"},"workspace_id":"workspace"}} diff --git a/crates/plugin-plan/src/lib.rs b/crates/plugin-plan/src/lib.rs new file mode 100644 index 000000000..2ee98754e --- /dev/null +++ b/crates/plugin-plan/src/lib.rs @@ -0,0 +1,446 @@ +#![doc = include_str!("../README.md")] + +use manifest::plugin; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::collections::BTreeSet; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +pub enum PlanError { + #[error("invalid Plugin execution plan")] + Invalid, + #[error("Plugin execution plan conflict")] + Conflict, +} + +pub type Result = std::result::Result; + +/// Neutral metadata used to validate and reproduce an exact pin. This excludes +/// descriptions, Tool schemas, schema source, credentials and artifact locations. +/// A projection is not proof of catalog admission or current authorization. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArtifactIdentityProjection { + pub package_id: String, + pub version: String, + pub package_digest: String, + pub artifact_digest: String, + pub artifact_size: u32, + pub schema_version: u32, + pub world: String, + pub required_imports: Vec, + pub tools: Vec, + pub connections: Vec, + pub requested_grants: Vec, + pub config_schema: PluginSchemaIdentity, + pub revoked: bool, +} + +/// Adapts an owner's artifact metadata without coupling the portable model to +/// that owner's API. Callers must still perform admission with the owner; an +/// implementation of this trait does not establish metadata trust. +pub trait ArtifactDescriptor { + fn artifact_identity(&self) -> ArtifactIdentityProjection; +} + +impl ArtifactDescriptor for ArtifactIdentityProjection { + fn artifact_identity(&self) -> ArtifactIdentityProjection { + self.clone() + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct ConfigProvenance { + pub content_digest: String, + pub toolchain_fingerprint: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct ProfileProvenance { + pub identity: String, + pub projection_digest: String, +} + +/// Supplied by the authenticated Host launch/delegation boundary, never by guest +/// or model input. These are identities only, not credentials or provider handles. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct HostExecutionIdentity { + pub runtime_id: String, + pub worker_id: String, + pub principal_id: String, + pub delegation_id: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct BindingPolicyReference { + pub slot: String, + pub binding_id: String, + pub policy_digest: String, +} + +#[derive( + Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, +)] +#[serde(rename_all = "snake_case")] +pub enum PluginSurface { + Tool, +} + +/// Only capabilities implemented by the initial Tool Host are representable. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum PluginGrant { + Logging, + Https { slot: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct PluginSchemaIdentity { + pub identity: String, + pub version: String, + pub digest: String, + pub namespace: String, +} + +/// The upstream config/schema authority must validate config values as non-secret +/// and intersect grants/surfaces with Workspace, Profile and delegation bounds. +/// Binding policies remain with their owner: only identities and digests travel. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct PluginPinProjection { + pub config: Value, + pub surfaces: Vec, + pub tools: Vec, + pub grants: Vec, + pub bindings: Vec, +} + +/// All fields are private: construct from admitted artifact metadata, not a +/// hand-built wire snapshot. Deserialization alone does not establish admission. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct WorkerPluginPin { + installation_id: String, + expected_last_operation: String, + package_id: String, + package_version: String, + package_digest: String, + artifact_digest: String, + artifact_size: u32, + package_schema_version: u32, + component_world: String, + required_imports: Vec, + config_schema: PluginSchemaIdentity, + config_digest: String, + projection: PluginPinProjection, +} + +impl WorkerPluginPin { + pub fn from_artifact( + installation_id: String, + expected_last_operation: String, + artifact: &(impl ArtifactDescriptor + ?Sized), + mut projection: PluginPinProjection, + ) -> Result { + let artifact = artifact.artifact_identity(); + if artifact.revoked { + return conflict(); + } + if !valid_identity(&installation_id) + || !valid_identity(&expected_last_operation) + || !valid_digest(&artifact.package_digest) + || !valid_digest(&artifact.artifact_digest) + || artifact.artifact_size == 0 + || artifact.schema_version != plugin::SUPPORTED_PLUGIN_API_VERSION + || !matches!( + artifact.world.as_str(), + plugin::PLUGIN_COMPONENT_TOOL_WORLD + | plugin::PLUGIN_COMPONENT_HTTPS_TOOL_WORLD + | plugin::PLUGIN_COMPONENT_LOGGED_TOOL_WORLD + | plugin::PLUGIN_COMPONENT_HTTPS_LOGGED_TOOL_WORLD + ) + || !valid_digest(&artifact.config_schema.digest) + || !valid_identity(&artifact.package_id) + || !valid_identity(&artifact.version) + || !valid_identity(&artifact.config_schema.identity) + || !valid_identity(&artifact.config_schema.version) + || !valid_identity(&artifact.config_schema.namespace) + || !projection.config.is_object() + { + return invalid(); + } + projection.surfaces.sort(); + projection.surfaces.dedup(); + projection.tools.sort(); + projection.tools.dedup(); + projection.grants.sort(); + projection.grants.dedup(); + projection.bindings.sort(); + projection.config = canonical_value(&projection.config); + if (!projection.tools.is_empty() && !projection.surfaces.contains(&PluginSurface::Tool)) + || projection.tools.iter().any(|name| { + !valid_identity(name) || !artifact.tools.iter().any(|tool| *tool == *name) + }) + || projection + .bindings + .windows(2) + .any(|b| b[0].slot == b[1].slot) + || projection.bindings.iter().any(|binding| { + !valid_identity(&binding.slot) + || !valid_identity(&binding.binding_id) + || !valid_digest(&binding.policy_digest) + || !artifact + .connections + .iter() + .any(|slot| *slot == binding.slot) + }) + || projection.grants.iter().any(|grant| match grant { + PluginGrant::Logging => !artifact.requested_grants.contains(&PluginGrant::Logging), + PluginGrant::Https { slot } => { + !valid_identity(slot) + || !artifact + .requested_grants + .contains(&PluginGrant::Https { slot: slot.clone() }) + || !projection + .bindings + .iter() + .any(|binding| binding.slot == *slot) + } + }) + || projection.bindings.iter().any(|binding| { + !projection.grants.contains(&PluginGrant::Https { + slot: binding.slot.clone(), + }) + }) + { + return invalid(); + } + let mut required_imports = artifact.required_imports.clone(); + required_imports.sort(); + required_imports.dedup(); + Ok(Self { + installation_id, + expected_last_operation, + package_id: artifact.package_id.clone(), + package_version: artifact.version.clone(), + package_digest: artifact.package_digest.clone(), + artifact_digest: artifact.artifact_digest.clone(), + artifact_size: artifact.artifact_size, + package_schema_version: artifact.schema_version, + component_world: artifact.world.clone(), + required_imports, + config_schema: PluginSchemaIdentity { + identity: artifact.config_schema.identity.clone(), + version: artifact.config_schema.version.clone(), + digest: artifact.config_schema.digest.clone(), + namespace: artifact.config_schema.namespace.clone(), + }, + config_digest: digest_json(&projection.config)?, + projection, + }) + } + + pub fn installation_id(&self) -> &str { + &self.installation_id + } + pub fn expected_last_operation(&self) -> &str { + &self.expected_last_operation + } + pub fn package_id(&self) -> &str { + &self.package_id + } + pub fn package_version(&self) -> &str { + &self.package_version + } + pub fn package_schema_version(&self) -> u32 { + self.package_schema_version + } + pub fn component_world(&self) -> &str { + &self.component_world + } + pub fn required_imports(&self) -> &[String] { + &self.required_imports + } + pub fn package_digest(&self) -> &str { + &self.package_digest + } + pub fn artifact_digest(&self) -> &str { + &self.artifact_digest + } + pub fn artifact_size(&self) -> u32 { + self.artifact_size + } + pub fn config_schema(&self) -> &PluginSchemaIdentity { + &self.config_schema + } + pub fn config_digest(&self) -> &str { + &self.config_digest + } + pub fn projection(&self) -> &PluginPinProjection { + &self.projection + } + + pub fn matches_artifact(&self, artifact: &(impl ArtifactDescriptor + ?Sized)) -> Result { + Ok(Self::from_artifact( + self.installation_id.clone(), + self.expected_last_operation.clone(), + artifact, + self.projection.clone(), + )? == *self) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct WorkerPluginPlanInput { + pub workspace_id: String, + pub config: ConfigProvenance, + pub profile: ProfileProvenance, + pub execution: HostExecutionIdentity, + pub plugins: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct WorkerPluginExecutionPlan { + format_version: u32, + input: WorkerPluginPlanInput, +} + +impl WorkerPluginExecutionPlan { + pub fn canonicalize(mut input: WorkerPluginPlanInput) -> Result { + if !valid_identity(&input.workspace_id) + || !valid_digest(&input.config.content_digest) + || !valid_identity(&input.config.toolchain_fingerprint) + || !valid_identity(&input.profile.identity) + || !valid_digest(&input.profile.projection_digest) + || !valid_identity(&input.execution.runtime_id) + || !valid_identity(&input.execution.worker_id) + || !valid_identity(&input.execution.principal_id) + || input + .execution + .delegation_id + .as_ref() + .is_some_and(|id| !valid_identity(id)) + || input.plugins.len() > 128 + { + return invalid(); + } + input + .plugins + .sort_by(|a, b| a.installation_id.cmp(&b.installation_id)); + // One selection per installation; two aliases of one package instead use + // distinct installation identities, as required by the catalog contract. + if input + .plugins + .windows(2) + .any(|pins| pins[0].installation_id == pins[1].installation_id) + { + return invalid(); + } + // Public Tool names use immutable installation identities, never aliases. + let mut names = BTreeSet::new(); + for pin in &input.plugins { + for tool in &pin.projection.tools { + let name = format!("{}__{}", pin.installation_id, tool); + if name.len() > 128 || !names.insert(name) { + return invalid(); + } + } + } + Ok(Self { + format_version: 1, + input, + }) + } + + pub fn input(&self) -> &WorkerPluginPlanInput { + &self.input + } + + pub fn workspace_id(&self) -> &str { + &self.input.workspace_id + } + pub fn config(&self) -> &ConfigProvenance { + &self.input.config + } + pub fn profile(&self) -> &ProfileProvenance { + &self.input.profile + } + pub fn execution(&self) -> &HostExecutionIdentity { + &self.input.execution + } + pub fn plugins(&self) -> &[WorkerPluginPin] { + &self.input.plugins + } + pub fn fingerprint(&self) -> Result { + digest_json(self) + } + pub fn canonical_json(&self) -> Result { + encode_canonical(self) + } + + pub fn validate_canonical(&self) -> Result<()> { + if self.format_version != 1 || Self::canonicalize(self.input.clone())? != *self { + return invalid(); + } + Ok(()) + } +} + +fn invalid() -> Result { + Err(PlanError::Invalid) +} +fn conflict() -> Result { + Err(PlanError::Conflict) +} +fn valid_identity(value: &str) -> bool { + !value.is_empty() && value.len() <= 512 && !value.chars().any(char::is_control) +} +fn valid_digest(value: &str) -> bool { + value.strip_prefix("sha256:").is_some_and(|hex| { + hex.len() == 64 + && hex + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + }) +} +fn canonical_value(value: &Value) -> Value { + match value { + Value::Object(fields) => { + let mut keys: Vec<_> = fields.keys().collect(); + keys.sort(); + Value::Object( + keys.into_iter() + .map(|key| (key.clone(), canonical_value(&fields[key]))) + .collect(), + ) + } + Value::Array(values) => Value::Array(values.iter().map(canonical_value).collect()), + _ => value.clone(), + } +} +fn encode_canonical(value: &impl Serialize) -> Result { + let value = serde_json::to_value(value).map_err(|_| PlanError::Invalid)?; + serde_json::to_string(&canonical_value(&value)).map_err(|_| PlanError::Invalid) +} +fn digest_json(value: &impl Serialize) -> Result { + let bytes = encode_canonical(value)?; + Ok(hash_bytes(bytes.as_bytes())) +} +fn hash_bytes(bytes: &[u8]) -> String { + let hex: String = Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + format!("sha256:{hex}") +} + +#[cfg(test)] +mod tests; diff --git a/crates/plugin-plan/src/tests.rs b/crates/plugin-plan/src/tests.rs new file mode 100644 index 000000000..a093157db --- /dev/null +++ b/crates/plugin-plan/src/tests.rs @@ -0,0 +1,204 @@ +use super::*; +use serde_json::json; + +fn digest(hex: char) -> String { + format!("sha256:{}", hex.to_string().repeat(64)) +} + +fn artifact() -> ArtifactIdentityProjection { + ArtifactIdentityProjection { + package_id: "package".into(), + version: "1.0.0".into(), + package_digest: digest('a'), + artifact_digest: digest('b'), + artifact_size: 42, + schema_version: 2, + world: plugin::PLUGIN_COMPONENT_HTTPS_LOGGED_TOOL_WORLD.into(), + required_imports: vec!["logging".into(), "https".into(), "logging".into()], + tools: vec!["other".into(), "echo".into()], + connections: vec!["api".into()], + requested_grants: vec![ + PluginGrant::Https { slot: "api".into() }, + PluginGrant::Logging, + ], + config_schema: PluginSchemaIdentity { + identity: "config".into(), + version: "1".into(), + digest: digest('c'), + namespace: "example".into(), + }, + revoked: false, + } +} + +fn projection() -> PluginPinProjection { + PluginPinProjection { + config: serde_json::from_str(r#"{"z":{"b":2,"a":1},"a":[{"z":0,"a":false}]}"#).unwrap(), + surfaces: vec![PluginSurface::Tool, PluginSurface::Tool], + tools: vec!["other".into(), "echo".into(), "other".into()], + grants: vec![ + PluginGrant::Https { slot: "api".into() }, + PluginGrant::Logging, + ], + bindings: vec![BindingPolicyReference { + slot: "api".into(), + binding_id: "binding".into(), + policy_digest: digest('f'), + }], + } +} + +fn pin( + artifact: &impl ArtifactDescriptor, + projection: PluginPinProjection, +) -> Result { + WorkerPluginPin::from_artifact( + "installation".into(), + "operation".into(), + artifact, + projection, + ) +} + +fn input() -> WorkerPluginPlanInput { + WorkerPluginPlanInput { + workspace_id: "workspace".into(), + config: ConfigProvenance { + content_digest: digest('d'), + toolchain_fingerprint: "toolchain".into(), + }, + profile: ProfileProvenance { + identity: "profile:coder".into(), + projection_digest: digest('e'), + }, + execution: HostExecutionIdentity { + runtime_id: "runtime".into(), + worker_id: "worker".into(), + principal_id: "principal".into(), + delegation_id: Some("delegation".into()), + }, + plugins: vec![pin(&artifact(), projection()).unwrap()], + } +} + +#[test] +fn canonical_wire_and_fingerprint_preserve_persisted_format() { + let plan = WorkerPluginExecutionPlan::canonicalize(input()).unwrap(); + let expected = include_str!("fixtures/canonical-plan.json").trim_end(); + assert_eq!(plan.canonical_json().unwrap(), expected); + assert_eq!( + plan.fingerprint().unwrap(), + "sha256:00ee15d79098f4afa137da5d4b0b9b78acde48a6157fdda0f3327d8c122bd889" + ); + let restored: WorkerPluginExecutionPlan = serde_json::from_str(expected).unwrap(); + restored.validate_canonical().unwrap(); + assert_eq!(restored, plan); + assert!(restored.plugins()[0].matches_artifact(&artifact()).unwrap()); +} + +#[test] +fn artifact_identity_changes_are_not_hidden_by_projection_adapter() { + let original = pin(&artifact(), projection()).unwrap(); + let mut changed = artifact(); + changed.artifact_size += 1; + assert!(!original.matches_artifact(&changed).unwrap()); + changed = artifact(); + changed.config_schema.digest = digest('d'); + assert!(!original.matches_artifact(&changed).unwrap()); + changed = artifact(); + changed.revoked = true; + assert_eq!( + original.matches_artifact(&changed), + Err(PlanError::Conflict) + ); +} + +#[test] +fn unsupported_or_malformed_artifact_identity_is_invalid() { + for change in [ + |a: &mut ArtifactIdentityProjection| a.package_digest = "sha256:ABC".into(), + |a: &mut ArtifactIdentityProjection| a.artifact_digest = "digest".into(), + |a: &mut ArtifactIdentityProjection| a.artifact_size = 0, + |a: &mut ArtifactIdentityProjection| a.schema_version = 1, + |a: &mut ArtifactIdentityProjection| a.world = "unsupported".into(), + |a: &mut ArtifactIdentityProjection| a.package_id.clear(), + |a: &mut ArtifactIdentityProjection| a.version = "bad\nversion".into(), + |a: &mut ArtifactIdentityProjection| a.config_schema.identity.clear(), + |a: &mut ArtifactIdentityProjection| a.config_schema.version.clear(), + |a: &mut ArtifactIdentityProjection| a.config_schema.digest.clear(), + |a: &mut ArtifactIdentityProjection| a.config_schema.namespace.clear(), + ] { + let mut value = artifact(); + change(&mut value); + assert_eq!(pin(&value, projection()), Err(PlanError::Invalid)); + } +} + +#[test] +fn selected_tools_grants_and_bindings_cannot_expand_artifact_metadata() { + for change in [ + |p: &mut PluginPinProjection| p.config = json!([]), + |p: &mut PluginPinProjection| p.surfaces.clear(), + |p: &mut PluginPinProjection| p.tools.push("absent".into()), + |p: &mut PluginPinProjection| p.bindings.push(p.bindings[0].clone()), + |p: &mut PluginPinProjection| p.bindings[0].slot = "absent".into(), + |p: &mut PluginPinProjection| p.bindings[0].binding_id.clear(), + |p: &mut PluginPinProjection| p.bindings[0].policy_digest.clear(), + |p: &mut PluginPinProjection| p.bindings.clear(), + |p: &mut PluginPinProjection| p.grants.retain(|g| *g == PluginGrant::Logging), + ] { + let mut value = projection(); + change(&mut value); + assert_eq!(pin(&artifact(), value), Err(PlanError::Invalid)); + } + for change in [ + |a: &mut ArtifactIdentityProjection| a.tools.clear(), + |a: &mut ArtifactIdentityProjection| a.connections.clear(), + |a: &mut ArtifactIdentityProjection| a.requested_grants.clear(), + ] { + let mut value = artifact(); + change(&mut value); + assert_eq!(pin(&value, projection()), Err(PlanError::Invalid)); + } +} + +#[test] +fn malformed_provenance_and_duplicate_installations_are_invalid() { + for change in [ + |i: &mut WorkerPluginPlanInput| i.workspace_id.clear(), + |i: &mut WorkerPluginPlanInput| i.config.content_digest.clear(), + |i: &mut WorkerPluginPlanInput| i.config.toolchain_fingerprint.clear(), + |i: &mut WorkerPluginPlanInput| i.profile.identity.clear(), + |i: &mut WorkerPluginPlanInput| i.profile.projection_digest.clear(), + |i: &mut WorkerPluginPlanInput| i.execution.runtime_id.clear(), + |i: &mut WorkerPluginPlanInput| i.execution.worker_id.clear(), + |i: &mut WorkerPluginPlanInput| i.execution.principal_id.clear(), + |i: &mut WorkerPluginPlanInput| i.execution.delegation_id = Some(String::new()), + |i: &mut WorkerPluginPlanInput| i.plugins.push(i.plugins[0].clone()), + |i: &mut WorkerPluginPlanInput| i.plugins = vec![i.plugins[0].clone(); 129], + ] { + let mut value = input(); + change(&mut value); + assert_eq!( + WorkerPluginExecutionPlan::canonicalize(value), + Err(PlanError::Invalid) + ); + } +} + +#[test] +fn deserialized_unknown_fields_versions_and_noncanonical_order_are_rejected() { + let plan = WorkerPluginExecutionPlan::canonicalize(input()).unwrap(); + let mut wire = serde_json::to_value(&plan).unwrap(); + wire["input"]["execution"]["token"] = json!("credential"); + assert!(serde_json::from_value::(wire).is_err()); + let mut wire = serde_json::to_value(&plan).unwrap(); + wire["format_version"] = json!(2); + let unsupported: WorkerPluginExecutionPlan = serde_json::from_value(wire).unwrap(); + assert_eq!(unsupported.validate_canonical(), Err(PlanError::Invalid)); + let mut unsorted = plan.clone(); + let mut second = unsorted.input.plugins[0].clone(); + second.installation_id = "aaa".into(); + unsorted.input.plugins.push(second); + assert_eq!(unsorted.validate_canonical(), Err(PlanError::Invalid)); +} diff --git a/crates/plugin-runtime/src/executor.rs b/crates/plugin-runtime/src/executor.rs index 9023e4d41..623afefbc 100644 --- a/crates/plugin-runtime/src/executor.rs +++ b/crates/plugin-runtime/src/executor.rs @@ -91,6 +91,11 @@ impl CompiledTool { pub fn identity(&self) -> &ArtifactIdentity { &self.identity } + + /// Declarations from the verified immutable package, not execution grants. + pub fn tools(&self) -> &[PluginToolManifest] { + &self.manifest.tools + } } #[derive(Debug)] diff --git a/crates/runtime-api/Cargo.toml b/crates/runtime-api/Cargo.toml index cc544d3e9..bf44b1d74 100644 --- a/crates/runtime-api/Cargo.toml +++ b/crates/runtime-api/Cargo.toml @@ -8,6 +8,7 @@ license.workspace = true [dependencies] api-macros = { workspace = true, features = ["reqwest", "axum"] } manifest.workspace = true +plugin-plan = { path = "../plugin-plan" } protocol.workspace = true session-store.workspace = true serde = { workspace = true, features = ["derive"] } diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 12c56a6b1..e82df412e 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -431,6 +431,17 @@ pub struct BackendJobExecutionBinding { pub subjektiv_journal: bool, } +/// Server-authored immutable launch input, retained unchanged across restoration. +/// A snapshot fixes selection and ceilings; each preparation and invocation must +/// still consult the current owner. No credential or cache location is serialized. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct PluginExecutionSnapshot { + pub plan_id: String, + pub fingerprint: String, + pub plan: plugin_plan::WorkerPluginExecutionPlan, +} + #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct CreateWorkerRequest { pub worker_id: WorkerId, @@ -458,6 +469,8 @@ pub struct CreateWorkerRequest { pub workspace_api: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub memory_settings: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub plugin_execution: Option, /// Backend-attested subject or consolidation attachment. Profile policy alone /// never activates subjektiv for an ordinary Workspace Worker. #[serde(default, skip_serializing_if = "std::ops::Not::not")] diff --git a/crates/server-api/Cargo.toml b/crates/server-api/Cargo.toml index dfc34d4a3..2bf3b75e0 100644 --- a/crates/server-api/Cargo.toml +++ b/crates/server-api/Cargo.toml @@ -12,6 +12,7 @@ typescript = ["dep:ts-rs", "protocol/typescript"] [dependencies] api-macros = { workspace = true, features = ["reqwest", "axum", "openapi"] } memory.workspace = true +plugin-plan.workspace = true merge-request.workspace = true protocol = { workspace = true, features = ["json-schema"] } runtime-api.workspace = true diff --git a/crates/server-api/src/plugin.rs b/crates/server-api/src/plugin.rs index 418021947..9d9693c55 100644 --- a/crates/server-api/src/plugin.rs +++ b/crates/server-api/src/plugin.rs @@ -122,6 +122,44 @@ pub struct PluginArtifact { pub revocation_reason: Option, } +impl plugin_plan::ArtifactDescriptor for PluginArtifact { + fn artifact_identity(&self) -> plugin_plan::ArtifactIdentityProjection { + plugin_plan::ArtifactIdentityProjection { + package_id: self.package_id.clone(), + version: self.version.clone(), + package_digest: self.package_digest.clone(), + artifact_digest: self.artifact_digest.clone(), + artifact_size: self.artifact_size, + schema_version: self.schema_version, + world: self.world.clone(), + required_imports: self.required_imports.clone(), + tools: self.tools.iter().map(|tool| tool.name.clone()).collect(), + connections: self + .connections + .iter() + .map(|slot| slot.name.clone()) + .collect(), + requested_grants: self + .requested_grants + .iter() + .map(|grant| match grant { + PluginGrantMetadata::Logging => plugin_plan::PluginGrant::Logging, + PluginGrantMetadata::Https { slot } => { + plugin_plan::PluginGrant::Https { slot: slot.clone() } + } + }) + .collect(), + config_schema: plugin_plan::PluginSchemaIdentity { + identity: self.config_schema.identity.clone(), + version: self.config_schema.version.clone(), + digest: self.config_schema.digest.clone(), + namespace: self.config_schema.namespace.clone(), + }, + revoked: self.revoked, + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(deny_unknown_fields)] diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index 89acce8bc..d6fe83605 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -37,6 +37,7 @@ flow = { path = "../flow" } fs4 = { workspace = true, optional = true } manifest.workspace = true yoi-plugin-runtime = { path = "../plugin-runtime" } +plugin-plan = { path = "../plugin-plan" } libc.workspace = true protocol.workspace = true runtime-api.workspace = true diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index f6bb365bb..913a918c4 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -298,6 +298,8 @@ pub struct CreateWorkerRequest { /// Backend-authored immutable Workspace Memory settings snapshot. #[serde(default, skip_serializing_if = "Option::is_none")] pub memory_settings: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub plugin_execution: Option, /// Backend-attested subject or consolidation attachment. A reusable Profile /// may enable subjektiv policy, but only this host-owned bit activates it. #[serde(default, skip_serializing_if = "is_false")] diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 06f551100..0548f66f1 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -3057,6 +3057,7 @@ mod tests { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle(profile.clone()); CreateWorkerRequest { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "test-create".to_string(), profile, @@ -5024,6 +5025,7 @@ mod ws_tests { fn ws_create_request() -> CreateWorkerRequest { let bundle = ws_test_bundle(ProfileSelector::Builtin("builtin:companion".to_string())); CreateWorkerRequest { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index c00c5fe77..637d56b6b 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -14,6 +14,7 @@ pub mod identity; pub mod interaction; pub mod management; pub mod observation; +pub mod plugin; pub mod profile_archive; pub mod resource; #[cfg(feature = "restore-checkpoints")] diff --git a/crates/worker-runtime/src/plugin.rs b/crates/worker-runtime/src/plugin.rs new file mode 100644 index 000000000..01d33db48 --- /dev/null +++ b/crates/worker-runtime/src/plugin.rs @@ -0,0 +1,87 @@ +//! Process-owned bridge for exact Plugin launch inputs. Implementations compose +//! the existing delivery/cache/executor and live owner protocols; they may not +//! reselect a Profile, search an artifact inventory, or mint execution authority. +use crate::identity::WorkerRef; +use async_trait::async_trait; + +#[async_trait] +pub trait PluginToolRuntime: Send + Sync { + /// Prepare unpublished contributions for this exact persisted plan. Both + /// create and restore call this boundary before Feature/controller publication. + /// Success requires current owner startup admission, not only verified bytes. + async fn prepare( + &self, + worker: &WorkerRef, + snapshot: &runtime_api::PluginExecutionSnapshot, + session_id: &str, + ) -> Result; +} + +/// Materialize only the exact Server-authored pins. This prepares contributions, +/// never publishes them or claims startup admission: the Host bridge must perform +/// its live owner preflight before returning from `PluginToolRuntime::prepare`. +/// Metadata from delivery becomes an expectation only after full immutable-pin +/// comparison; byte equality never issues an execution grant. +pub async fn materialize( + worker: &WorkerRef, + snapshot: &runtime_api::PluginExecutionSnapshot, + session_id: &str, + source: &dyn yoi_plugin_runtime::delivery::ArtifactSource, + owner: std::sync::Arc, + cache: &yoi_plugin_runtime::delivery::ArtifactCache, + limits: yoi_plugin_runtime::Limits, +) -> Result { + use yoi_plugin_runtime::{delivery::ArtifactExpectation, host::Grants}; + let plan = &snapshot.plan; + if plan.validate_canonical().is_err() + || plan.fingerprint().ok().as_ref() != Some(&snapshot.fingerprint) + || plan.execution().worker_id != worker.worker_id.to_string() + || snapshot.plan_id.is_empty() + { + return Err("Plugin launch snapshot mismatch".into()); + } + let mut pins = Vec::new(); + for pin in plan.plugins() { + let request = server_api::PluginArtifactDeliveryRequest { + plan_id: snapshot.plan_id.clone(), + plan_fingerprint: snapshot.fingerprint.clone(), + installation_id: pin.installation_id().into(), + artifact_digest: pin.artifact_digest().into(), + }; + let metadata = source + .authorize(&request) + .await + .map_err(|_| "Plugin artifact admission unavailable")?; + if metadata.workspace_id != plan.workspace_id() + || metadata.request != request + || !pin + .matches_artifact(&metadata.artifact) + .map_err(|_| "Plugin artifact metadata mismatch")? + { + return Err("Plugin artifact metadata mismatch".into()); + } + let mut grants = Grants::default(); + for grant in &pin.projection().grants { + match grant { + plugin_plan::PluginGrant::Logging => grants.logging = true, + plugin_plan::PluginGrant::Https { slot } => { + grants.connection_slots.insert(slot.clone()); + } + } + } + pins.push(worker::plugin::PluginToolPin { + workspace_id: plan.workspace_id().into(), + installation_id: pin.installation_id().into(), + plan_id: snapshot.plan_id.clone(), + subject: plan.execution().principal_id.clone(), + request, + expectation: ArtifactExpectation::new(metadata.artifact), + tools: pin.projection().tools.clone(), + config: pin.projection().config.clone(), + grants, + }); + } + worker::plugin::PluginToolsFeature::prepare(pins, source, cache, owner, session_id, limits) + .await + .map_err(|error| error.to_string()) +} diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 71583069d..a6cdd56df 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1242,6 +1242,7 @@ impl Runtime { })?; if existing_request.create_fingerprint != request.create_fingerprint || existing_request.backend_job != request.backend_job + || existing_request.plugin_execution != request.plugin_execution { return Err(RuntimeError::InvalidRequest(format!( "worker {} was already created with a different fingerprint", @@ -1427,6 +1428,7 @@ impl Runtime { })?; if existing_request.create_fingerprint != request.create_fingerprint || existing_request.backend_job != request.backend_job + || existing_request.plugin_execution != request.plugin_execution { return Err(RuntimeError::InvalidRequest(format!( "worker {} was already created with a different fingerprint", @@ -7428,6 +7430,21 @@ fn validate_logical_workdir_attachments( } fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> { + if let Some(snapshot) = &request.plugin_execution { + let valid = snapshot.plan.validate_canonical().is_ok() + && snapshot.plan.fingerprint().ok().as_ref() == Some(&snapshot.fingerprint) + && !snapshot.plan_id.is_empty() + && snapshot.plan.execution().worker_id == request.worker_id.to_string() + && request + .workspace_api + .as_ref() + .is_some_and(|api| api.workspace_id == snapshot.plan.workspace_id()); + if !valid { + return Err(RuntimeError::InvalidRequest( + "Plugin execution snapshot does not match this Worker launch".into(), + )); + } + } if let Some(binding) = &request.backend_job { binding .validate() @@ -8239,6 +8256,7 @@ mod tests { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle_for_profile(profile.clone()); CreateWorkerRequest { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "test-create".to_string(), profile, diff --git a/crates/worker-runtime/src/test_support_tests.rs b/crates/worker-runtime/src/test_support_tests.rs index 562ec2e8e..9a458d28e 100644 --- a/crates/worker-runtime/src/test_support_tests.rs +++ b/crates/worker-runtime/src/test_support_tests.rs @@ -67,6 +67,7 @@ fn resolved() -> manifest::WorkerManifest { fn request() -> CreateWorkerRequest { CreateWorkerRequest { + plugin_execution: None, worker_id: worker_id(), create_fingerprint: "new-create-fingerprint".into(), profile: ProfileSelector::Builtin("default".into()), diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index df85e3602..53a173433 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -825,6 +825,7 @@ type SpawnModelClientFactory = #[derive(Clone)] pub struct ProfileRuntimeWorkerFactory { spawn_model_client_factory: Option, + plugin_runtime: Option>, observation_hub: Arc, profile_base_dir: PathBuf, worker_aggregate_root: Option, @@ -949,6 +950,7 @@ impl ProfileRuntimeWorkerFactory { let profile_base_dir = profile_base_dir.into(); Self { spawn_model_client_factory: None, + plugin_runtime: None, observation_hub: Arc::new(RuntimeWorkerObservationHub::default()), profile_base_dir, worker_aggregate_root: None, @@ -991,6 +993,16 @@ impl ProfileRuntimeWorkerFactory { self } + /// Bind process-owned Plugin composition. Profile selection without this + /// service fails closed; no ambient or offline fallback is available. + pub fn with_plugin_runtime( + mut self, + runtime: Arc, + ) -> Self { + self.plugin_runtime = Some(runtime); + self + } + #[cfg(feature = "http-server")] pub fn with_workspace_request_client(mut self, client: RuntimeWorkspaceRequestClient) -> Self { self.runtime_id = Some(client.runtime_id().to_string()); @@ -2125,6 +2137,23 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { } })?; let worker = prepared.worker_mut(); + if let Some(snapshot) = &request.request.plugin_execution { + if self.runtime_id.as_deref() != Some(snapshot.plan.execution().runtime_id.as_str()) { + return Err("Plugin plan Runtime identity mismatch".into()); + } + let runtime = self + .plugin_runtime + .as_ref() + .ok_or("Plugin runtime provider unavailable")?; + let tools = runtime + .prepare( + &request.worker_ref, + snapshot, + &worker.session_id().to_string(), + ) + .await?; + worker.bind_plugin_tools(tools); + } if let Some(binding) = request.request.backend_job.as_ref() { worker .bind_backend_job(binding.clone()) @@ -2933,6 +2962,27 @@ impl ProfileRuntimeWorkerFactory { Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} Err(_) => return Err("restore run artifacts cannot be inspected"), } + if let Some(snapshot) = &request.request.plugin_execution { + if self.runtime_id.as_deref() != Some(snapshot.plan.execution().runtime_id.as_str()) + { + return Err("Plugin plan Runtime identity mismatch"); + } + let runtime = self + .plugin_runtime + .as_ref() + .ok_or("Plugin runtime provider unavailable")?; + let tools = runtime + .prepare( + &request.worker_ref, + snapshot, + &candidate.session_id().to_string(), + ) + .await + .map_err(|_| "Plugin restore preparation refused or unavailable")?; + candidate + .bind_plugin_tools(tools) + .map_err(|_| "restored Plugin binding failed")?; + } let prepared = candidate .prepare_runtime_controller( &run_dir, @@ -10430,6 +10480,7 @@ mod tests { fn create_request(_name: &str) -> CreateWorkerRequest { let bundle = test_bundle(); CreateWorkerRequest { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index f8b4f26f6..d90d069c6 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -15,6 +15,7 @@ agen = { workspace = true, features = ["codex"] } session-store = { workspace = true } secrets = { workspace = true } manifest = { workspace = true } +yoi-plugin-runtime = { path = "../plugin-runtime" } job = { workspace = true } mcp = { workspace = true } protocol = { workspace = true, features = ["json-schema"] } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index dfc9fd7cb..9b77466d1 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1544,6 +1544,18 @@ where } else { (FeatureRegistryBuilder::new(), None) }; + match worker.plugin_tools() { + Some(tools) => { + feature_registry.add_module(tools); + } + None if !worker.manifest().plugins.is_empty() => { + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "Plugin selection requires a persisted plan and current Host preparation", + )); + } + None => {} + } feature_registry .add_module(crate::feature::builtin::chat_invocation::AttachmentInvocationFeature); if let Some(binding) = backend_job.as_ref() { diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index d9492b077..e8052500b 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -17,6 +17,7 @@ pub mod job; mod metadata_lifecycle; pub mod model_client; mod paste_artifact_tool; +pub mod plugin; pub mod profile_manifest; pub mod prompt; #[cfg(test)] diff --git a/crates/worker/src/plugin.rs b/crates/worker/src/plugin.rs new file mode 100644 index 000000000..5924043e7 --- /dev/null +++ b/crates/worker/src/plugin.rs @@ -0,0 +1,389 @@ +//! Host-selected Plugin Tools use the ordinary Feature install transaction and +//! Agen execution/history path. This module neither selects a Plugin nor grants +//! access: callers bind exact persisted pins and current owner adapters before +//! controller publication. Verified bytes and compiled Components are not leases. + +use crate::feature::{ + FeatureDescriptor, FeatureHookPoint, FeatureId, FeatureInstallContext, FeatureInstallError, + FeatureModule, FeatureRuntimeKind, HookDeclaration, ToolContribution, ToolDeclaration, +}; +use crate::hook::{ + Hook, HookError, HookErrorCategory, HookExecutionPolicy, WorkerStopping, WorkerStoppingContext, +}; +use agen::tool::{ + Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOrigin, ToolOutput, +}; +use async_trait::async_trait; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::{ + collections::{BTreeMap, BTreeSet}, + sync::{Arc, Mutex}, +}; +use tokio::sync::Semaphore; +use yoi_plugin_runtime::{ + CompiledTool, Executor, Limits, + delivery::{ArtifactCache, ArtifactExpectation, ArtifactSource}, + host::{CancellationToken, Grants, Host, InvocationContext, Operation}, + owner::{OwnerHost, PluginHostOwner}, +}; + +/// Upper bounds projected by the trusted launch owner from one persisted pin. +/// Never construct this from model input, current Profile, cache inventory, or +/// unvalidated delivery metadata. The expectation must match the persisted pin. +pub struct PluginToolPin { + pub workspace_id: String, + pub installation_id: String, + pub plan_id: String, + pub subject: String, + pub request: server_api::PluginArtifactDeliveryRequest, + pub expectation: ArtifactExpectation, + pub tools: Vec, + pub config: Value, + pub grants: Grants, +} + +#[derive(Debug, thiserror::Error)] +pub enum PluginPreparationError { + #[error("invalid Plugin launch snapshot")] + InvalidSnapshot, + #[error("Plugin artifact delivery refused or unavailable")] + Delivery, + #[error("Plugin Component or execution policy is unsupported")] + Component, +} + +struct ActiveCall { + cancellation: CancellationToken, + host: Mutex>>, +} +impl ActiveCall { + async fn cancel(&self) -> Result<(), ToolError> { + self.cancellation.cancel(); + let host = self.host.lock().expect("Plugin call lock").clone(); + if let Some(host) = host { + host.cancel() + .await + .map_err(|_| failure("Plugin cancellation outcome requires reconciliation"))?; + } + Ok(()) + } +} +#[derive(Default)] +struct Calls { + stopped: bool, + active: BTreeMap>, +} +struct CallLease { + key: String, + state: Arc>, + call: Arc, +} +impl Drop for CallLease { + fn drop(&mut self) { + self.call.cancellation.cancel(); + self.state + .lock() + .expect("Plugin calls lock") + .active + .remove(&self.key); + } +} + +/// Fully materialized Tool contributions. Preparation publishes nothing; the +/// existing Feature planner stages all names, schemas and hooks atomically. +#[derive(Clone)] +pub struct PluginToolsFeature { + tools: Vec<(ToolMeta, Arc)>, + calls: Arc>, +} +impl PluginToolsFeature { + pub async fn prepare( + pins: Vec, + source: &dyn ArtifactSource, + cache: &ArtifactCache, + owner: Arc, + session_id: &str, + limits: Limits, + ) -> Result { + use PluginPreparationError::*; + if session_id.is_empty() || pins.len() > 128 { + return Err(InvalidSnapshot); + } + let executor = Arc::new(Executor::new(limits).map_err(|_| Component)?); + let calls = Arc::new(Mutex::new(Calls::default())); + let capacity = Arc::new(Semaphore::new(16)); + let mut names = BTreeSet::new(); + let mut installations = BTreeSet::new(); + let mut tools = Vec::new(); + for pin in pins { + let descriptor = pin.expectation.descriptor(); + if pin.workspace_id.is_empty() + || pin.installation_id.is_empty() + || pin.plan_id.is_empty() + || pin.plan_id != pin.request.plan_id + || pin.installation_id != pin.request.installation_id + || descriptor.artifact_digest != pin.request.artifact_digest + || !pin.config.is_object() + || !installations.insert((pin.workspace_id.clone(), pin.installation_id.clone())) + || pin.tools.iter().collect::>().len() != pin.tools.len() + { + return Err(InvalidSnapshot); + } + let config_json = serde_json::to_string(&pin.config).map_err(|_| InvalidSnapshot)?; + if config_json.len() > executor.limits().config_bytes { + return Err(InvalidSnapshot); + } + let artifact = cache + .resolve(source, &pin.workspace_id, &pin.request, &pin.expectation) + .await + .map_err(|_| Delivery)?; + let compiled = Arc::new( + executor + .compile( + artifact.bytes(), + &descriptor.package_digest, + &descriptor.artifact_digest, + u64::from(descriptor.artifact_size), + ) + .map_err(|_| Component)?, + ); + for name in &pin.tools { + let declaration = compiled + .tools() + .iter() + .find(|tool| &tool.name == name) + .ok_or(InvalidSnapshot)?; + let public_name = format!("{}__{}", pin.installation_id, name); + if public_name.len() > 128 || !names.insert(public_name.clone()) { + return Err(InvalidSnapshot); + } + let meta = ToolMeta::new(public_name) + .description(declaration.description.clone()) + .input_schema(declaration.input_schema.clone()) + .origin(ToolOrigin { + kind: "plugin".into(), + plugin_id: descriptor.package_id.clone(), + plugin_ref: pin.installation_id.clone(), + source: "workspace".into(), + digest: descriptor.package_digest.clone(), + package_version: descriptor.version.clone(), + package_api_version: descriptor.schema_version, + surface: "tool".into(), + }); + let tool = Arc::new(PluginTool { + executor: executor.clone(), + compiled: compiled.clone(), + owner: owner.clone(), + workspace: pin.workspace_id.clone(), + installation: pin.installation_id.clone(), + plan: pin.plan_id.clone(), + subject: pin.subject.clone(), + grants: pin.grants.clone(), + tool_name: name.clone(), + config_json: config_json.clone(), + session_id: session_id.into(), + calls: calls.clone(), + capacity: capacity.clone(), + }); + tools.push((meta, tool)); + } + } + Ok(Self { tools, calls }) + } +} +impl FeatureModule for PluginToolsFeature { + fn descriptor(&self) -> FeatureDescriptor { + let mut descriptor = + FeatureDescriptor::builtin("plugin-tools", "Workspace Plugin Tools").with_hook( + HookDeclaration::new("stop", FeatureHookPoint::WorkerStopping), + ); + descriptor.id = FeatureId::new("plugin:tools").expect("fixed Feature identity"); + descriptor.runtime = FeatureRuntimeKind::ExternalPlugin; + for (meta, _) in &self.tools { + descriptor + .tools + .push(ToolDeclaration::new(&meta.name, &meta.description)); + } + descriptor + } + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + for (meta, tool) in &self.tools { + let meta = meta.clone(); + let tool = tool.clone(); + let name = meta.name.clone(); + let definition: ToolDefinition = Arc::new(move || (meta.clone(), tool.clone())); + context + .tools() + .register(ToolContribution::new(name, definition))?; + } + context.hooks().add_worker_stopping( + "stop", + HookExecutionPolicy::fail_closed(), + self.clone(), + ) + } +} +#[async_trait] +impl Hook for PluginToolsFeature { + async fn call(&self, _: &WorkerStoppingContext) -> Result<(), HookError> { + let active = { + let mut calls = self.calls.lock().expect("Plugin calls lock"); + calls.stopped = true; + calls.active.values().cloned().collect::>() + }; + let mut failed = false; + for call in active { + if call.cancel().await.is_err() { + failed = true; + } + } + if failed { + return Err(HookError::new( + HookErrorCategory::Dependency, + "Plugin cancellation requires reconciliation", + )); + } + Ok(()) + } +} +struct PluginTool { + executor: Arc, + compiled: Arc, + owner: Arc, + workspace: String, + installation: String, + plan: String, + subject: String, + grants: Grants, + tool_name: String, + config_json: String, + session_id: String, + calls: Arc>, + capacity: Arc, +} +fn failure(message: &str) -> ToolError { + ToolError::Internal(message.into()) +} +impl PluginTool { + fn start(&self, context: &ToolExecutionContext) -> Result { + let key = context.execution_id(); + let mut calls = self.calls.lock().expect("Plugin calls lock"); + if calls.stopped || calls.active.contains_key(&key) { + return Err(failure("Plugin Tool is stopping or already running")); + } + let call = Arc::new(ActiveCall { + cancellation: CancellationToken::new(), + host: Mutex::new(None), + }); + calls.active.insert(key.clone(), call.clone()); + Ok(CallLease { + key, + state: self.calls.clone(), + call, + }) + } +} +#[async_trait] +impl Tool for PluginTool { + async fn execute( + &self, + input_json: &str, + context: ToolExecutionContext, + ) -> Result { + if context.call_id.is_empty() || context.call_id == "direct" { + return Err(failure("Plugin Tool requires committed Tool-call identity")); + } + let _capacity = self + .capacity + .clone() + .try_acquire_owned() + .map_err(|_| failure("Plugin invocation capacity exceeded"))?; + let lease = self.start(&context)?; + // The persisted Session and provider call ID survive Runtime restoration; + // the engine-local batch ID identifies only cancellation of a live attempt. + let id_bytes = serde_json::to_vec(&(&self.session_id, &context.call_id)) + .expect("identity serialization"); + let hex: String = Sha256::digest(&id_bytes) + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + let call_id = format!("tool-{hex}"); + let invocation = InvocationContext { + workspace: self.workspace.clone(), + installation: self.installation.clone(), + plan: self.plan.clone(), + subject: self.subject.clone(), + invocation_id: String::new(), + artifact: self.compiled.identity().clone(), + tool_name: self.tool_name.clone(), + grants: self.grants.clone(), + deadline: tokio::time::Instant::now() + self.executor.limits().wall_time, + cancellation: lease.call.cancellation.clone(), + }; + let host = Arc::new( + OwnerHost::begin( + self.owner.clone(), + invocation, + server_api::PluginConsumer::ToolCall { + tool_call_id: call_id, + tool_name: self.tool_name.clone(), + }, + ) + .await + .map_err(|_| failure("Plugin live admission refused or unavailable"))?, + ); + *lease.call.host.lock().expect("Plugin call lock") = Some(host.clone()); + if lease.call.cancellation.is_cancelled() { + host.cancel() + .await + .map_err(|_| failure("Plugin cancellation requires reconciliation"))?; + return Err(failure("Plugin invocation cancelled")); + } + let output = self + .executor + .invoke( + &self.compiled, + host.context().clone(), + host.clone(), + &self.tool_name, + input_json, + &self.config_json, + ) + .await + .map_err(|_| failure("Plugin invocation failed; no automatic retry"))? + .output; + // Fence again at the adapter publication boundary, not just on guest return. + host.authorize(host.context(), Operation::Publish) + .await + .map_err(|_| failure("Plugin result publication refused"))?; + let content = match output.structured_json { + Some(structured) => Some(serde_json::json!({ + "content": output.content, + "structured": serde_json::from_str::(&structured).map_err(|_| failure("Plugin output is invalid"))?, + }).to_string()), + None => output.content, + }; + Ok(ToolOutput { + summary: output.summary, + content, + attachments: Vec::new(), + }) + } + async fn cancel_execution(&self, context: &ToolExecutionContext) -> Result<(), ToolError> { + let call = self + .calls + .lock() + .expect("Plugin calls lock") + .active + .get(&context.execution_id()) + .cloned(); + if let Some(call) = call { + call.cancel().await?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/worker/src/plugin/tests.rs b/crates/worker/src/plugin/tests.rs new file mode 100644 index 000000000..a3088e56b --- /dev/null +++ b/crates/worker/src/plugin/tests.rs @@ -0,0 +1,676 @@ +//! Component coverage through canonical T-791 packing, real ArtifactCache/Executor, +//! Feature installation and the Engine Tool registry. Only ArtifactSource and +//! PluginHostOwner are scripted: this is not a full Server/HTTPS vertical test, +//! nor proof of owner-side durable storage. Cancellation checks the exact owner +//! Cancel request and acknowledgement, not an inferred effect rollback. +//! Run with `cargo test -p worker --lib plugin::tests::`; no guest build is needed. + +use super::*; +use crate::feature::FeatureRegistryBuilder; +use crate::hook::{HookInvocationContext, HookRegistry, HookRegistryBuilder}; +use agen::Engine; +use agen::llm_client::client::LlmClient; +use agen::llm_client::{ClientError, Request, ResponseStream}; +use agen::state::Mutable; +use manifest::plugin::{self, PluginPackageLimits, PluginSourceKind}; +use serde_json::json; +use server_api::{ + PluginArtifact, PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest, PluginConsumer, + PluginEffectObservation, PluginHostError, PluginHostErrorCode, PluginHostOperation, + PluginHostRequest, PluginHostResponse, PluginSchemaContribution, +}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use yoi_plugin_runtime::delivery::{ArtifactStream, CacheLimits, DeliveryError}; + +const MANIFEST: &str = + include_str!("../../../../resources/plugin/templates/rust-component-tool/plugin.toml"); +const COMPONENT: &[u8] = + include_bytes!("../../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"); +const TOOL_A: &str = "installation_a__example_echo"; +const TOOL_C: &str = "installation_c__example_echo"; +const INPUT: &str = r#"{"text":"hello from a Component"}"#; + +struct Source { + artifact: PluginArtifact, + bytes: Vec, + fetches: AtomicUsize, + authorizations: AtomicUsize, +} + +#[async_trait] +impl ArtifactSource for Source { + fn server_authority_identity(&self) -> &str { + "component-fixture-owner" + } + + async fn authorize( + &self, + request: &PluginArtifactDeliveryRequest, + ) -> Result { + self.authorizations.fetch_add(1, Ordering::SeqCst); + assert_eq!(request.artifact_digest, self.artifact.artifact_digest); + Ok(PluginArtifactDeliveryMetadata { + workspace_id: "workspace-a".into(), + request: request.clone(), + artifact: self.artifact.clone(), + }) + } + + async fn fetch( + &self, + _: &PluginArtifactDeliveryRequest, + ) -> Result { + self.fetches.fetch_add(1, Ordering::SeqCst); + Ok(Box::pin(std::io::Cursor::new(self.bytes.clone()))) + } +} + +struct Fixture { + _directory: tempfile::TempDir, + source: Source, + cache: ArtifactCache, +} + +impl Fixture { + fn new() -> Self { + let directory = tempfile::tempdir().unwrap(); + std::fs::write(directory.path().join("plugin.toml"), MANIFEST).unwrap(); + std::fs::write(directory.path().join("plugin.component.wasm"), COMPONENT).unwrap(); + let limits = PluginPackageLimits::default(); + let package = + plugin::read_plugin_directory(directory.path(), PluginSourceKind::Explicit, &limits) + .unwrap(); + let packed = plugin::pack_plugin_package(&package, &limits).unwrap(); + let m = package.package.manifest; + let schema_hex: String = Sha256::digest(m.config_schema.as_bytes()) + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + let schema_digest = format!("sha256:{schema_hex}"); + let artifact = PluginArtifact { + package_id: m.id.clone(), + name: m.name, + version: m.version.clone(), + description: m.description, + package_digest: packed.package_digest, + artifact_digest: packed.artifact_digest, + artifact_size: packed.artifact_size.try_into().unwrap(), + schema_version: m.schema_version, + world: m.runtime.world, + required_imports: m.runtime.required_imports, + tools: serde_json::from_value(serde_json::to_value(m.tools).unwrap()).unwrap(), + connections: vec![], + requested_grants: vec![], + config_schema: PluginSchemaContribution { + identity: format!("{}/config", m.id), + version: m.version.clone(), + namespace: format!("{}/config/{}/{}", m.id, m.version, schema_digest), + digest: schema_digest, + dcdl: m.config_schema, + }, + revoked: false, + revocation_reason: None, + }; + let cache = + ArtifactCache::new(directory.path().join("cache"), CacheLimits::default()).unwrap(); + Self { + _directory: directory, + source: Source { + artifact, + bytes: packed.bytes, + fetches: AtomicUsize::new(0), + authorizations: AtomicUsize::new(0), + }, + cache, + } + } + + async fn prepare( + &self, + installations: &[&str], + owner: Arc, + session: &str, + ) -> PluginToolsFeature { + let pins = installations + .iter() + .map(|installation| PluginToolPin { + workspace_id: "workspace-a".into(), + installation_id: (*installation).into(), + plan_id: "persisted-plan".into(), + subject: "worker-a".into(), + request: PluginArtifactDeliveryRequest { + plan_id: "persisted-plan".into(), + plan_fingerprint: "persisted-fingerprint".into(), + installation_id: (*installation).into(), + artifact_digest: self.source.artifact.artifact_digest.clone(), + }, + expectation: ArtifactExpectation::new(self.source.artifact.clone()), + tools: vec!["example_echo".into()], + config: json!({}), + grants: Grants::default(), + }) + .collect(); + PluginToolsFeature::prepare( + pins, + &self.source, + &self.cache, + owner, + session, + Limits::default(), + ) + .await + .unwrap() + } +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Behavior { + Allow, + DenyBegin, + DenyAdmit, + DenyLatePublish, + HoldAdmit, +} + +struct Owner { + behavior: Behavior, + requests: Mutex>, + entered: Semaphore, +} + +impl Owner { + fn new(behavior: Behavior) -> Arc { + Arc::new(Self { + behavior, + requests: Mutex::new(Vec::new()), + entered: Semaphore::new(0), + }) + } + + fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + fn consumers(&self) -> Vec { + self.requests() + .into_iter() + .filter_map(|request| match request { + PluginHostRequest::Begin { consumer, .. } => Some(consumer), + _ => None, + }) + .collect() + } + + fn publications(&self) -> usize { + self.requests() + .iter() + .filter(|request| { + matches!( + request, + PluginHostRequest::Authorize { + operation: PluginHostOperation::Publish, + .. + } + ) + }) + .count() + } +} + +fn denied() -> PluginHostError { + PluginHostError { + code: PluginHostErrorCode::Denied, + effect: PluginEffectObservation::NotDispatched, + diagnostic_id: "component-fixture-denial".into(), + } +} + +#[async_trait] +impl PluginHostOwner for Owner { + async fn call( + &self, + request: PluginHostRequest, + ) -> Result { + self.requests.lock().unwrap().push(request.clone()); + match request { + PluginHostRequest::Begin { + plan_id, + installation_id, + consumer: + PluginConsumer::ToolCall { + tool_call_id, + tool_name, + }, + } => { + assert_eq!(plan_id, "persisted-plan"); + assert!(matches!( + installation_id.as_str(), + "installation_a" | "installation_c" + )); + assert_eq!(tool_name, "example_echo"); + let invocation_id = format!("owner-{installation_id}-{tool_call_id}"); + // This fixture scripts owner rejection of an already cancelled + // consumer; the production owner persistence belongs to its suite. + let cancelled = self.requests().iter().any(|request| { + matches!(request, PluginHostRequest::Cancel { invocation_id: id } if id == &invocation_id) + }); + if self.behavior == Behavior::DenyBegin || cancelled { + return Err(denied()); + } + Ok(PluginHostResponse::Invocation { + invocation_id, + expires_at_ms: (SystemTime::now().duration_since(UNIX_EPOCH).unwrap() + + Duration::from_secs(60)) + .as_millis() + .try_into() + .unwrap(), + }) + } + PluginHostRequest::Authorize { operation, .. } => { + if operation == PluginHostOperation::Admit { + if self.behavior == Behavior::DenyAdmit { + return Err(denied()); + } + if self.behavior == Behavior::HoldAdmit { + self.entered.add_permits(1); + std::future::pending::<()>().await; + } + } + // Executor publication succeeds; only the adapter's final + // publication fence is denied after the guest has returned. + if operation == PluginHostOperation::Publish + && self.behavior == Behavior::DenyLatePublish + && self.publications() == 2 + { + return Err(denied()); + } + Ok(PluginHostResponse::Authorized) + } + PluginHostRequest::Cancel { .. } => Ok(PluginHostResponse::Authorized), + other => panic!("pure Component must not request Host effects: {other:?}"), + } + } +} + +#[derive(Clone)] +struct NoLlm; + +#[async_trait] +impl LlmClient for NoLlm { + async fn stream(&self, _: Request) -> Result { + panic!("tools are driven through the Engine registry, without an LLM") + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +fn install(feature: PluginToolsFeature) -> (Engine, HookRegistry) { + let mut engine = Engine::new_annotated(NoLlm); + let mut hooks = HookRegistryBuilder::new(); + let mut registry = FeatureRegistryBuilder::new(); + registry.add_module(feature); + let report = registry.install_into_engine(&mut engine, &mut hooks); + assert!(!report.has_errors(), "{}", report.error_message()); + engine.tool_server_handle().flush_pending(); + (engine, hooks.build()) +} + +fn context(call: &str, batch: &str) -> ToolExecutionContext { + ToolExecutionContext::new(call, batch, 0) +} + +fn stopping() -> WorkerStoppingContext { + WorkerStoppingContext { + invocation: HookInvocationContext { + workspace_id: Some("workspace-a".into()), + worker_id: "worker-a".into(), + session_id: "session-a".into(), + ..Default::default() + }, + reason: "component test stop".into(), + } +} + +#[tokio::test] +async fn selected_installations_publish_only_namespaced_tools_and_manifest_schema() { + let fixture = Fixture::new(); + let owner = Owner::new(Behavior::Allow); + // The same package is available to A/B/C, but only persisted A/C pins are selected. + let feature = fixture + .prepare( + &["installation_a", "installation_c"], + owner.clone(), + "session-a", + ) + .await; + let (engine, _) = install(feature); + let tools = engine.tool_server_handle(); + let definitions = tools.tool_definitions_sorted(); + assert_eq!( + definitions + .iter() + .map(|tool| tool.name.as_str()) + .collect::>(), + [TOOL_A, TOOL_C] + ); + for (installation, name) in [("installation_a", TOOL_A), ("installation_c", TOOL_C)] { + let (meta, _) = tools.get_tool(name).unwrap(); + assert_eq!( + meta.input_schema, + fixture.source.artifact.tools[0].input_schema + ); + assert_eq!(meta.input_schema["required"], json!(["text"])); + assert_eq!(meta.input_schema["additionalProperties"], false); + let origin = meta.origin.unwrap(); + assert_eq!(origin.kind, "plugin"); + assert_eq!(origin.plugin_ref, installation); + assert_eq!(origin.plugin_id, fixture.source.artifact.package_id); + assert_eq!(origin.digest, fixture.source.artifact.package_digest); + assert_eq!(origin.surface, "tool"); + } + for absent in ["example_echo", "installation_b__example_echo"] { + assert!(tools.get_tool(absent).is_none()); + assert!( + tools + .call_tool(absent, INPUT, context("missing", "batch")) + .await + .is_err() + ); + } + assert!( + owner.requests().is_empty(), + "registration must not admit a call" + ); + assert_eq!(fixture.source.fetches.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn pure_component_returns_normal_tool_output_and_structured_json() { + let fixture = Fixture::new(); + let owner = Owner::new(Behavior::Allow); + let feature = fixture + .prepare(&["installation_a"], owner.clone(), "session-a") + .await; + let (engine, _) = install(feature); + let tools = engine.tool_server_handle(); + let output = tools + .call_tool(TOOL_A, INPUT, context("call-echo", "batch-a")) + .await + .unwrap(); + assert_eq!(output.summary, "example_echo echoed text"); + assert!(output.attachments.is_empty()); + let output: Value = serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!( + output["structured"], + json!({"tool":"example_echo", "text":"hello from a Component"}) + ); + assert!(output["content"].is_null()); + assert_eq!( + owner.publications(), + 2, + "executor and adapter each fence publication" + ); + assert!( + tools + .call_tool(TOOL_A, r#"{"text":""}"#, context("empty", "batch-a")) + .await + .is_err() + ); + assert_eq!( + owner.publications(), + 3, + "guest error passes the executor fence but never publishes through the adapter" + ); +} + +#[tokio::test] +async fn pure_component_owner_denial_applies_to_cold_and_cached_artifacts() { + for behavior in [Behavior::DenyBegin, Behavior::DenyAdmit] { + let fixture = Fixture::new(); + for session in ["cold", "cache-hit"] { + let authorizations = fixture.source.authorizations.load(Ordering::SeqCst); + let owner = Owner::new(behavior); + let feature = fixture + .prepare(&["installation_a"], owner.clone(), session) + .await; + let (engine, _) = install(feature); + let result = engine + .tool_server_handle() + .call_tool(TOOL_A, INPUT, context("denied", "batch")) + .await; + assert!( + result.is_err(), + "{session}: pure code is not an execution lease" + ); + assert_eq!(owner.consumers().len(), 1); + assert_eq!(owner.publications(), 0); + let admits = owner + .requests() + .iter() + .filter(|request| { + matches!( + request, + PluginHostRequest::Authorize { + operation: PluginHostOperation::Admit, + .. + } + ) + }) + .count(); + assert_eq!(admits, usize::from(behavior == Behavior::DenyAdmit)); + assert_eq!( + fixture.source.fetches.load(Ordering::SeqCst), + 1, + "{session}: second preparation must use the verified cache entry" + ); + assert!(fixture.source.authorizations.load(Ordering::SeqCst) > authorizations); + } + } +} + +#[tokio::test] +async fn adapter_late_publish_denial_withholds_successful_guest_output() { + let fixture = Fixture::new(); + let owner = Owner::new(Behavior::DenyLatePublish); + let feature = fixture + .prepare(&["installation_a"], owner.clone(), "session-a") + .await; + let (engine, _) = install(feature); + let error = engine + .tool_server_handle() + .call_tool(TOOL_A, INPUT, context("late", "batch")) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("Plugin result publication refused") + ); + assert_eq!( + owner.publications(), + 2, + "guest/executor fence passed before adapter denial" + ); + assert_eq!( + owner.consumers().len(), + 1, + "publication refusal must not retry the call" + ); +} + +#[tokio::test] +async fn consumer_identity_survives_batch_changes_but_isolates_sessions_and_calls() { + let fixture = Fixture::new(); + let owner = Owner::new(Behavior::Allow); + // Re-preparation simulates a restored adapter, not just reuse of a live Tool. + for (session, call, batch) in [ + ("session-a", "provider-call", "batch-before-restore"), + ("session-a", "provider-call", "batch-after-restore"), + ("session-b", "provider-call", "batch-after-restore"), + ("session-a", "different-call", "batch-after-restore"), + ] { + let feature = fixture + .prepare(&["installation_a"], owner.clone(), session) + .await; + let (engine, _) = install(feature); + engine + .tool_server_handle() + .call_tool(TOOL_A, INPUT, context(call, batch)) + .await + .unwrap(); + } + let consumers = owner.consumers(); + assert_eq!(consumers[0], consumers[1]); + assert_ne!(consumers[0], consumers[2]); + assert_ne!(consumers[0], consumers[3]); +} + +struct FailInstall; + +impl FeatureModule for FailInstall { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("late-failure", "Injected install failure") + } + + fn install(&self, _: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + Err(FeatureInstallError::Install( + "injected after Plugin staging".into(), + )) + } +} + +#[tokio::test] +async fn late_install_failure_rolls_back_plugin_tools_and_stop_hook_only() { + let fixture = Fixture::new(); + let owner = Owner::new(Behavior::Allow); + let mut engine: Engine = Engine::new_annotated(NoLlm); + let mut hooks = HookRegistryBuilder::new(); + let mut existing = FeatureRegistryBuilder::new(); + existing.add_module( + fixture + .prepare(&["installation_c"], owner.clone(), "existing") + .await, + ); + let report = existing.install_into_engine(&mut engine, &mut hooks); + assert!(!report.has_errors(), "{}", report.error_message()); + engine.tool_server_handle().flush_pending(); + + let candidate = fixture + .prepare(&["installation_a"], owner.clone(), "candidate") + .await; + let mut registry = FeatureRegistryBuilder::new(); + registry.add_module(candidate.clone()); + registry.add_module(FailInstall); + let report = registry.install_into_engine(&mut engine, &mut hooks); + assert!(report.has_errors()); + assert!( + report.plan_error.is_none(), + "failure must happen during installation" + ); + assert!( + report + .reports + .iter() + .all(|report| !report.installed && report.installed_tools.is_empty()) + ); + let tools = engine.tool_server_handle(); + tools.flush_pending(); + assert!(tools.get_tool(TOOL_C).is_some()); + assert!(tools.get_tool(TOOL_A).is_none()); + assert!(owner.requests().is_empty()); + + hooks.build().on_worker_stopping(&stopping()).await.unwrap(); + assert!( + tools + .call_tool(TOOL_C, INPUT, context("existing", "batch")) + .await + .is_err(), + "preexisting stop hook must survive rollback" + ); + let (retry, _) = install(candidate); + retry + .tool_server_handle() + .call_tool(TOOL_A, INPUT, context("retry", "batch")) + .await + .unwrap(); + // A leaked staged stop hook would have stopped the candidate before this call. + assert_eq!(owner.consumers().len(), 1); +} + +#[tokio::test] +async fn active_cancel_and_stopping_send_owner_cancel_and_refuse_reentry() { + for stop in [false, true] { + let fixture = Fixture::new(); + let owner = Owner::new(Behavior::HoldAdmit); + let feature = fixture + .prepare(&["installation_a"], owner.clone(), "session-a") + .await; + let (engine, hooks) = install(feature); + let tools = engine.tool_server_handle(); + let (_, tool) = tools.get_tool(TOOL_A).unwrap(); + let call_context = context("active", "batch-a"); + let (result, ()) = tokio::time::timeout(Duration::from_secs(10), async { + tokio::join!( + tools.call_tool(TOOL_A, INPUT, call_context.clone()), + async { + owner.entered.acquire().await.unwrap().forget(); + if stop { + hooks.on_worker_stopping(&stopping()).await.unwrap(); + } else { + tool.cancel_execution(&call_context).await.unwrap(); + } + } + ) + }) + .await + .expect("cancellation must settle the active owner wait"); + assert!(result.is_err()); + let requests = owner.requests(); + let cancels = requests + .iter() + .filter_map(|request| match request { + PluginHostRequest::Cancel { invocation_id } => Some(invocation_id), + _ => None, + }) + .collect::>(); + assert_eq!( + cancels.len(), + 1, + "local cancellation must also reach the owner" + ); + let consumers = owner.consumers(); + let PluginConsumer::ToolCall { tool_call_id, .. } = &consumers[0] else { + panic!("expected a Tool-call consumer") + }; + assert_eq!(cancels[0], &format!("owner-installation_a-{tool_call_id}")); + assert_eq!(owner.publications(), 0); + + let begins = consumers.len(); + // A fresh batch cannot revive the cancelled stable consumer. The fake + // owner denies it; stopping instead fences all new calls locally. + let retry = tokio::time::timeout( + Duration::from_secs(5), + tools.call_tool(TOOL_A, INPUT, context("active", "batch-b")), + ) + .await + .expect("cancelled consumer must not reenter admission"); + assert!(retry.is_err()); + assert_eq!(owner.consumers().len(), begins + usize::from(!stop)); + if stop { + assert!( + tools + .call_tool(TOOL_A, INPUT, context("new-call", "batch-b")) + .await + .is_err() + ); + assert_eq!( + owner.consumers().len(), + begins, + "stopped Tool must not ask Begin" + ); + } + } +} diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index f09399780..29b4cc1b5 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -4232,6 +4232,21 @@ where &self.worker().manifest } + /// Stable Session identity used to correlate restored Plugin Tool calls. + pub fn session_id(&self) -> SessionId { + self.worker().session_id() + } + + /// Bind exact Host-prepared contributions without exposing execution or + /// reopening a candidate whose cleanup has begun. + pub fn bind_plugin_tools( + &mut self, + tools: crate::plugin::PluginToolsFeature, + ) -> Result<(), WorkerError> { + self.worker_mut()?.bind_plugin_tools(tools); + Ok(()) + } + /// Retain the same Workdir resource owner for explicit cleanup, including /// after discard fails. This does not grant Worker input or execution access. pub fn workdir_cleanup(&self) -> RestoreCandidateWorkdirCleanup { @@ -4485,6 +4500,7 @@ pub struct Worker { /// Host-owned projection of Worker sessions explicitly granted to this Worker. /// The provider reauthorizes every capture and never derives authority from model input. worker_observation_provider: Option>, + plugin_tools: Option, /// Parsed system-prompt template awaiting first-turn materialisation. /// Retained until the complete initial seed and active metadata are durable. system_prompt_template: Option, @@ -4853,6 +4869,7 @@ impl Worker { tracker: None, task_feature: TaskFeature::new(), worker_observation_provider: None, + plugin_tools: None, system_prompt_template: None, feature_instructions: Vec::new(), alerter: None, @@ -5367,6 +5384,16 @@ impl Worker { } } + /// Bind Host-prepared exact Plugin contributions before controller start. + /// This handle is never persisted in a Manifest or inherited by a SubWorker. + pub fn bind_plugin_tools(&mut self, tools: crate::plugin::PluginToolsFeature) { + self.plugin_tools = Some(tools); + } + + pub(crate) fn plugin_tools(&self) -> Option { + self.plugin_tools.clone() + } + /// Install enabled feature modules into the Worker host surfaces. pub fn install_features( &mut self, @@ -9579,6 +9606,7 @@ where tracker: None, task_feature: TaskFeature::new(), worker_observation_provider: None, + plugin_tools: None, system_prompt_template: common.system_prompt_template, feature_instructions: common.feature_instructions, alerter: None, @@ -9987,6 +10015,7 @@ where tracker: None, task_feature: TaskFeature::new(), worker_observation_provider: None, + plugin_tools: None, system_prompt_template: common.system_prompt_template, feature_instructions: common.feature_instructions, alerter: None, @@ -10452,6 +10481,7 @@ where tracker: None, task_feature, worker_observation_provider: None, + plugin_tools: None, // Restore replays the saved system_prompt verbatim — no // template re-render on resume. system_prompt_template: None, diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index fd9096bc4..955b7d83e 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -26,7 +26,7 @@ workspace-drive.workspace = true subjektiv.workspace = true flow = { path = "../flow" } manifest.workspace = true -yoi-plugin-runtime = { path = "../plugin-runtime" } +plugin-plan.workspace = true job.workspace = true protocol = { workspace = true } runtime-api.workspace = true diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index e72990b20..e192e2de3 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -517,6 +517,7 @@ pub struct WorkerControlOperation { pub struct WorkerCreateBinding { pub worker_id: EmbeddedWorkerId, pub create_fingerprint: String, + pub plugin_execution: Option, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -6130,6 +6131,7 @@ fn runtime_create_worker_request( worker_observation_grants: request.resolved_worker_observation_grants.clone(), workspace_api: Some(workspace_api), memory_settings: request.resolved_memory_settings.clone(), + plugin_execution: binding.plugin_execution, subjektiv_attached: request.resolved_subjektiv_attached, backend_job: match &request.intent { WorkerSpawnIntent::BackendJob { @@ -7247,6 +7249,7 @@ mod tests { fn test_create_binding() -> WorkerCreateBinding { WorkerCreateBinding { + plugin_execution: None, worker_id: EmbeddedWorkerId::now_v7(), create_fingerprint: "sha256:test-create".to_string(), } diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 33e2a9190..9fe3f4f43 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -14,9 +14,9 @@ pub mod memory_staging; pub mod observation; pub mod plugin_binding; pub mod plugin_catalog; +pub mod plugin_delivery; pub mod plugin_host; pub mod plugin_https; -pub mod plugin_delivery; pub mod plugin_plan; pub mod plugin_receipt; pub mod plugin_settings; diff --git a/crates/workspace-server/src/plugin_plan.rs b/crates/workspace-server/src/plugin_plan.rs index de0dfa234..35960ff30 100644 --- a/crates/workspace-server/src/plugin_plan.rs +++ b/crates/workspace-server/src/plugin_plan.rs @@ -10,12 +10,14 @@ //! Their local savepoints also roll back partial pin acquisition if an owner //! catches an error and commits other work. There is no cross-store compensation. use crate::{Error, Result, plugin_catalog::PluginCatalogError}; -use manifest::plugin; use serde::{Deserialize, Serialize}; -use serde_json::Value; -use server_api::{PluginArtifact, PluginGrantMetadata}; -use sha2::{Digest, Sha256}; -use std::collections::BTreeSet; + +pub use plugin_plan::{ + ArtifactDescriptor, ArtifactIdentityProjection, BindingPolicyReference, ConfigProvenance, + HostExecutionIdentity, PlanError, PluginGrant, PluginPinProjection, PluginSchemaIdentity, + PluginSurface, ProfileProvenance, WorkerPluginExecutionPlan, WorkerPluginPin, + WorkerPluginPlanInput, +}; mod store; pub use store::*; @@ -23,343 +25,13 @@ pub use store::*; /// Included by the config owner's ordered database migration, never at runtime. pub const SCHEMA_SQL: &str = include_str!("plugin_plan/schema.sql"); -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ConfigProvenance { - pub content_digest: String, - pub toolchain_fingerprint: String, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ProfileProvenance { - pub identity: String, - pub projection_digest: String, -} - -/// Supplied by the authenticated Host launch/delegation boundary, never by guest -/// or model input. These are identities only, not credentials or provider handles. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct HostExecutionIdentity { - pub runtime_id: String, - pub worker_id: String, - pub principal_id: String, - pub delegation_id: Option, -} - -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct BindingPolicyReference { - pub slot: String, - pub binding_id: String, - pub policy_digest: String, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum PluginSurface { - Tool, -} - -/// Only capabilities implemented by the initial Tool Host are representable. -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] -pub enum PluginGrant { - Logging, - Https { slot: String }, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PluginSchemaIdentity { - pub identity: String, - pub version: String, - pub digest: String, - pub namespace: String, -} - -/// The upstream config/schema authority must validate config values as non-secret -/// and intersect grants/surfaces with Workspace, Profile and delegation bounds. -/// Binding policies remain with their owner: only identities and digests travel. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PluginPinProjection { - pub config: Value, - pub surfaces: Vec, - pub tools: Vec, - pub grants: Vec, - pub bindings: Vec, -} - -/// All fields are private: construct from admitted artifact metadata, not a -/// hand-built wire snapshot. Deserialization alone does not establish admission. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct WorkerPluginPin { - installation_id: String, - expected_last_operation: String, - package_id: String, - package_version: String, - package_digest: String, - artifact_digest: String, - artifact_size: u32, - package_schema_version: u32, - component_world: String, - required_imports: Vec, - config_schema: PluginSchemaIdentity, - config_digest: String, - projection: PluginPinProjection, -} - -impl WorkerPluginPin { - pub fn from_artifact( - installation_id: String, - expected_last_operation: String, - artifact: &PluginArtifact, - mut projection: PluginPinProjection, - ) -> Result { - if artifact.revoked { - return conflict(); - } - if !valid_identity(&installation_id) - || !valid_identity(&expected_last_operation) - || !valid_digest(&artifact.package_digest) - || !valid_digest(&artifact.artifact_digest) - || artifact.artifact_size == 0 - || artifact.schema_version != plugin::SUPPORTED_PLUGIN_API_VERSION - || !matches!( - artifact.world.as_str(), - plugin::PLUGIN_COMPONENT_TOOL_WORLD - | plugin::PLUGIN_COMPONENT_HTTPS_TOOL_WORLD - | plugin::PLUGIN_COMPONENT_LOGGED_TOOL_WORLD - | plugin::PLUGIN_COMPONENT_HTTPS_LOGGED_TOOL_WORLD - ) - || !valid_digest(&artifact.config_schema.digest) - || !valid_identity(&artifact.package_id) - || !valid_identity(&artifact.version) - || !valid_identity(&artifact.config_schema.identity) - || !valid_identity(&artifact.config_schema.version) - || !valid_identity(&artifact.config_schema.namespace) - || !projection.config.is_object() - { - return invalid(); - } - projection.surfaces.sort(); - projection.surfaces.dedup(); - projection.tools.sort(); - projection.tools.dedup(); - projection.grants.sort(); - projection.grants.dedup(); - projection.bindings.sort(); - projection.config = canonical_value(&projection.config); - if (!projection.tools.is_empty() && !projection.surfaces.contains(&PluginSurface::Tool)) - || projection.tools.iter().any(|name| { - !valid_identity(name) || !artifact.tools.iter().any(|tool| tool.name == *name) - }) - || projection - .bindings - .windows(2) - .any(|b| b[0].slot == b[1].slot) - || projection.bindings.iter().any(|binding| { - !valid_identity(&binding.slot) - || !valid_identity(&binding.binding_id) - || !valid_digest(&binding.policy_digest) - || !artifact - .connections - .iter() - .any(|slot| slot.name == binding.slot) - }) - || projection.grants.iter().any(|grant| match grant { - PluginGrant::Logging => !artifact - .requested_grants - .contains(&PluginGrantMetadata::Logging), - PluginGrant::Https { slot } => { - !valid_identity(slot) - || !artifact - .requested_grants - .contains(&PluginGrantMetadata::Https { slot: slot.clone() }) - || !projection - .bindings - .iter() - .any(|binding| binding.slot == *slot) - } - }) - || projection.bindings.iter().any(|binding| { - !projection.grants.contains(&PluginGrant::Https { - slot: binding.slot.clone(), - }) - }) - { - return invalid(); - } - let mut required_imports = artifact.required_imports.clone(); - required_imports.sort(); - required_imports.dedup(); - Ok(Self { - installation_id, - expected_last_operation, - package_id: artifact.package_id.clone(), - package_version: artifact.version.clone(), - package_digest: artifact.package_digest.clone(), - artifact_digest: artifact.artifact_digest.clone(), - artifact_size: artifact.artifact_size, - package_schema_version: artifact.schema_version, - component_world: artifact.world.clone(), - required_imports, - config_schema: PluginSchemaIdentity { - identity: artifact.config_schema.identity.clone(), - version: artifact.config_schema.version.clone(), - digest: artifact.config_schema.digest.clone(), - namespace: artifact.config_schema.namespace.clone(), - }, - config_digest: digest_json(&projection.config)?, - projection, +impl From for Error { + fn from(error: PlanError) -> Self { + Self::PluginCatalog(match error { + PlanError::Invalid => PluginCatalogError::Invalid, + PlanError::Conflict => PluginCatalogError::Conflict, }) } - - pub fn installation_id(&self) -> &str { - &self.installation_id - } - pub fn expected_last_operation(&self) -> &str { - &self.expected_last_operation - } - pub fn package_id(&self) -> &str { - &self.package_id - } - pub fn package_version(&self) -> &str { - &self.package_version - } - pub fn package_schema_version(&self) -> u32 { - self.package_schema_version - } - pub fn component_world(&self) -> &str { - &self.component_world - } - pub fn required_imports(&self) -> &[String] { - &self.required_imports - } - pub fn package_digest(&self) -> &str { - &self.package_digest - } - pub fn artifact_digest(&self) -> &str { - &self.artifact_digest - } - pub fn artifact_size(&self) -> u32 { - self.artifact_size - } - pub fn config_schema(&self) -> &PluginSchemaIdentity { - &self.config_schema - } - pub fn config_digest(&self) -> &str { - &self.config_digest - } - pub fn projection(&self) -> &PluginPinProjection { - &self.projection - } - - pub(super) fn matches_artifact(&self, artifact: &PluginArtifact) -> Result { - Ok(Self::from_artifact( - self.installation_id.clone(), - self.expected_last_operation.clone(), - artifact, - self.projection.clone(), - )? == *self) - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct WorkerPluginPlanInput { - pub workspace_id: String, - pub config: ConfigProvenance, - pub profile: ProfileProvenance, - pub execution: HostExecutionIdentity, - pub plugins: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct WorkerPluginExecutionPlan { - format_version: u32, - input: WorkerPluginPlanInput, -} - -impl WorkerPluginExecutionPlan { - pub fn canonicalize(mut input: WorkerPluginPlanInput) -> Result { - if !valid_identity(&input.workspace_id) - || !valid_digest(&input.config.content_digest) - || !valid_identity(&input.config.toolchain_fingerprint) - || !valid_identity(&input.profile.identity) - || !valid_digest(&input.profile.projection_digest) - || !valid_identity(&input.execution.runtime_id) - || !valid_identity(&input.execution.worker_id) - || !valid_identity(&input.execution.principal_id) - || input - .execution - .delegation_id - .as_ref() - .is_some_and(|id| !valid_identity(id)) - || input.plugins.len() > 128 - { - return invalid(); - } - input - .plugins - .sort_by(|a, b| a.installation_id.cmp(&b.installation_id)); - // One selection per installation; two aliases of one package instead use - // distinct installation identities, as required by the catalog contract. - if input - .plugins - .windows(2) - .any(|pins| pins[0].installation_id == pins[1].installation_id) - { - return invalid(); - } - // Public Tool names use immutable installation identities, never aliases. - let mut names = BTreeSet::new(); - for pin in &input.plugins { - for tool in &pin.projection.tools { - let name = format!("{}__{}", pin.installation_id, tool); - if name.len() > 128 || !names.insert(name) { - return invalid(); - } - } - } - Ok(Self { - format_version: 1, - input, - }) - } - - pub fn workspace_id(&self) -> &str { - &self.input.workspace_id - } - pub fn config(&self) -> &ConfigProvenance { - &self.input.config - } - pub fn profile(&self) -> &ProfileProvenance { - &self.input.profile - } - pub fn execution(&self) -> &HostExecutionIdentity { - &self.input.execution - } - pub fn plugins(&self) -> &[WorkerPluginPin] { - &self.input.plugins - } - pub fn fingerprint(&self) -> Result { - digest_json(self) - } - pub fn canonical_json(&self) -> Result { - encode_canonical(self) - } - - pub(super) fn validate_canonical(&self) -> Result<()> { - if self.format_version != 1 || Self::canonicalize(self.input.clone())? != *self { - return invalid(); - } - Ok(()) - } } fn invalid() -> Result { @@ -374,46 +46,6 @@ fn not_found() -> Result { fn valid_identity(value: &str) -> bool { !value.is_empty() && value.len() <= 512 && !value.chars().any(char::is_control) } -fn valid_digest(value: &str) -> bool { - value.strip_prefix("sha256:").is_some_and(|hex| { - hex.len() == 64 - && hex - .bytes() - .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) - }) -} -fn canonical_value(value: &Value) -> Value { - match value { - Value::Object(fields) => { - let mut keys: Vec<_> = fields.keys().collect(); - keys.sort(); - Value::Object( - keys.into_iter() - .map(|key| (key.clone(), canonical_value(&fields[key]))) - .collect(), - ) - } - Value::Array(values) => Value::Array(values.iter().map(canonical_value).collect()), - _ => value.clone(), - } -} -fn encode_canonical(value: &impl Serialize) -> Result { - let value = serde_json::to_value(value) - .map_err(|_| Error::PluginCatalog(PluginCatalogError::Invalid))?; - serde_json::to_string(&canonical_value(&value)) - .map_err(|_| Error::PluginCatalog(PluginCatalogError::Invalid)) -} -fn digest_json(value: &impl Serialize) -> Result { - let bytes = encode_canonical(value)?; - Ok(hash_bytes(bytes.as_bytes())) -} -fn hash_bytes(bytes: &[u8]) -> String { - let hex: String = Sha256::digest(bytes) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect(); - format!("sha256:{hex}") -} #[cfg(test)] mod tests; diff --git a/crates/workspace-server/src/plugin_plan/store.rs b/crates/workspace-server/src/plugin_plan/store.rs index 40c8f6973..48276e48a 100644 --- a/crates/workspace-server/src/plugin_plan/store.rs +++ b/crates/workspace-server/src/plugin_plan/store.rs @@ -112,10 +112,10 @@ pub fn plugin_plan_create( plugin_catalog::plugin_artifact_admit( tx, plan.workspace_id(), - &pin.installation_id, - &pin.expected_last_operation, - &pin.package_digest, - &pin.artifact_digest, + pin.installation_id(), + pin.expected_last_operation(), + pin.package_digest(), + pin.artifact_digest(), )?; } crate::plugin_settings::validate_plan_source(tx, plan)?; @@ -128,10 +128,10 @@ pub fn plugin_plan_create( let artifact = plugin_catalog::plugin_artifact_admit( tx, plan.workspace_id(), - &pin.installation_id, - &pin.expected_last_operation, - &pin.package_digest, - &pin.artifact_digest, + pin.installation_id(), + pin.expected_last_operation(), + pin.package_digest(), + pin.artifact_digest(), )?; if !pin.matches_artifact(&artifact)? { return conflict(); @@ -142,10 +142,10 @@ pub fn plugin_plan_create( &PluginReference { kind: PluginReferenceKind::Pending, owner: plan_id.clone(), - installation_id: pin.installation_id.clone(), - expected_last_operation: pin.expected_last_operation.clone(), - package_digest: pin.package_digest.clone(), - artifact_digest: pin.artifact_digest.clone(), + installation_id: pin.installation_id().to_owned(), + expected_last_operation: pin.expected_last_operation().to_owned(), + package_digest: pin.package_digest().to_owned(), + artifact_digest: pin.artifact_digest().to_owned(), }, )?; } @@ -233,7 +233,7 @@ pub fn plugin_plan_transition( for pin in saved.plan.plugins() { let operation: String = tx.query_row( "SELECT last_operation FROM plugin_installations WHERE workspace_id=?1 AND installation_id=?2", - params![workspace, pin.installation_id], |row| row.get(0), + params![workspace, pin.installation_id()], |row| row.get(0), )?; plugin_catalog::plugin_reference_put( tx, @@ -241,10 +241,10 @@ pub fn plugin_plan_transition( &PluginReference { kind, owner: plan_id.into(), - installation_id: pin.installation_id.clone(), + installation_id: pin.installation_id().to_owned(), expected_last_operation: operation, - package_digest: pin.package_digest.clone(), - artifact_digest: pin.artifact_digest.clone(), + package_digest: pin.package_digest().to_owned(), + artifact_digest: pin.artifact_digest().to_owned(), }, )?; } @@ -256,8 +256,8 @@ pub fn plugin_plan_transition( workspace, kind, plan_id, - &pin.installation_id, - &pin.artifact_digest, + pin.installation_id(), + pin.artifact_digest(), )?; } } diff --git a/crates/workspace-server/src/plugin_plan/tests.rs b/crates/workspace-server/src/plugin_plan/tests.rs index c0a7e3db1..e0c0bee7f 100644 --- a/crates/workspace-server/src/plugin_plan/tests.rs +++ b/crates/workspace-server/src/plugin_plan/tests.rs @@ -2,8 +2,11 @@ //! Component. Execution/Host side effects are intentionally outside this boundary. use super::*; use crate::store::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord}; +use manifest::plugin; use rusqlite::{Connection, TransactionBehavior, params}; +use serde_json::Value; use server_api::*; +use sha2::{Digest, Sha256}; use std::fs; use tempfile::{TempDir, tempdir}; @@ -111,7 +114,17 @@ fn package(version: &str) -> Vec { .bytes } fn digest(seed: &str) -> String { - hash_bytes(seed.as_bytes()) + let hex: String = Sha256::digest(seed.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + format!("sha256:{hex}") +} +// Exercise untrusted serialized snapshots without exposing mutable pin internals. +fn with_pin_field(pin: &WorkerPluginPin, field: &str, value: Value) -> WorkerPluginPin { + let mut wire = serde_json::to_value(pin).unwrap(); + wire[field] = value; + serde_json::from_value(wire).unwrap() } fn projection() -> PluginPinProjection { PluginPinProjection { @@ -202,13 +215,25 @@ async fn canonical_order_and_fingerprint_bind_all_provenance_and_both_digests() changed.execution.delegation_id = Some("different".into()); changes.push(changed); let mut changed = base.clone(); - changed.plugins[0].package_digest = digest("new package"); + changed.plugins[0] = with_pin_field( + &changed.plugins[0], + "package_digest", + digest("new package").into(), + ); changes.push(changed); let mut changed = base.clone(); - changed.plugins[0].artifact_digest = digest("new artifact"); + changed.plugins[0] = with_pin_field( + &changed.plugins[0], + "artifact_digest", + digest("new artifact").into(), + ); changes.push(changed); let mut changed = base; - changed.plugins[0].expected_last_operation = "new observation".into(); + changed.plugins[0] = with_pin_field( + &changed.plugins[0], + "expected_last_operation", + "new observation".into(), + ); changes.push(changed); for changed in changes { assert_ne!( @@ -220,7 +245,7 @@ async fn canonical_order_and_fingerprint_bind_all_provenance_and_both_digests() ); } error( - WorkerPluginExecutionPlan::canonicalize(input(vec![pin(&a), pin(&a)])), + WorkerPluginExecutionPlan::canonicalize(input(vec![pin(&a), pin(&a)])).map_err(Into::into), PluginCatalogError::Invalid, ); } @@ -326,7 +351,7 @@ async fn create_replay_is_immutable_across_restart_and_rejects_changed_input() { .pin_state, PlanPinState::Active ); - let mut changed = plan.input.clone(); + let mut changed = plan.input().clone(); changed.profile.projection_digest = digest("new profile"); error( reopened.plugin_plan_create( @@ -541,10 +566,14 @@ async fn owner_savepoint_and_outer_transaction_rollback_leave_no_partial_plan_or let f = Fixture::new().await; let first = f.install("one"); let second = f.install("two"); - let mut bad_input = f.configure(&[&first, &second]).input; + let mut bad_input = f.configure(&[&first, &second]).input().clone(); let mut pins = bad_input.plugins.clone(); - pins.sort_by(|a, b| a.installation_id.cmp(&b.installation_id)); - pins[1].artifact_size += 1; + pins.sort_by(|a, b| a.installation_id().cmp(b.installation_id())); + pins[1] = with_pin_field( + &pins[1], + "artifact_size", + (pins[1].artifact_size() + 1).into(), + ); bad_input.plugins = pins; let bad_plan = WorkerPluginExecutionPlan::canonicalize(bad_input).unwrap(); f.store @@ -583,7 +612,7 @@ async fn cross_workspace_and_exact_digest_mismatches_fail_without_retaining_refe let f = Fixture::new().await; let installed = f.install("one"); let base = f.configure(&[&installed]); - let mut other_workspace = base.input.clone(); + let mut other_workspace = base.input().clone(); other_workspace.workspace_id = "b".into(); error( f.store.plugin_plan_create( @@ -594,11 +623,19 @@ async fn cross_workspace_and_exact_digest_mismatches_fail_without_retaining_refe PluginCatalogError::NotFound, ); for change_package in [true, false] { - let mut changed = base.input.clone(); + let mut changed = base.input().clone(); if change_package { - changed.plugins[0].package_digest = digest("wrong-package"); + changed.plugins[0] = with_pin_field( + &changed.plugins[0], + "package_digest", + digest("wrong-package").into(), + ); } else { - changed.plugins[0].artifact_digest = digest("wrong-artifact"); + changed.plugins[0] = with_pin_field( + &changed.plugins[0], + "artifact_digest", + digest("wrong-artifact").into(), + ); } let result = f.store.plugin_plan_create( "actor", @@ -749,8 +786,8 @@ impl SqliteWorkspaceStore { let two = f.install("migration-two"); let plan = f.configure(&[&one, &two]); assert_ne!( - plan.plugins()[0].package_digest, - plan.plugins()[0].artifact_digest + plan.plugins()[0].package_digest(), + plan.plugins()[0].artifact_digest() ); let mut receipts = Vec::new(); for (key, state) in [ diff --git a/crates/workspace-server/src/plugin_settings.rs b/crates/workspace-server/src/plugin_settings.rs index 36899e56b..13458b2df 100644 --- a/crates/workspace-server/src/plugin_settings.rs +++ b/crates/workspace-server/src/plugin_settings.rs @@ -564,6 +564,7 @@ pub fn resolve_worker_plugin_plan( execution, plugins: project_profile_plugins(&resolved, &profile.manifest.plugins)?, }) + .map_err(Into::into) } fn resolve_profile( diff --git a/crates/workspace-server/src/runtime_subscription_tests.rs b/crates/workspace-server/src/runtime_subscription_tests.rs index 73dbff4d3..f5c53b057 100644 --- a/crates/workspace-server/src/runtime_subscription_tests.rs +++ b/crates/workspace-server/src/runtime_subscription_tests.rs @@ -51,6 +51,7 @@ const TOKEN: &str = "runtime-subscription-test-token"; fn create_request(name: &str) -> CreateWorkerRequest { CreateWorkerRequest { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index a502d7b62..911b4b851 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -48,10 +48,10 @@ mod drive; mod drive_grants; #[path = "server_plugin.rs"] mod plugin; -#[path = "server_plugin_host.rs"] -mod plugin_host; #[path = "server_plugin_delivery.rs"] mod plugin_delivery; +#[path = "server_plugin_host.rs"] +mod plugin_host; pub use plugin_delivery::InProcessPluginArtifactSource; #[path = "server_workspace_config.rs"] mod workspace_config; @@ -6066,6 +6066,7 @@ impl WorkspaceApi { request.resolved_memory_settings = Some(reservation.memory_settings); let reservation_fingerprint = reservation.create_fingerprint.clone(); let create_binding = WorkerCreateBinding { + plugin_execution: None, worker_id, create_fingerprint: reservation.create_fingerprint, }; @@ -43498,6 +43499,7 @@ mod tests { fn test_create_binding() -> WorkerCreateBinding { WorkerCreateBinding { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "sha256:test-create".to_string(), } @@ -52175,6 +52177,7 @@ mod tests { .create_worker_scoped( &worker_runtime::RuntimeWorkspaceScope::new(TEST_WORKSPACE_ID, "embedded-backend"), worker_runtime::catalog::CreateWorkerRequest { + plugin_execution: None, worker_id: binding.worker_id, create_fingerprint: binding.create_fingerprint, profile: ProfileSelector::Builtin(request.profile.clone()), @@ -65670,6 +65673,7 @@ mod tests { let mut memory_settings = test_worker_memory_settings(); memory_settings.workspace_id = "local".to_owned(); worker_runtime::catalog::CreateWorkerRequest { + plugin_execution: None, worker_id: WorkerId::now_v7(), create_fingerprint: "test-create".to_string(), profile: worker_runtime::catalog::ProfileSelector::Builtin( diff --git a/openapi/server-api.json b/openapi/server-api.json index ff972d481..c94b247d8 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -19818,7 +19818,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41" + "x-yoi-source-digest": "sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55" }, "openapi": "3.1.0", "paths": { diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index dfbc8a1eb..cbb64ae82 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:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41) +// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) // 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/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 039be0c4c..00fef6b0c 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:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41) +// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) // 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 5adac08c4..6f08b8bd3 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:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41) +// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) // 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 da3af1e44..7a5eefb9d 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:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41) +// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) // 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/workdir-api.ts b/web/workspace/src/lib/generated/workdir-api.ts index 423a4f5d0..ed1a8e853 100644 --- a/web/workspace/src/lib/generated/workdir-api.ts +++ b/web/workspace/src/lib/generated/workdir-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-workdir-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41) +// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/workdir-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_workdir_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 a04e08d4f..f7b406e6e 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:4b347d659dbdedb2ae773c617ee8a8b8acb13936860a93d8dd0f651af6e4fb41) +// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) // 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