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.
This commit is contained in:
Generated
+17
@@ -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",
|
||||
|
||||
@@ -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" }
|
||||
|
||||
@@ -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
|
||||
@@ -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.
|
||||
@@ -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"}}
|
||||
@@ -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<T> = std::result::Result<T, PlanError>;
|
||||
|
||||
/// 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<String>,
|
||||
pub tools: Vec<String>,
|
||||
pub connections: Vec<String>,
|
||||
pub requested_grants: Vec<PluginGrant>,
|
||||
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<String>,
|
||||
}
|
||||
|
||||
#[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<PluginSurface>,
|
||||
pub tools: Vec<String>,
|
||||
pub grants: Vec<PluginGrant>,
|
||||
pub bindings: Vec<BindingPolicyReference>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
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<Self> {
|
||||
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<bool> {
|
||||
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<WorkerPluginPin>,
|
||||
}
|
||||
|
||||
#[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<Self> {
|
||||
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<String> {
|
||||
digest_json(self)
|
||||
}
|
||||
pub fn canonical_json(&self) -> Result<String> {
|
||||
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<T>() -> Result<T> {
|
||||
Err(PlanError::Invalid)
|
||||
}
|
||||
fn conflict<T>() -> Result<T> {
|
||||
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<String> {
|
||||
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<String> {
|
||||
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;
|
||||
@@ -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> {
|
||||
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::<WorkerPluginExecutionPlan>(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));
|
||||
}
|
||||
@@ -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)]
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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<WorkspaceApiRef>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub memory_settings: Option<manifest::WorkspaceMemorySettingsSnapshot>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub plugin_execution: Option<PluginExecutionSnapshot>,
|
||||
/// 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")]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -122,6 +122,44 @@ pub struct PluginArtifact {
|
||||
pub revocation_reason: Option<String>,
|
||||
}
|
||||
|
||||
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)]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<manifest::WorkspaceMemorySettingsSnapshot>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub plugin_execution: Option<runtime_api::PluginExecutionSnapshot>,
|
||||
/// 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")]
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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")]
|
||||
|
||||
@@ -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<worker::plugin::PluginToolsFeature, String>;
|
||||
}
|
||||
|
||||
/// 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<dyn yoi_plugin_runtime::owner::PluginHostOwner>,
|
||||
cache: &yoi_plugin_runtime::delivery::ArtifactCache,
|
||||
limits: yoi_plugin_runtime::Limits,
|
||||
) -> Result<worker::plugin::PluginToolsFeature, String> {
|
||||
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())
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -825,6 +825,7 @@ type SpawnModelClientFactory =
|
||||
#[derive(Clone)]
|
||||
pub struct ProfileRuntimeWorkerFactory {
|
||||
spawn_model_client_factory: Option<SpawnModelClientFactory>,
|
||||
plugin_runtime: Option<Arc<dyn crate::plugin::PluginToolRuntime>>,
|
||||
observation_hub: Arc<RuntimeWorkerObservationHub>,
|
||||
profile_base_dir: PathBuf,
|
||||
worker_aggregate_root: Option<PathBuf>,
|
||||
@@ -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<dyn crate::plugin::PluginToolRuntime>,
|
||||
) -> 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()),
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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<String>,
|
||||
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<Option<Arc<OwnerHost>>>,
|
||||
}
|
||||
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<String, Arc<ActiveCall>>,
|
||||
}
|
||||
struct CallLease {
|
||||
key: String,
|
||||
state: Arc<Mutex<Calls>>,
|
||||
call: Arc<ActiveCall>,
|
||||
}
|
||||
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<PluginTool>)>,
|
||||
calls: Arc<Mutex<Calls>>,
|
||||
}
|
||||
impl PluginToolsFeature {
|
||||
pub async fn prepare(
|
||||
pins: Vec<PluginToolPin>,
|
||||
source: &dyn ArtifactSource,
|
||||
cache: &ArtifactCache,
|
||||
owner: Arc<dyn PluginHostOwner>,
|
||||
session_id: &str,
|
||||
limits: Limits,
|
||||
) -> Result<Self, PluginPreparationError> {
|
||||
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::<BTreeSet<_>>().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<WorkerStopping> 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::<Vec<_>>()
|
||||
};
|
||||
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<Executor>,
|
||||
compiled: Arc<CompiledTool>,
|
||||
owner: Arc<dyn PluginHostOwner>,
|
||||
workspace: String,
|
||||
installation: String,
|
||||
plan: String,
|
||||
subject: String,
|
||||
grants: Grants,
|
||||
tool_name: String,
|
||||
config_json: String,
|
||||
session_id: String,
|
||||
calls: Arc<Mutex<Calls>>,
|
||||
capacity: Arc<Semaphore>,
|
||||
}
|
||||
fn failure(message: &str) -> ToolError {
|
||||
ToolError::Internal(message.into())
|
||||
}
|
||||
impl PluginTool {
|
||||
fn start(&self, context: &ToolExecutionContext) -> Result<CallLease, ToolError> {
|
||||
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<ToolOutput, ToolError> {
|
||||
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::<Value>(&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;
|
||||
@@ -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<u8>,
|
||||
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<PluginArtifactDeliveryMetadata, DeliveryError> {
|
||||
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<ArtifactStream, DeliveryError> {
|
||||
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<Owner>,
|
||||
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<Vec<PluginHostRequest>>,
|
||||
entered: Semaphore,
|
||||
}
|
||||
|
||||
impl Owner {
|
||||
fn new(behavior: Behavior) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
behavior,
|
||||
requests: Mutex::new(Vec::new()),
|
||||
entered: Semaphore::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
fn requests(&self) -> Vec<PluginHostRequest> {
|
||||
self.requests.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
fn consumers(&self) -> Vec<PluginConsumer> {
|
||||
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<PluginHostResponse, PluginHostError> {
|
||||
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<ResponseStream, ClientError> {
|
||||
panic!("tools are driven through the Engine registry, without an LLM")
|
||||
}
|
||||
|
||||
fn clone_boxed(&self) -> Box<dyn LlmClient> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
}
|
||||
|
||||
fn install(feature: PluginToolsFeature) -> (Engine<NoLlm, Mutable, ()>, 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::<Vec<_>>(),
|
||||
[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<NoLlm, Mutable, ()> = 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::<Vec<_>>();
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<C: LlmClient, St: Store> {
|
||||
/// 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<Arc<dyn WorkerObservationProvider>>,
|
||||
plugin_tools: Option<crate::plugin::PluginToolsFeature>,
|
||||
/// Parsed system-prompt template awaiting first-turn materialisation.
|
||||
/// Retained until the complete initial seed and active metadata are durable.
|
||||
system_prompt_template: Option<SystemPromptTemplate>,
|
||||
@@ -4853,6 +4869,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
|
||||
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<crate::plugin::PluginToolsFeature> {
|
||||
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -517,6 +517,7 @@ pub struct WorkerControlOperation {
|
||||
pub struct WorkerCreateBinding {
|
||||
pub worker_id: EmbeddedWorkerId,
|
||||
pub create_fingerprint: String,
|
||||
pub plugin_execution: Option<runtime_api::PluginExecutionSnapshot>,
|
||||
}
|
||||
|
||||
#[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(),
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
#[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<PluginSurface>,
|
||||
pub tools: Vec<String>,
|
||||
pub grants: Vec<PluginGrant>,
|
||||
pub bindings: Vec<BindingPolicyReference>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
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<Self> {
|
||||
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<PlanError> 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<bool> {
|
||||
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<WorkerPluginPin>,
|
||||
}
|
||||
|
||||
#[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<Self> {
|
||||
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<String> {
|
||||
digest_json(self)
|
||||
}
|
||||
pub fn canonical_json(&self) -> Result<String> {
|
||||
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<T>() -> Result<T> {
|
||||
@@ -374,46 +46,6 @@ fn not_found<T>() -> Result<T> {
|
||||
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<String> {
|
||||
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<String> {
|
||||
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;
|
||||
|
||||
@@ -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(),
|
||||
)?;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<u8> {
|
||||
.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 [
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user