From 382db71de022816e59fab92d6023c49312f9165b Mon Sep 17 00:00:00 2001 From: Reiase Date: Sun, 26 Jul 2026 22:47:04 +0800 Subject: [PATCH 1/2] Add pVisor runtime and Capture access control --- Cargo.lock | 19 +- Cargo.toml | 2 +- crates/persisting-capture/Cargo.toml | 3 +- crates/persisting-capture/src/proxy/common.rs | 25 + .../persisting-capture/src/proxy/dispatch.rs | 41 +- .../persisting-capture/src/proxy/forward.rs | 20 +- .../src/proxy/llm_capture.rs | 38 ++ crates/persisting-capture/src/proxy/mod.rs | 5 +- .../src/proxy/network_policy.rs | 262 ++++---- crates/persisting-capture/src/proxy/state.rs | 28 + .../tests/network_policy_http.rs | 100 ++- crates/persisting-proto/src/lib.rs | 2 + crates/persisting-proto/src/runtime.rs | 605 ++++++++++++++++++ crates/persisting-pvisor/Cargo.toml | 23 + crates/persisting-pvisor/README.md | 56 ++ crates/persisting-pvisor/src/access.rs | 290 +++++++++ crates/persisting-pvisor/src/event.rs | 109 ++++ crates/persisting-pvisor/src/executor.rs | 88 +++ crates/persisting-pvisor/src/lib.rs | 19 + crates/persisting-pvisor/src/process.rs | 264 ++++++++ crates/persisting-pvisor/src/runtime.rs | 329 ++++++++++ justfile | 5 +- 22 files changed, 2169 insertions(+), 164 deletions(-) create mode 100644 crates/persisting-proto/src/runtime.rs create mode 100644 crates/persisting-pvisor/Cargo.toml create mode 100644 crates/persisting-pvisor/README.md create mode 100644 crates/persisting-pvisor/src/access.rs create mode 100644 crates/persisting-pvisor/src/event.rs create mode 100644 crates/persisting-pvisor/src/executor.rs create mode 100644 crates/persisting-pvisor/src/lib.rs create mode 100644 crates/persisting-pvisor/src/process.rs create mode 100644 crates/persisting-pvisor/src/runtime.rs diff --git a/Cargo.lock b/Cargo.lock index 76cde0d..07174b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5292,8 +5292,9 @@ dependencies = [ "http-body-util", "hyper", "hyper-util", - "ipnet", "lazy_static", + "persisting-proto", + "persisting-pvisor", "pulsing-actor", "regex", "reqwest 0.12.28", @@ -5431,6 +5432,22 @@ dependencies = [ "serde_json", ] +[[package]] +name = "persisting-pvisor" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "ipnet", + "libc", + "persisting-proto", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "uuid", +] + [[package]] name = "petgraph" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index aae774f..e11b3e8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/persisting-cli", "crates/persisting-dlcapt", "crates/persisting-compute", + "crates/persisting-pvisor", ] [workspace.package] @@ -33,4 +34,3 @@ codegen-units = 1 [workspace.dependencies] pulsing-actor = { version = "0.1.2", default-features = false } - diff --git a/crates/persisting-capture/Cargo.toml b/crates/persisting-capture/Cargo.toml index e3543e2..bb97e12 100644 --- a/crates/persisting-capture/Cargo.toml +++ b/crates/persisting-capture/Cargo.toml @@ -34,7 +34,8 @@ pulsing-actor = { workspace = true } dashmap = "6" blake3 = "1" hostname = "0.4" -ipnet = "2" +persisting-proto = { path = "../persisting-proto" } +persisting-pvisor = { path = "../persisting-pvisor" } [dev-dependencies] tempfile = "3" diff --git a/crates/persisting-capture/src/proxy/common.rs b/crates/persisting-capture/src/proxy/common.rs index da0a30b..a87f730 100644 --- a/crates/persisting-capture/src/proxy/common.rs +++ b/crates/persisting-capture/src/proxy/common.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use axum::http::HeaderMap; use bytes::Bytes; +use persisting_proto::ModelAccessPolicy; use serde_json::Value; use super::state::ProxyState; @@ -22,6 +23,30 @@ pub(crate) fn effective_config(state: &ProxyState, route: &CaptureRoute) -> Arc< .unwrap_or_else(|| Arc::clone(&state.config)) } +pub(crate) fn model_access_policy(config: &ProxyConfig) -> ModelAccessPolicy { + let allowed_models = config + .models + .iter() + .map(|route| route.name.clone()) + .collect(); + let providers: Vec = config + .models + .iter() + .filter_map(|route| route.provider.clone()) + .collect(); + // An inferred/custom provider must remain representable during migration. + // An empty provider list means model identity is enforced but provider is open. + let allowed_providers = if providers.len() == config.models.len() { + providers + } else { + Vec::new() + }; + ModelAccessPolicy { + allowed_models, + allowed_providers, + } +} + #[allow(clippy::too_many_arguments)] pub(crate) fn call_context( state: &ProxyState, diff --git a/crates/persisting-capture/src/proxy/dispatch.rs b/crates/persisting-capture/src/proxy/dispatch.rs index 9a5462e..62d2c95 100644 --- a/crates/persisting-capture/src/proxy/dispatch.rs +++ b/crates/persisting-capture/src/proxy/dispatch.rs @@ -10,14 +10,16 @@ use axum::routing::any; use axum::Router; use bytes::Bytes; use http_body_util::BodyExt; +use persisting_proto::{NetworkAccessRequest, NetworkTransport, RunId, StorylineId}; use super::common::effective_config; use super::forward::{ - handle_connect, is_forward_proxy_request, is_llm_capture_path, transparent_forward, + handle_connect_authorized, is_forward_proxy_request, is_llm_capture_path, + transparent_forward_authorized, }; use super::llm_capture::llm_capture; use super::network_policy::{ - assert_egress_allowed, forbidden_response, host_from_authority, NetworkPolicy, + authorize_egress, forbidden_response, host_from_authority, NetworkPolicy, }; use super::state::ProxyState; use crate::debug::{self, is_debug_enabled}; @@ -98,7 +100,18 @@ async fn dispatch_impl( .map(|a| a.to_string()) .unwrap_or_else(|| uri.clone()); let host = host_from_authority(&authority); - if let Err(reason) = assert_egress_allowed(&policy, &host) { + if let Err(reason) = authorize_egress( + state.access_controller.as_ref(), + &policy, + &NetworkAccessRequest { + run_id: log_route.root_session.clone().map(RunId), + attempt_id: None, + storyline_id: Some(StorylineId(log_route.session_id.clone())), + host: host.clone(), + port: req.uri().port_u16(), + transport: NetworkTransport::TcpTunnel, + }, + ) { return Ok(deny_egress( &state, &policy, @@ -111,13 +124,29 @@ async fn dispatch_impl( if debug_on { debug::log_connect(state.storage.as_path(), &authority, &session_id); } - return Ok(handle_connect(req, &policy).await); + return Ok(handle_connect_authorized(req).await); } let path = req.uri().path().to_string(); if is_forward_proxy_request(req.method(), req.uri()) { let host = req.uri().host().map(str::to_string).unwrap_or_default(); - if let Err(reason) = assert_egress_allowed(&policy, &host) { + let transport = if req.uri().scheme_str() == Some("https") { + NetworkTransport::Https + } else { + NetworkTransport::Http + }; + if let Err(reason) = authorize_egress( + state.access_controller.as_ref(), + &policy, + &NetworkAccessRequest { + run_id: log_route.root_session.clone().map(RunId), + attempt_id: None, + storyline_id: Some(StorylineId(log_route.session_id.clone())), + host: host.clone(), + port: req.uri().port_u16(), + transport, + }, + ) { return Ok(deny_egress( &state, &policy, @@ -148,7 +177,7 @@ async fn dispatch_impl( "forward", ); } - let resp = transparent_forward(&state.client, req, &policy).await?; + let resp = transparent_forward_authorized(&state.client, req).await?; if debug_on { let status = resp.status(); let headers = resp.headers().clone(); diff --git a/crates/persisting-capture/src/proxy/forward.rs b/crates/persisting-capture/src/proxy/forward.rs index 84f7518..fc14442 100644 --- a/crates/persisting-capture/src/proxy/forward.rs +++ b/crates/persisting-capture/src/proxy/forward.rs @@ -27,6 +27,14 @@ pub async fn handle_connect(req: Request, policy: &NetworkPolicy) -> Response { let (status, msg) = forbidden_response(&host, &reason); return (status, msg).into_response(); } + handle_connect_authorized(req).await +} + +/// Execute a CONNECT request after the caller's pVisor controller authorized it. +pub async fn handle_connect_authorized(req: Request) -> Response { + let Some(authority) = req.uri().authority().map(|a| a.to_string()) else { + return StatusCode::BAD_REQUEST.into_response(); + }; let target = connect_target(&authority); let on_upgrade: OnUpgrade = hyper::upgrade::on(req); tokio::spawn(async move { @@ -85,7 +93,15 @@ pub async fn transparent_forward( .body(Body::from(msg)) .expect("403 body")); } + transparent_forward_authorized(client, req).await +} +/// Forward an absolute-URI request after the caller's pVisor controller +/// authorized it. +pub async fn transparent_forward_authorized( + client: &reqwest::Client, + req: Request, +) -> anyhow::Result> { let (parts, body) = req.into_parts(); let url = parts.uri.to_string(); let body_bytes = body @@ -120,9 +136,9 @@ pub async fn transparent_forward( } builder = builder.header(name, value); } - Ok(builder + builder .body(Body::from(bytes)) - .map_err(|e| anyhow::anyhow!("build response: {e}"))?) + .map_err(|e| anyhow::anyhow!("build response: {e}")) } #[cfg(test)] diff --git a/crates/persisting-capture/src/proxy/llm_capture.rs b/crates/persisting-capture/src/proxy/llm_capture.rs index b8b62b4..3e5491a 100644 --- a/crates/persisting-capture/src/proxy/llm_capture.rs +++ b/crates/persisting-capture/src/proxy/llm_capture.rs @@ -8,11 +8,13 @@ use axum::extract::Request; use axum::http::{Method, StatusCode}; use axum::response::{IntoResponse, Response}; use http_body_util::BodyExt; +use persisting_proto::{ModelCallRequest, RunId, StorylineId}; use serde_json::Value; use super::auth::{apply_upstream_headers, resolve_upstream_api_key}; use super::common::{ attach_capture_headers, call_context, effective_config, extract_model, is_models_list_path, + model_access_policy, }; use super::http_headers::{is_websocket_upgrade, skip_response_header_when_body_changed}; use super::models_list::build_models_response; @@ -134,6 +136,42 @@ pub(super) async fn llm_capture( upstream_url.set_query(Some(q)); } + let access_decision = state.access_controller.authorize_model( + &model_access_policy(&cfg), + &ModelCallRequest { + run_id: capture_route.root_session.clone().map(RunId::new), + attempt_id: None, + storyline_id: Some(StorylineId::new(capture_route.session_id.clone())), + call_id: call.call_id.clone(), + client_model: client_model.clone(), + upstream_model: upstream_model.clone(), + provider: provider.as_str().to_string(), + protocol: protocol.as_str().to_string(), + upstream_host: upstream_url.host_str().unwrap_or_default().to_string(), + }, + ); + if !access_decision.is_allowed() { + tracing::warn!( + target: "persisting_capture", + run_id = capture_route.root_session.as_deref().unwrap_or("-"), + storyline_id = %capture_route.session_id, + call_id = %call.call_id, + client_model = %client_model, + upstream_model = %upstream_model, + provider = provider.as_str(), + reason = access_decision.reason.code(), + "pVisor denied model call" + ); + return Ok(( + StatusCode::FORBIDDEN, + format!( + "persisting-proxy: pVisor denied model `{client_model}` ({})", + access_decision.reason.code() + ), + ) + .into_response()); + } + if debug_on { let body_preview = truncate_body_bytes(&upstream_body); debug::log_llm_request( diff --git a/crates/persisting-capture/src/proxy/mod.rs b/crates/persisting-capture/src/proxy/mod.rs index 4756bb4..cbb3e75 100644 --- a/crates/persisting-capture/src/proxy/mod.rs +++ b/crates/persisting-capture/src/proxy/mod.rs @@ -19,5 +19,8 @@ pub mod upstream; pub use auth::{apply_upstream_headers, resolve_upstream_api_key}; pub use model::rewrite_model_in_body; pub use reasoning::ReasoningCacheHandle; -pub use state::{serve, serve_with_shutdown, serve_with_shutdown_and_ready, ProxyState}; +pub use state::{ + serve, serve_with_runtime_control, serve_with_shutdown, serve_with_shutdown_and_ready, + ProxyState, +}; pub use upstream::prepare_upstream_body; diff --git a/crates/persisting-capture/src/proxy/network_policy.rs b/crates/persisting-capture/src/proxy/network_policy.rs index a5fb55b..d7e11ee 100644 --- a/crates/persisting-capture/src/proxy/network_policy.rs +++ b/crates/persisting-capture/src/proxy/network_policy.rs @@ -4,11 +4,13 @@ //! Matching: exact hostname, leading `*.suffix`, IPv4/IPv6 literal, CIDR. //! Entries are not URLs, ports, or paths. -use std::net::IpAddr; -use std::str::FromStr; - use axum::http::StatusCode; -use ipnet::IpNet; +use persisting_proto::{AccessReason, NetworkAccessRequest, NetworkCapability, NetworkTransport}; +pub use persisting_pvisor::{ + host_matches, normalize_host, parse_network_rule as parse_allowed_entry, + NetworkRule as AllowedEntry, +}; +use persisting_pvisor::{AccessController, NetworkGuard, PolicyAccessController}; use crate::config::{NetworkConfig, NetworkMode, ProxyConfig}; @@ -20,14 +22,9 @@ pub struct NetworkPolicy { pub allowed: Vec, /// Host part of `listen` (always bypassed with other loopbacks). pub listen_host: String, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum AllowedEntry { - Exact(String), - WildcardSuffix(String), - Ip(IpAddr), - Cidr(IpNet), + /// Compiled pVisor policy. Capture owns protocol adaptation; pVisor owns + /// the allow/deny decision. + pub guard: NetworkGuard, } impl NetworkPolicy { @@ -41,14 +38,18 @@ impl NetworkPolicy { } } } - let mut allowed = Vec::with_capacity(raw.len()); - for entry in &raw { - allowed.push(parse_allowed_entry(entry)?); - } + let capability = match cfg.network.mode { + NetworkMode::Public => NetworkCapability::Ambient, + NetworkMode::NoNetwork => NetworkCapability::Deny, + NetworkMode::Allowlist => NetworkCapability::AllowList { hosts: raw }, + }; + let guard = NetworkGuard::compile(capability, [listen_host.clone()])?; + let allowed = guard.rules().to_vec(); Ok(Self { mode: cfg.network.mode, allowed, listen_host, + guard, }) } @@ -68,15 +69,6 @@ pub fn validate_network_config(network: &NetworkConfig) -> anyhow::Result<()> { Ok(()) } -/// Normalize host for comparison: trim, lowercase, strip trailing dot. -pub fn normalize_host(host: &str) -> String { - host.trim() - .trim_matches(|c| c == '[' || c == ']') - .to_ascii_lowercase() - .trim_end_matches('.') - .to_string() -} - /// Parse `listen` (`127.0.0.1:19081` or `http://127.0.0.1:19081`) into host. pub fn host_from_listen(listen: &str) -> String { let s = listen.trim(); @@ -103,114 +95,6 @@ pub fn host_from_authority(authority: &str) -> String { normalize_host(authority) } -pub fn parse_allowed_entry(raw: &str) -> anyhow::Result { - let entry = raw.trim(); - if entry.is_empty() { - anyhow::bail!("allowed_hosts entry must not be empty"); - } - if entry.contains("://") || entry.contains(']') || entry.contains('[') { - anyhow::bail!( - "allowed_hosts entry `{entry}` must be a hostname, `*.suffix`, IP, or CIDR \ - (not a URL or bracketed IPv6)" - ); - } - // Path-like (but allow CIDR `a.b.c.d/nn`). - if entry.contains('/') && IpNet::from_str(entry).is_err() { - anyhow::bail!( - "allowed_hosts entry `{entry}` must be a hostname, `*.suffix`, IP, or CIDR \ - (not a URL path)" - ); - } - // Port in entry is forbidden (Harbor semantics). - if let Some((host_part, maybe_port)) = entry.rsplit_once(':') { - if !host_part.is_empty() - && maybe_port.chars().all(|c| c.is_ascii_digit()) - && !entry.contains('/') - && host_part.parse::().is_err() - && !host_part.contains(':') - { - // hostname:port — reject - anyhow::bail!( - "allowed_hosts entry `{entry}` must not include a port (got hostname:port)" - ); - } - } - - if let Some(suffix) = entry.strip_prefix("*.") { - let suffix = normalize_host(suffix); - if suffix.is_empty() || suffix.contains('*') { - anyhow::bail!("invalid wildcard allowed_hosts entry `{entry}`"); - } - if suffix.parse::().is_ok() { - anyhow::bail!("wildcard allowed_hosts cannot wrap an IP (`{entry}`)"); - } - return Ok(AllowedEntry::WildcardSuffix(suffix)); - } - - if entry.contains('*') { - anyhow::bail!( - "allowed_hosts entry `{entry}`: only leading `*.suffix` wildcards are supported" - ); - } - - if let Ok(cidr) = IpNet::from_str(entry) { - // Prefer CIDR form when a prefix length is present. - if entry.contains('/') { - return Ok(AllowedEntry::Cidr(cidr)); - } - // `IpNet::from_str("1.1.1.1")` succeeds as /32 — treat as literal IP. - return Ok(AllowedEntry::Ip(cidr.addr())); - } - if let Ok(ip) = IpAddr::from_str(entry) { - return Ok(AllowedEntry::Ip(ip)); - } - - let host = normalize_host(entry); - if host.is_empty() || host.contains(':') { - anyhow::bail!("invalid allowed_hosts hostname `{entry}`"); - } - Ok(AllowedEntry::Exact(host)) -} - -pub fn host_matches(host: &str, allowed: &[AllowedEntry]) -> bool { - let host = normalize_host(host); - if host.is_empty() { - return false; - } - let host_ip = IpAddr::from_str(&host).ok(); - for entry in allowed { - match entry { - AllowedEntry::Exact(h) => { - if host == *h { - return true; - } - } - AllowedEntry::WildcardSuffix(suffix) => { - // `*.example.com` matches subdomains only, not apex `example.com`. - if host.ends_with(suffix) - && host.len() > suffix.len() - && host.as_bytes()[host.len() - suffix.len() - 1] == b'.' - { - return true; - } - } - AllowedEntry::Ip(ip) => { - if host_ip == Some(*ip) { - return true; - } - } - AllowedEntry::Cidr(net) => { - if let Some(ip) = host_ip { - if net.contains(&ip) { - return true; - } - } - } - } - } - false -} - pub fn is_loopback_host(host: &str, listen_host: &str) -> bool { let h = normalize_host(host); if h == "localhost" || h == "127.0.0.1" || h == "::1" || h == "0:0:0:0:0:0:0:1" { @@ -219,7 +103,7 @@ pub fn is_loopback_host(host: &str, listen_host: &str) -> bool { if !listen_host.is_empty() && h == normalize_host(listen_host) { return true; } - if let Ok(ip) = IpAddr::from_str(&h) { + if let Ok(ip) = h.parse::() { return ip.is_loopback(); } false @@ -244,23 +128,51 @@ impl DenyReason { /// Check whether forward-proxy egress to `host` is allowed. pub fn assert_egress_allowed(policy: &NetworkPolicy, host: &str) -> Result<(), DenyReason> { - if is_loopback_host(host, &policy.listen_host) { + assert_egress_allowed_with( + &PolicyAccessController, + policy, + host, + None, + NetworkTransport::TcpTunnel, + ) +} + +pub fn assert_egress_allowed_with( + controller: &dyn AccessController, + policy: &NetworkPolicy, + host: &str, + port: Option, + transport: NetworkTransport, +) -> Result<(), DenyReason> { + authorize_egress( + controller, + policy, + &NetworkAccessRequest { + run_id: None, + attempt_id: None, + storyline_id: None, + host: host.to_string(), + port, + transport, + }, + ) +} + +/// Authorize an identity-bearing egress request through pVisor. +pub fn authorize_egress( + controller: &dyn AccessController, + policy: &NetworkPolicy, + request: &NetworkAccessRequest, +) -> Result<(), DenyReason> { + let decision = controller.authorize_network(&policy.guard, request); + if decision.is_allowed() { return Ok(()); } - match policy.mode { - NetworkMode::Public => Ok(()), - NetworkMode::NoNetwork => Err(DenyReason::NoNetwork), - NetworkMode::Allowlist => { - if policy.allowed.is_empty() { - return Err(DenyReason::AllowlistEmpty); - } - if host_matches(host, &policy.allowed) { - Ok(()) - } else { - Err(DenyReason::NotInAllowlist) - } - } - } + Err(match decision.reason { + AccessReason::NetworkDenied => DenyReason::NoNetwork, + AccessReason::NetworkAllowListEmpty => DenyReason::AllowlistEmpty, + _ => DenyReason::NotInAllowlist, + }) } pub fn forbidden_response(host: &str, reason: &DenyReason) -> (StatusCode, String) { @@ -297,6 +209,34 @@ fn upstream_hosts_from_models(cfg: &ProxyConfig) -> Vec { mod tests { use super::*; use crate::config::{CaptureLevel, ModelRoute, NetworkConfig, NetworkMode, ProxyConfig}; + use persisting_proto::{ + AccessDecision, AccessReason, ModelAccessPolicy, ModelCallRequest, RunId, StorylineId, + }; + use std::sync::Mutex; + + #[derive(Default)] + struct RecordingDenyController { + network_request: Mutex>, + } + + impl AccessController for RecordingDenyController { + fn authorize_network( + &self, + _policy: &NetworkGuard, + request: &NetworkAccessRequest, + ) -> AccessDecision { + *self.network_request.lock().unwrap() = Some(request.clone()); + AccessDecision::deny(AccessReason::HostNotAllowed) + } + + fn authorize_model( + &self, + _policy: &ModelAccessPolicy, + _request: &ModelCallRequest, + ) -> AccessDecision { + AccessDecision::deny(AccessReason::ModelNotAllowed) + } + } fn cfg(mode: NetworkMode, allowed: &[&str], upstream: Option<&str>) -> ProxyConfig { ProxyConfig { @@ -373,6 +313,30 @@ mod tests { assert!(assert_egress_allowed(&p, "evil.example").is_ok()); } + #[test] + fn injected_pvisor_controller_is_authoritative_and_receives_identity() { + let policy = NetworkPolicy::from_config(&cfg(NetworkMode::Public, &[], None)).unwrap(); + let controller = RecordingDenyController::default(); + let request = NetworkAccessRequest { + run_id: Some(RunId::new("run-42")), + attempt_id: None, + storyline_id: Some(StorylineId::new("story-7")), + host: "api.example.com".into(), + port: Some(443), + transport: NetworkTransport::Https, + }; + + assert_eq!( + authorize_egress(&controller, &policy, &request), + Err(DenyReason::NotInAllowlist) + ); + let seen = controller.network_request.lock().unwrap(); + let seen = seen.as_ref().expect("controller saw request"); + assert_eq!(seen.run_id.as_ref().unwrap().as_str(), "run-42"); + assert_eq!(seen.storyline_id.as_ref().unwrap().as_str(), "story-7"); + assert_eq!(seen.host, "api.example.com"); + } + #[test] fn no_network_denies_non_loopback() { let p = NetworkPolicy::from_config(&cfg(NetworkMode::NoNetwork, &[], None)).unwrap(); diff --git a/crates/persisting-capture/src/proxy/state.rs b/crates/persisting-capture/src/proxy/state.rs index d75578f..ff44162 100644 --- a/crates/persisting-capture/src/proxy/state.rs +++ b/crates/persisting-capture/src/proxy/state.rs @@ -5,6 +5,7 @@ use std::sync::atomic::AtomicUsize; use std::sync::Arc; use std::time::Duration; +use persisting_pvisor::{AccessController, PolicyAccessController}; use tokio::task::JoinHandle; use super::admin::{admin_router, AdminState}; @@ -27,6 +28,7 @@ pub struct ProxyState { pub index: SessionIndexHandle, pub session_clients: Arc, pub reasoning_cache: Arc, + pub access_controller: Arc, pub active_requests: Arc, pub started_at: String, } @@ -65,6 +67,31 @@ pub async fn serve_with_shutdown_and_ready( stream_markdown: bool, ready: Option>, shutdown: impl std::future::Future + Send + 'static, +) -> anyhow::Result<()> { + serve_with_runtime_control( + config, + storage, + sink, + stream_markdown, + Arc::new(PolicyAccessController), + ready, + shutdown, + ) + .await +} + +/// Capture data plane with an injected pVisor access controller. +/// +/// pVisor owns model/network authorization; Capture retains HTTP protocol +/// adaptation, streaming, and trajectory extraction. +pub async fn serve_with_runtime_control( + config: ProxyConfig, + storage: impl AsRef, + sink: Arc, + stream_markdown: bool, + access_controller: Arc, + ready: Option>, + shutdown: impl std::future::Future + Send + 'static, ) -> anyhow::Result<()> { let (stop_tx, stop_rx) = tokio::sync::watch::channel(()); tokio::spawn(async move { @@ -108,6 +135,7 @@ pub async fn serve_with_shutdown_and_ready( index: index.clone(), session_clients: Arc::new(SessionClientRegistry::default()), reasoning_cache: Arc::new(ReasoningCacheHandle::new()), + access_controller, active_requests: Arc::clone(&active_requests), started_at: started_at.clone(), }; diff --git a/crates/persisting-capture/tests/network_policy_http.rs b/crates/persisting-capture/tests/network_policy_http.rs index 57dea86..dbb1c14 100644 --- a/crates/persisting-capture/tests/network_policy_http.rs +++ b/crates/persisting-capture/tests/network_policy_http.rs @@ -8,10 +8,34 @@ use axum::http::StatusCode; use axum::routing::{get, post}; use axum::Router; use persisting_capture::config::ProxyConfig; -use persisting_capture::proxy::serve_with_shutdown_and_ready; +use persisting_capture::proxy::{serve_with_runtime_control, serve_with_shutdown_and_ready}; use persisting_capture::sink::SeqOnlySink; +use persisting_proto::{ + AccessDecision, AccessReason, ModelAccessPolicy, ModelCallRequest, NetworkAccessRequest, +}; +use persisting_pvisor::{AccessController, NetworkGuard, PolicyAccessController}; use tokio::sync::oneshot; +struct DenyModelController; + +impl AccessController for DenyModelController { + fn authorize_network( + &self, + policy: &NetworkGuard, + request: &NetworkAccessRequest, + ) -> AccessDecision { + PolicyAccessController.authorize_network(policy, request) + } + + fn authorize_model( + &self, + _policy: &ModelAccessPolicy, + _request: &ModelCallRequest, + ) -> AccessDecision { + AccessDecision::deny(AccessReason::ModelNotAllowed) + } +} + fn free_port() -> u16 { std::net::TcpListener::bind("127.0.0.1:0") .unwrap() @@ -74,6 +98,39 @@ async fn spawn_proxy(toml: &str) -> (String, tempfile::TempDir, oneshot::Sender< (format!("http://127.0.0.1:{listen_port}"), tmp, stop_tx) } +async fn spawn_proxy_with_controller( + toml: &str, + controller: Arc, +) -> (String, tempfile::TempDir, oneshot::Sender<()>) { + let listen_port = free_port(); + let admin_port = free_port(); + let toml = toml + .replace("{{LISTEN}}", &format!("127.0.0.1:{listen_port}")) + .replace("{{ADMIN}}", &format!("127.0.0.1:{admin_port}")); + let cfg = ProxyConfig::from_toml_str(&toml).expect("proxy toml"); + let tmp = tempfile::tempdir().unwrap(); + let (ready_tx, ready_rx) = oneshot::channel(); + let (stop_tx, stop_rx) = oneshot::channel::<()>(); + let storage = tmp.path().to_path_buf(); + let sink: Arc = Arc::new(SeqOnlySink::new()); + tokio::spawn(async move { + let _ = serve_with_runtime_control( + cfg, + storage, + sink, + false, + controller, + Some(ready_tx), + async move { + let _ = stop_rx.await; + }, + ) + .await; + }); + ready_rx.await.expect("proxy ready"); + (format!("http://127.0.0.1:{listen_port}"), tmp, stop_tx) +} + async fn raw_connect(proxy: &str, authority: &str) -> (StatusCode, String) { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -290,6 +347,47 @@ upstream = "http://127.0.0.1:{mock_port}/v1" let _ = mock_stop.send(()); } +#[tokio::test] +async fn e2e_injected_pvisor_controller_denies_model_before_upstream() { + let (mock_port, mock_stop) = spawn_mock_http().await; + let toml = format!( + r#" +listen = "{{{{LISTEN}}}}" +admin_listen = "{{{{ADMIN}}}}" +agent_id = "t" + +[network] +mode = "public" + +[[models]] +name = "*" +upstream = "http://127.0.0.1:{mock_port}/v1" +"# + ); + let (proxy, _tmp, stop) = + spawn_proxy_with_controller(&toml, Arc::new(DenyModelController)).await; + + let client = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .build() + .unwrap(); + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("content-type", "application/json") + .body(r#"{"model":"test","messages":[{"role":"user","content":"hi"}]}"#) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + assert!( + resp.text().await.unwrap().contains("model-not-allowed"), + "denial should expose the stable pVisor reason" + ); + let _ = stop.send(()); + let _ = mock_stop.send(()); +} + #[tokio::test] async fn e2e_public_mode_allows_absolute_uri_forward() { let (mock_port, mock_stop) = spawn_mock_http().await; diff --git a/crates/persisting-proto/src/lib.rs b/crates/persisting-proto/src/lib.rs index ac084bb..d8d4b57 100644 --- a/crates/persisting-proto/src/lib.rs +++ b/crates/persisting-proto/src/lib.rs @@ -6,6 +6,7 @@ pub mod invoke_abi; mod messages; pub mod rpc_dispatch; +pub mod runtime; pub use invoke_abi::{ invoke_ron_utf8_via_jobs_sync, job_take_result_utf8_with_buffer, poll_status_label, @@ -27,6 +28,7 @@ pub use rpc_dispatch::{ application_error_response, dispatch_bincode_with, handle_rpc_request_with, malformed_request_response, version_mismatch_response, }; +pub use runtime::*; use anyhow::{Context, Result}; diff --git a/crates/persisting-proto/src/runtime.rs b/crates/persisting-proto/src/runtime.rs new file mode 100644 index 0000000..bc781c7 --- /dev/null +++ b/crates/persisting-proto/src/runtime.rs @@ -0,0 +1,605 @@ +//! Stable value types shared by pVisor, Compute, capture, and storage. +//! +//! The runtime and narrative dimensions are deliberately orthogonal: +//! +//! - one [`RunId`] may have multiple execution [`AttemptId`]s; +//! - events may additionally belong to a [`StorylineId`], turn, and call. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::BTreeMap; +use std::fmt; + +pub const RUNTIME_SCHEMA_VERSION: u32 = 1; +pub const EVENT_SCHEMA_VERSION: u32 = 2; + +macro_rules! string_id { + ($name:ident) => { + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] + #[serde(transparent)] + pub struct $name(pub String); + + impl $name { + pub fn new(value: impl Into) -> Self { + Self(value.into()) + } + + pub fn as_str(&self) -> &str { + &self.0 + } + + pub fn is_empty(&self) -> bool { + self.0.trim().is_empty() + } + } + + impl fmt::Display for $name { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } + } + + impl From for $name { + fn from(value: String) -> Self { + Self(value) + } + } + + impl From<&str> for $name { + fn from(value: &str) -> Self { + Self(value.to_string()) + } + } + }; +} + +string_id!(RunId); +string_id!(AttemptId); +string_id!(StorylineId); +string_id!(EventId); +string_id!(CheckpointId); + +/// A versioned logical Agent reference. It describes identity, not placement. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AgentRef { + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub adapter: Option, +} + +impl AgentRef { + pub fn new(name: impl Into) -> Self { + Self { + name: name.into(), + version: None, + adapter: None, + } + } +} + +/// One semantic Agent execution submitted to pVisor. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunSpec { + #[serde(default = "runtime_schema_version")] + pub schema_version: u32, + pub run_id: RunId, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub task_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_run_id: Option, + pub agent: AgentRef, + pub invocation: RunInvocation, + #[serde(default)] + pub input: Value, + #[serde(default)] + pub runtime: RuntimeConfig, + #[serde(default)] + pub capabilities: CapabilitySet, + #[serde(default)] + pub metadata: BTreeMap, +} + +impl RunSpec { + pub fn process( + run_id: impl Into, + agent: impl Into, + program: impl Into, + ) -> Self { + Self { + schema_version: RUNTIME_SCHEMA_VERSION, + run_id: run_id.into(), + task_id: None, + parent_run_id: None, + agent: AgentRef::new(agent), + invocation: RunInvocation::Process(ProcessInvocation::new(program)), + input: Value::Null, + runtime: RuntimeConfig::default(), + capabilities: CapabilitySet::default(), + metadata: BTreeMap::new(), + } + } +} + +fn runtime_schema_version() -> u32 { + RUNTIME_SCHEMA_VERSION +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum RunInvocation { + Process(ProcessInvocation), +} + +/// Local or container-hosted process invocation. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProcessInvocation { + pub program: String, + #[serde(default)] + pub args: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cwd: Option, + #[serde(default)] + pub env: BTreeMap, + #[serde(default = "default_true")] + pub inherit_env: bool, + #[serde(default)] + pub stdin: StdioMode, + #[serde(default)] + pub stdout: StdioMode, + #[serde(default)] + pub stderr: StdioMode, +} + +impl ProcessInvocation { + pub fn new(program: impl Into) -> Self { + Self { + program: program.into(), + args: Vec::new(), + cwd: None, + env: BTreeMap::new(), + inherit_env: true, + stdin: StdioMode::Inherit, + stdout: StdioMode::Inherit, + stderr: StdioMode::Inherit, + } + } +} + +fn default_true() -> bool { + true +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StdioMode { + #[default] + Inherit, + Capture, + Null, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RuntimeConfig { + /// Wall-clock limit for one Attempt. `None` means no pVisor deadline. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timeout_ms: Option, + /// Grace period between a cooperative process-tree termination request and + /// forced termination. + #[serde(default = "default_termination_grace_ms")] + pub termination_grace_ms: u64, + /// Maximum retained bytes for each captured stdout/stderr stream. + #[serde(default = "default_max_output_bytes")] + pub max_output_bytes: usize, + #[serde(default)] + pub policy_mode: PolicyMode, +} + +impl Default for RuntimeConfig { + fn default() -> Self { + Self { + timeout_ms: None, + termination_grace_ms: default_termination_grace_ms(), + max_output_bytes: default_max_output_bytes(), + policy_mode: PolicyMode::Audit, + } + } +} + +fn default_max_output_bytes() -> usize { + 1024 * 1024 +} + +fn default_termination_grace_ms() -> u64 { + 2_000 +} + +/// `Audit` records requested capabilities. `Enforce` requires an executor that +/// can actually prevent ambient access. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PolicyMode { + #[default] + Audit, + Enforce, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct CapabilitySet { + #[serde(default)] + pub models: Vec, + #[serde(default)] + pub tools: Vec, + #[serde(default)] + pub filesystem: Vec, + #[serde(default)] + pub network: NetworkCapability, + #[serde(default)] + pub secrets: Vec, + #[serde(default)] + pub allow_subprocess: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FilesystemCapability { + pub path: String, + pub access: FilesystemAccess, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FilesystemAccess { + Read, + ReadWrite, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "mode", rename_all = "snake_case")] +pub enum NetworkCapability { + /// Use the executor's ambient network. Only valid for audit/compatibility runs. + #[default] + Ambient, + Deny, + AllowList { + hosts: Vec, + }, +} + +/// Runtime-neutral network request presented to pVisor for authorization. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NetworkAccessRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub attempt_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub storyline_id: Option, + pub host: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub port: Option, + pub transport: NetworkTransport, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum NetworkTransport { + Http, + Https, + TcpTunnel, +} + +/// Model invocation metadata. Request/response bodies intentionally stay in +/// Capture; pVisor receives only the information needed for policy and audit. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ModelCallRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub attempt_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub storyline_id: Option, + pub call_id: String, + pub client_model: String, + pub upstream_model: String, + pub provider: String, + pub protocol: String, + pub upstream_host: String, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ModelAccessPolicy { + #[serde(default)] + pub allowed_models: Vec, + #[serde(default)] + pub allowed_providers: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AccessEffect { + Allow, + Deny, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AccessReason { + AmbientNetwork, + TrustedLocal, + NetworkAllowList, + NetworkDenied, + NetworkAllowListEmpty, + HostNotAllowed, + ModelAllowed, + ModelNotAllowed, + ProviderNotAllowed, +} + +impl AccessReason { + pub fn code(self) -> &'static str { + match self { + Self::AmbientNetwork => "ambient-network", + Self::TrustedLocal => "trusted-local", + Self::NetworkAllowList => "network-allowlist", + Self::NetworkDenied => "network-denied", + Self::NetworkAllowListEmpty => "network-allowlist-empty", + Self::HostNotAllowed => "host-not-allowed", + Self::ModelAllowed => "model-allowed", + Self::ModelNotAllowed => "model-not-allowed", + Self::ProviderNotAllowed => "provider-not-allowed", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AccessDecision { + pub effect: AccessEffect, + pub reason: AccessReason, +} + +impl AccessDecision { + pub fn allow(reason: AccessReason) -> Self { + Self { + effect: AccessEffect::Allow, + reason, + } + } + + pub fn deny(reason: AccessReason) -> Self { + Self { + effect: AccessEffect::Deny, + reason, + } + } + + pub fn is_allowed(&self) -> bool { + self.effect == AccessEffect::Allow + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunState { + Created, + Starting, + Running, + Checkpointing, + Suspended, + Cancelling, + Completed, + Failed, + Cancelled, +} + +impl RunState { + pub fn is_terminal(self) -> bool { + matches!(self, Self::Completed | Self::Failed | Self::Cancelled) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ExecutorKind { + Process, + Container, + Wasm, + Remote, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IsolationKind { + HostProcess, + Container, + Wasm, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ExecutorDescriptor { + pub name: String, + pub kind: ExecutorKind, + pub isolation: IsolationKind, + pub enforces_capabilities: bool, + pub supports_checkpoint: bool, + pub supports_migration: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AttemptInfo { + pub attempt_id: AttemptId, + pub number: u32, + pub executor: ExecutorDescriptor, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at_unix_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at_unix_ms: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunStatus { + pub run_id: RunId, + pub state: RunState, + pub attempt: AttemptInfo, + pub updated_at_unix_ms: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunFailureKind { + InvalidSpec, + Unsupported, + Spawn, + ProcessExit, + DeadlineExceeded, + Infrastructure, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunFailure { + pub kind: RunFailureKind, + pub message: String, + #[serde(default)] + pub retryable: bool, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ProcessOutput { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stdout: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stderr: Option, + #[serde(default)] + pub stdout_truncated: bool, + #[serde(default)] + pub stderr_truncated: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ArtifactRef { + pub name: String, + pub uri: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub media_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub digest: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunResult { + pub run_id: RunId, + pub attempt_id: AttemptId, + pub state: RunState, + pub started_at_unix_ms: u64, + pub finished_at_unix_ms: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure: Option, + #[serde(default)] + pub output: ProcessOutput, + #[serde(default)] + pub metrics: BTreeMap, + #[serde(default)] + pub artifacts: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event_stream_ref: Option, + #[serde(default)] + pub warnings: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CheckpointKind { + Logical, + ExecutorSnapshot, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CheckpointRef { + pub checkpoint_id: CheckpointId, + pub run_id: RunId, + pub attempt_id: AttemptId, + pub kind: CheckpointKind, + pub uri: String, + pub event_seq: u64, + pub created_at_unix_ms: u64, +} + +/// Canonical event envelope. Provider-specific data belongs in `payload`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventEnvelope { + #[serde(default = "event_schema_version")] + pub schema_version: u32, + pub event_id: EventId, + pub run_id: RunId, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub attempt_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub storyline_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub turn_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub call_id: Option, + /// Monotonic within `(run_id, attempt_id, storyline_id, producer)`. + /// Runtime lifecycle events have no storyline and therefore use Attempt scope. + pub seq: u64, + pub timestamp_unix_ms: u64, + pub kind: String, + pub source: String, + pub producer: String, + #[serde(default)] + pub payload: Value, +} + +fn event_schema_version() -> u32 { + EVENT_SCHEMA_VERSION +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn process_run_spec_json_roundtrips() { + let mut spec = RunSpec::process("run-1", "codex", "codex"); + let RunInvocation::Process(process) = &mut spec.invocation; + process.args = vec!["exec".into(), "review".into()]; + process.stdout = StdioMode::Capture; + spec.capabilities.network = NetworkCapability::AllowList { + hosts: vec!["api.openai.com".into()], + }; + + let json = serde_json::to_string(&spec).unwrap(); + let decoded: RunSpec = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.run_id.as_str(), "run-1"); + assert_eq!(decoded.schema_version, RUNTIME_SCHEMA_VERSION); + assert!(matches!( + decoded.capabilities.network, + NetworkCapability::AllowList { .. } + )); + } + + #[test] + fn attempt_is_orthogonal_to_storyline() { + let event = EventEnvelope { + schema_version: EVENT_SCHEMA_VERSION, + event_id: EventId::from("event-1"), + run_id: RunId::from("run-1"), + attempt_id: Some(AttemptId::from("attempt-1")), + storyline_id: Some(StorylineId::from("main")), + turn_id: Some("turn-0".into()), + call_id: Some("call-0".into()), + seq: 0, + timestamp_unix_ms: 1, + kind: "llm.request".into(), + source: "model".into(), + producer: "pvisor".into(), + payload: Value::Null, + }; + assert_eq!(event.attempt_id.unwrap().as_str(), "attempt-1"); + assert_eq!(event.storyline_id.unwrap().as_str(), "main"); + } +} diff --git a/crates/persisting-pvisor/Cargo.toml b/crates/persisting-pvisor/Cargo.toml new file mode 100644 index 0000000..39f4331 --- /dev/null +++ b/crates/persisting-pvisor/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "persisting-pvisor" +version.workspace = true +edition.workspace = true +authors.workspace = true +license.workspace = true +description = "pVisor: Portable Agent Execution Runtime" +readme = "README.md" + +[dependencies] +anyhow = "1" +async-trait = "0.1" +ipnet = "2" +libc = "0.2" +persisting-proto = { path = "../persisting-proto" } +serde_json = "1" +thiserror = "2" +tokio = { version = "1", features = ["io-util", "macros", "process", "rt", "sync", "time"] } +tokio-util = { version = "0.7", features = ["rt"] } +uuid = { version = "1", features = ["v4"] } + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } diff --git a/crates/persisting-pvisor/README.md b/crates/persisting-pvisor/README.md new file mode 100644 index 0000000..3c14949 --- /dev/null +++ b/crates/persisting-pvisor/README.md @@ -0,0 +1,56 @@ +# pVisor + +**Portable Agent Execution Runtime** — the Persisting component that owns one +Agent Run execution. + +```text +RunSpec → PVisor::submit → RunHandle → RunResult + ├── status + ├── events + └── cancel +``` + +Version 1 includes: + +- stable Run, Attempt, capability, checkpoint, result, and event value types in + `persisting-proto`; +- model-call and network-access request/decision contracts; +- an injectable access controller used by `persisting-capture`; +- a local process executor; +- Run lifecycle status and events; +- cancellation and wall-clock deadlines; +- bounded stdout/stderr capture; +- an event sink interface for the canonical trajectory store. + +The local process executor is a compatibility executor. It reports +`HostProcess` isolation and supports audit-mode capabilities only. It does not +claim filesystem, network, syscall, checkpoint, or migration enforcement. +Those guarantees require later container or WASM executors behind the same +`RunExecutor` contract. + +Batch expansion and fleet scheduling are intentionally outside this crate and +remain responsibilities of `persisting-compute`. + +## Capture integration + +Capture remains the HTTP/streaming compatibility data plane. It resolves model +routes, adapts protocols, forwards bytes, and extracts trajectory events. Before +opening a general network connection or sending an LLM request, it presents an +identity-bearing request to pVisor's `AccessController`. + +```text +Agent request + │ + ▼ +Capture: decode + route + │ + ├── NetworkAccessRequest ──► pVisor policy ──► allow / deny + └── ModelCallRequest ──► pVisor policy ──► allow / deny + │ + ▼ allow +Capture: forward + stream + record trajectory +``` + +The default controller preserves existing Capture configuration semantics. +`serve_with_runtime_control` allows a Run-aware controller to be injected +without replacing Capture's protocol implementation. diff --git a/crates/persisting-pvisor/src/access.rs b/crates/persisting-pvisor/src/access.rs new file mode 100644 index 0000000..df2a077 --- /dev/null +++ b/crates/persisting-pvisor/src/access.rs @@ -0,0 +1,290 @@ +//! pVisor policy decisions for resources reached through compatibility proxies. + +use ipnet::IpNet; +use persisting_proto::{ + AccessDecision, AccessReason, ModelAccessPolicy, ModelCallRequest, NetworkAccessRequest, + NetworkCapability, +}; +use std::net::IpAddr; +use std::str::FromStr; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NetworkRule { + Exact(String), + WildcardSuffix(String), + Ip(IpAddr), + Cidr(IpNet), +} + +#[derive(Debug, Clone)] +pub struct NetworkGuard { + capability: NetworkCapability, + rules: Vec, + trusted_hosts: Vec, +} + +impl NetworkGuard { + pub fn compile( + capability: NetworkCapability, + trusted_hosts: impl IntoIterator, + ) -> anyhow::Result { + let rules = match &capability { + NetworkCapability::AllowList { hosts } => hosts + .iter() + .map(|host| parse_network_rule(host)) + .collect::>>()?, + _ => Vec::new(), + }; + Ok(Self { + capability, + rules, + trusted_hosts: trusted_hosts + .into_iter() + .map(|host| normalize_host(&host)) + .collect(), + }) + } + + pub fn capability(&self) -> &NetworkCapability { + &self.capability + } + + pub fn rules(&self) -> &[NetworkRule] { + &self.rules + } + + fn is_trusted(&self, host: &str) -> bool { + let host = normalize_host(host); + if matches!( + host.as_str(), + "localhost" | "127.0.0.1" | "::1" | "0:0:0:0:0:0:0:1" + ) { + return true; + } + if host.parse::().is_ok_and(|ip| ip.is_loopback()) { + return true; + } + self.trusted_hosts.iter().any(|trusted| trusted == &host) + } +} + +pub trait AccessController: Send + Sync { + fn authorize_network( + &self, + policy: &NetworkGuard, + request: &NetworkAccessRequest, + ) -> AccessDecision; + + fn authorize_model( + &self, + policy: &ModelAccessPolicy, + request: &ModelCallRequest, + ) -> AccessDecision; +} + +#[derive(Debug, Default)] +pub struct PolicyAccessController; + +impl AccessController for PolicyAccessController { + fn authorize_network( + &self, + policy: &NetworkGuard, + request: &NetworkAccessRequest, + ) -> AccessDecision { + if policy.is_trusted(&request.host) { + return AccessDecision::allow(AccessReason::TrustedLocal); + } + match policy.capability() { + NetworkCapability::Ambient => AccessDecision::allow(AccessReason::AmbientNetwork), + NetworkCapability::Deny => AccessDecision::deny(AccessReason::NetworkDenied), + NetworkCapability::AllowList { .. } if policy.rules().is_empty() => { + AccessDecision::deny(AccessReason::NetworkAllowListEmpty) + } + NetworkCapability::AllowList { .. } if host_matches(&request.host, policy.rules()) => { + AccessDecision::allow(AccessReason::NetworkAllowList) + } + NetworkCapability::AllowList { .. } => { + AccessDecision::deny(AccessReason::HostNotAllowed) + } + } + } + + fn authorize_model( + &self, + policy: &ModelAccessPolicy, + request: &ModelCallRequest, + ) -> AccessDecision { + let model_allowed = policy.allowed_models.iter().any(|pattern| { + model_matches(pattern, &request.client_model) + || model_matches(pattern, &request.upstream_model) + }); + if !model_allowed { + return AccessDecision::deny(AccessReason::ModelNotAllowed); + } + if !policy.allowed_providers.is_empty() + && !policy + .allowed_providers + .iter() + .any(|provider| provider.eq_ignore_ascii_case(&request.provider)) + { + return AccessDecision::deny(AccessReason::ProviderNotAllowed); + } + AccessDecision::allow(AccessReason::ModelAllowed) + } +} + +pub fn normalize_host(host: &str) -> String { + host.trim() + .trim_matches(|character| character == '[' || character == ']') + .to_ascii_lowercase() + .trim_end_matches('.') + .to_string() +} + +pub fn parse_network_rule(raw: &str) -> anyhow::Result { + let entry = raw.trim(); + if entry.is_empty() { + anyhow::bail!("network allowlist entry must not be empty"); + } + if entry.contains("://") || entry.contains(']') || entry.contains('[') { + anyhow::bail!( + "network allowlist entry `{entry}` must be a hostname, `*.suffix`, IP, or CIDR" + ); + } + if entry.contains('/') && IpNet::from_str(entry).is_err() { + anyhow::bail!( + "network allowlist entry `{entry}` must be a hostname, `*.suffix`, IP, or CIDR" + ); + } + if let Some((host, port)) = entry.rsplit_once(':') { + if !host.is_empty() + && port.chars().all(|character| character.is_ascii_digit()) + && !entry.contains('/') + && host.parse::().is_err() + && !host.contains(':') + { + anyhow::bail!("network allowlist entry `{entry}` must not include a port"); + } + } + if let Some(suffix) = entry.strip_prefix("*.") { + let suffix = normalize_host(suffix); + if suffix.is_empty() || suffix.contains('*') || suffix.parse::().is_ok() { + anyhow::bail!("invalid wildcard network allowlist entry `{entry}`"); + } + return Ok(NetworkRule::WildcardSuffix(suffix)); + } + if entry.contains('*') { + anyhow::bail!("only leading `*.suffix` host wildcards are supported"); + } + if let Ok(network) = IpNet::from_str(entry) { + if entry.contains('/') { + return Ok(NetworkRule::Cidr(network)); + } + return Ok(NetworkRule::Ip(network.addr())); + } + if let Ok(ip) = IpAddr::from_str(entry) { + return Ok(NetworkRule::Ip(ip)); + } + let host = normalize_host(entry); + if host.is_empty() || host.contains(':') { + anyhow::bail!("invalid network allowlist hostname `{entry}`"); + } + Ok(NetworkRule::Exact(host)) +} + +pub fn host_matches(host: &str, rules: &[NetworkRule]) -> bool { + let host = normalize_host(host); + let host_ip = IpAddr::from_str(&host).ok(); + rules.iter().any(|rule| match rule { + NetworkRule::Exact(allowed) => host == *allowed, + NetworkRule::WildcardSuffix(suffix) => { + host.ends_with(suffix) + && host.len() > suffix.len() + && host.as_bytes()[host.len() - suffix.len() - 1] == b'.' + } + NetworkRule::Ip(allowed) => host_ip == Some(*allowed), + NetworkRule::Cidr(network) => host_ip.is_some_and(|ip| network.contains(&ip)), + }) +} + +fn model_matches(pattern: &str, model: &str) -> bool { + if pattern == "*" { + return true; + } + if let Some(prefix) = pattern.strip_suffix('*') { + return !prefix.is_empty() && model.starts_with(prefix); + } + if let Some(suffix) = pattern.strip_prefix('*') { + return !suffix.is_empty() && model.ends_with(suffix); + } + pattern == model +} + +#[cfg(test)] +mod tests { + use super::*; + use persisting_proto::NetworkTransport; + + fn request(host: &str) -> NetworkAccessRequest { + NetworkAccessRequest { + run_id: None, + attempt_id: None, + storyline_id: None, + host: host.into(), + port: Some(443), + transport: NetworkTransport::TcpTunnel, + } + } + + #[test] + fn network_policy_handles_trusted_deny_and_allowlist() { + let controller = PolicyAccessController; + let deny = NetworkGuard::compile(NetworkCapability::Deny, ["proxy.local".into()]).unwrap(); + assert!(controller + .authorize_network(&deny, &request("127.0.0.1")) + .is_allowed()); + assert!(!controller + .authorize_network(&deny, &request("example.com")) + .is_allowed()); + + let allow = NetworkGuard::compile( + NetworkCapability::AllowList { + hosts: vec!["*.example.com".into(), "10.0.0.0/8".into()], + }, + Vec::new(), + ) + .unwrap(); + assert!(controller + .authorize_network(&allow, &request("api.example.com")) + .is_allowed()); + assert!(controller + .authorize_network(&allow, &request("10.1.2.3")) + .is_allowed()); + assert!(!controller + .authorize_network(&allow, &request("example.com")) + .is_allowed()); + } + + #[test] + fn model_policy_checks_model_and_provider() { + let controller = PolicyAccessController; + let policy = ModelAccessPolicy { + allowed_models: vec!["claude-*".into()], + allowed_providers: vec!["anthropic".into()], + }; + let mut request = ModelCallRequest { + run_id: None, + attempt_id: None, + storyline_id: None, + call_id: "call-1".into(), + client_model: "claude-sonnet".into(), + upstream_model: "claude-sonnet".into(), + provider: "anthropic".into(), + protocol: "messages".into(), + upstream_host: "api.anthropic.com".into(), + }; + assert!(controller.authorize_model(&policy, &request).is_allowed()); + request.provider = "custom".into(); + assert!(!controller.authorize_model(&policy, &request).is_allowed()); + } +} diff --git a/crates/persisting-pvisor/src/event.rs b/crates/persisting-pvisor/src/event.rs new file mode 100644 index 0000000..339a724 --- /dev/null +++ b/crates/persisting-pvisor/src/event.rs @@ -0,0 +1,109 @@ +use anyhow::Result; +use async_trait::async_trait; +use persisting_proto::{AttemptId, EventEnvelope, EventId, RunId, EVENT_SCHEMA_VERSION}; +use serde_json::Value; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use tokio::sync::broadcast; + +#[async_trait] +pub trait EventSink: Send + Sync { + async fn append(&self, event: &EventEnvelope) -> Result<()>; +} + +#[derive(Debug, Default)] +pub struct NoopEventSink; + +#[async_trait] +impl EventSink for NoopEventSink { + async fn append(&self, _event: &EventEnvelope) -> Result<()> { + Ok(()) + } +} + +/// In-memory sink intended for embedding, tests, and early integrations. +#[derive(Debug, Default)] +pub struct MemoryEventSink { + events: Mutex>, +} + +impl MemoryEventSink { + pub fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } +} + +#[async_trait] +impl EventSink for MemoryEventSink { + async fn append(&self, event: &EventEnvelope) -> Result<()> { + self.events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(event.clone()); + Ok(()) + } +} + +/// Assigns one monotonic event sequence to an Attempt and fans events out to +/// the canonical sink plus live subscribers. +#[derive(Clone)] +pub struct RunEventPublisher { + run_id: RunId, + attempt_id: AttemptId, + producer: String, + next_seq: Arc, + sink: Arc, + live: broadcast::Sender, +} + +impl RunEventPublisher { + pub(crate) fn new( + run_id: RunId, + attempt_id: AttemptId, + producer: impl Into, + sink: Arc, + live: broadcast::Sender, + ) -> Self { + Self { + run_id, + attempt_id, + producer: producer.into(), + next_seq: Arc::new(AtomicU64::new(0)), + sink, + live, + } + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.live.subscribe() + } + + pub async fn publish( + &self, + kind: impl Into, + source: impl Into, + payload: Value, + ) -> Result { + let event = EventEnvelope { + schema_version: EVENT_SCHEMA_VERSION, + event_id: EventId::new(format!("event-{}", uuid::Uuid::new_v4())), + run_id: self.run_id.clone(), + attempt_id: Some(self.attempt_id.clone()), + storyline_id: None, + turn_id: None, + call_id: None, + seq: self.next_seq.fetch_add(1, Ordering::AcqRel), + timestamp_unix_ms: crate::runtime::unix_now_ms(), + kind: kind.into(), + source: source.into(), + producer: self.producer.clone(), + payload, + }; + let _ = self.live.send(event.clone()); + self.sink.append(&event).await?; + Ok(event) + } +} diff --git a/crates/persisting-pvisor/src/executor.rs b/crates/persisting-pvisor/src/executor.rs new file mode 100644 index 0000000..47f2d78 --- /dev/null +++ b/crates/persisting-pvisor/src/executor.rs @@ -0,0 +1,88 @@ +use crate::event::RunEventPublisher; +use async_trait::async_trait; +use persisting_proto::{ + AttemptId, ExecutorDescriptor, RunInvocation, RunResult, RunSpec, RunState, RunStatus, +}; +use serde_json::json; +use std::sync::Arc; +use tokio::sync::watch; +use tokio_util::sync::CancellationToken; + +#[derive(Clone)] +pub struct AttemptContext { + spec: Arc, + attempt_id: AttemptId, + cancel: CancellationToken, + status: watch::Sender, + events: RunEventPublisher, +} + +impl AttemptContext { + pub(crate) fn new( + spec: Arc, + attempt_id: AttemptId, + cancel: CancellationToken, + status: watch::Sender, + events: RunEventPublisher, + ) -> Self { + Self { + spec, + attempt_id, + cancel, + status, + events, + } + } + + pub fn spec(&self) -> &RunSpec { + &self.spec + } + + pub fn attempt_id(&self) -> &AttemptId { + &self.attempt_id + } + + pub fn cancellation(&self) -> CancellationToken { + self.cancel.clone() + } + + pub fn events(&self) -> &RunEventPublisher { + &self.events + } + + pub async fn transition(&self, state: RunState, message: impl Into>) { + let now = crate::runtime::unix_now_ms(); + let message = message.into(); + self.status.send_modify(|status| { + status.state = state; + status.updated_at_unix_ms = now; + status.message = message.clone(); + if matches!(state, RunState::Starting | RunState::Running) + && status.attempt.started_at_unix_ms.is_none() + { + status.attempt.started_at_unix_ms = Some(now); + } + if state.is_terminal() { + status.attempt.finished_at_unix_ms = Some(now); + } + }); + let _ = self + .events + .publish( + "run.state_changed", + "runtime", + json!({ + "state": state, + "message": message, + }), + ) + .await; + } +} + +#[async_trait] +pub trait RunExecutor: Send + Sync { + fn descriptor(&self) -> ExecutorDescriptor; + fn supports(&self, invocation: &RunInvocation) -> bool; + async fn execute(&self, context: AttemptContext) -> RunResult; +} diff --git a/crates/persisting-pvisor/src/lib.rs b/crates/persisting-pvisor/src/lib.rs new file mode 100644 index 0000000..22eb2dc --- /dev/null +++ b/crates/persisting-pvisor/src/lib.rs @@ -0,0 +1,19 @@ +//! pVisor — Portable Agent Execution Runtime. +//! +//! pVisor owns the execution of one [`persisting_proto::RunSpec`]. Batch +//! expansion, fleet scheduling, and result collection remain Compute concerns. + +mod access; +mod event; +mod executor; +mod process; +mod runtime; + +pub use access::{ + host_matches, normalize_host, parse_network_rule, AccessController, NetworkGuard, NetworkRule, + PolicyAccessController, +}; +pub use event::{EventSink, MemoryEventSink, NoopEventSink, RunEventPublisher}; +pub use executor::{AttemptContext, RunExecutor}; +pub use process::ProcessExecutor; +pub use runtime::{PVisor, PVisorError, RunEventStream, RunHandle}; diff --git a/crates/persisting-pvisor/src/process.rs b/crates/persisting-pvisor/src/process.rs new file mode 100644 index 0000000..802fde7 --- /dev/null +++ b/crates/persisting-pvisor/src/process.rs @@ -0,0 +1,264 @@ +use crate::executor::{AttemptContext, RunExecutor}; +use async_trait::async_trait; +use persisting_proto::{ + ExecutorDescriptor, ExecutorKind, IsolationKind, ProcessInvocation, ProcessOutput, RunFailure, + RunFailureKind, RunInvocation, RunResult, RunState, StdioMode, +}; +use std::process::Stdio; +use tokio::io::{AsyncRead, AsyncReadExt}; +use tokio::process::{Child, Command}; + +#[derive(Debug, Default)] +pub struct ProcessExecutor; + +#[derive(Debug)] +struct Captured { + text: String, + truncated: bool, +} + +async fn read_limited( + mut reader: R, + limit: usize, +) -> std::io::Result { + let mut retained = Vec::with_capacity(limit.min(8192)); + let mut buf = [0_u8; 8192]; + let mut truncated = false; + loop { + let read = reader.read(&mut buf).await?; + if read == 0 { + break; + } + let remaining = limit.saturating_sub(retained.len()); + let keep = remaining.min(read); + retained.extend_from_slice(&buf[..keep]); + truncated |= keep < read; + } + Ok(Captured { + text: String::from_utf8_lossy(&retained).into_owned(), + truncated, + }) +} + +fn stdio(mode: StdioMode) -> Stdio { + match mode { + StdioMode::Inherit => Stdio::inherit(), + StdioMode::Capture => Stdio::piped(), + StdioMode::Null => Stdio::null(), + } +} + +impl ProcessExecutor { + fn spawn_command(invocation: &ProcessInvocation) -> Command { + let mut command = Command::new(&invocation.program); + command + .args(&invocation.args) + .stdin(stdio(invocation.stdin)) + .stdout(stdio(invocation.stdout)) + .stderr(stdio(invocation.stderr)) + .kill_on_drop(true); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + command.as_std_mut().process_group(0); + } + if let Some(cwd) = &invocation.cwd { + command.current_dir(cwd); + } + if !invocation.inherit_env { + command.env_clear(); + } + command.envs(&invocation.env); + command + } +} + +async fn terminate_process_tree(child: &mut Child, grace_ms: u64) { + #[cfg(unix)] + { + if let Some(pid) = child.id() { + // The child is the leader of the process group configured above. + let process_group = -(pid as i32); + unsafe { + libc::kill(process_group, libc::SIGTERM); + } + if tokio::time::timeout(std::time::Duration::from_millis(grace_ms), child.wait()) + .await + .is_ok() + { + return; + } + unsafe { + libc::kill(process_group, libc::SIGKILL); + } + let _ = child.wait().await; + return; + } + } + + let _ = child.kill().await; + let _ = child.wait().await; +} + +#[async_trait] +impl RunExecutor for ProcessExecutor { + fn descriptor(&self) -> ExecutorDescriptor { + ExecutorDescriptor { + name: "local-process-v1".into(), + kind: ExecutorKind::Process, + isolation: IsolationKind::HostProcess, + enforces_capabilities: false, + supports_checkpoint: false, + supports_migration: false, + } + } + + fn supports(&self, invocation: &RunInvocation) -> bool { + matches!(invocation, RunInvocation::Process(_)) + } + + async fn execute(&self, context: AttemptContext) -> RunResult { + let spec = context.spec().clone(); + let RunInvocation::Process(invocation) = &spec.invocation; + let started_at = crate::runtime::unix_now_ms(); + context + .transition(RunState::Starting, Some("spawning local process".into())) + .await; + + let mut child = match Self::spawn_command(invocation).spawn() { + Ok(child) => child, + Err(error) => { + return RunResult { + run_id: spec.run_id, + attempt_id: context.attempt_id().clone(), + state: RunState::Failed, + started_at_unix_ms: started_at, + finished_at_unix_ms: crate::runtime::unix_now_ms(), + exit_code: None, + failure: Some(RunFailure { + kind: RunFailureKind::Spawn, + message: error.to_string(), + retryable: false, + }), + output: ProcessOutput::default(), + metrics: Default::default(), + artifacts: Vec::new(), + event_stream_ref: None, + warnings: Vec::new(), + }; + } + }; + + let stdout_task = child.stdout.take().map(|stdout| { + let limit = spec.runtime.max_output_bytes; + tokio::spawn(async move { read_limited(stdout, limit).await }) + }); + let stderr_task = child.stderr.take().map(|stderr| { + let limit = spec.runtime.max_output_bytes; + tokio::spawn(async move { read_limited(stderr, limit).await }) + }); + + context.transition(RunState::Running, None).await; + + enum End { + Exited(std::io::Result), + Cancelled, + Deadline, + } + + let cancellation = context.cancellation(); + let end = if let Some(timeout_ms) = spec.runtime.timeout_ms { + tokio::select! { + biased; + status = child.wait() => End::Exited(status), + _ = cancellation.cancelled() => End::Cancelled, + _ = tokio::time::sleep(std::time::Duration::from_millis(timeout_ms)) => End::Deadline, + } + } else { + tokio::select! { + biased; + status = child.wait() => End::Exited(status), + _ = cancellation.cancelled() => End::Cancelled, + } + }; + + if matches!(end, End::Cancelled | End::Deadline) { + if matches!(end, End::Cancelled) { + context + .transition(RunState::Cancelling, Some("cancellation requested".into())) + .await; + } + terminate_process_tree(&mut child, spec.runtime.termination_grace_ms).await; + } + + let mut output = ProcessOutput::default(); + if let Some(task) = stdout_task { + if let Ok(Ok(captured)) = task.await { + output.stdout = Some(captured.text); + output.stdout_truncated = captured.truncated; + } + } + if let Some(task) = stderr_task { + if let Ok(Ok(captured)) = task.await { + output.stderr = Some(captured.text); + output.stderr_truncated = captured.truncated; + } + } + + let finished_at = crate::runtime::unix_now_ms(); + let (state, exit_code, failure) = match end { + End::Exited(Ok(status)) if status.success() => { + (RunState::Completed, status.code(), None) + } + End::Exited(Ok(status)) => ( + RunState::Failed, + status.code(), + Some(RunFailure { + kind: RunFailureKind::ProcessExit, + message: match status.code() { + Some(code) => format!("process exited with code {code}"), + None => "process terminated without an exit code".into(), + }, + retryable: false, + }), + ), + End::Exited(Err(error)) => ( + RunState::Failed, + None, + Some(RunFailure { + kind: RunFailureKind::Infrastructure, + message: error.to_string(), + retryable: true, + }), + ), + End::Cancelled => (RunState::Cancelled, None, None), + End::Deadline => ( + RunState::Failed, + None, + Some(RunFailure { + kind: RunFailureKind::DeadlineExceeded, + message: format!( + "attempt exceeded {} ms deadline", + spec.runtime.timeout_ms.unwrap_or_default() + ), + retryable: false, + }), + ), + }; + + RunResult { + run_id: spec.run_id, + attempt_id: context.attempt_id().clone(), + state, + started_at_unix_ms: started_at, + finished_at_unix_ms: finished_at, + exit_code, + failure, + output, + metrics: Default::default(), + artifacts: Vec::new(), + event_stream_ref: None, + warnings: Vec::new(), + } + } +} diff --git a/crates/persisting-pvisor/src/runtime.rs b/crates/persisting-pvisor/src/runtime.rs new file mode 100644 index 0000000..c622cd3 --- /dev/null +++ b/crates/persisting-pvisor/src/runtime.rs @@ -0,0 +1,329 @@ +use crate::event::{EventSink, NoopEventSink, RunEventPublisher}; +use crate::executor::{AttemptContext, RunExecutor}; +use crate::process::ProcessExecutor; +use persisting_proto::{ + AttemptId, AttemptInfo, EventEnvelope, PolicyMode, RunResult, RunSpec, RunState, RunStatus, + RUNTIME_SCHEMA_VERSION, +}; +use serde_json::json; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::sync::{broadcast, watch}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +#[derive(Debug, thiserror::Error)] +pub enum PVisorError { + #[error("invalid RunSpec: {0}")] + InvalidSpec(String), + #[error("no executor supports this invocation")] + UnsupportedInvocation, + #[error("executor `{0}` cannot enforce the requested capability policy")] + UnsupportedPolicy(String), + #[error("event sink rejected run creation: {0}")] + EventSink(#[source] anyhow::Error), + #[error("run task failed to join: {0}")] + Join(#[from] tokio::task::JoinError), +} + +pub fn unix_now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .unwrap_or(0) +} + +pub type RunEventStream = broadcast::Receiver; + +pub struct RunHandle { + run_id: persisting_proto::RunId, + attempt_id: AttemptId, + status: watch::Receiver, + cancellation: CancellationToken, + events: RunEventPublisher, + join: JoinHandle, +} + +impl RunHandle { + pub fn run_id(&self) -> &persisting_proto::RunId { + &self.run_id + } + + pub fn attempt_id(&self) -> &AttemptId { + &self.attempt_id + } + + pub fn status(&self) -> RunStatus { + self.status.borrow().clone() + } + + pub async fn status_changed(&mut self) -> Option { + self.status.changed().await.ok()?; + Some(self.status()) + } + + pub fn subscribe_events(&self) -> RunEventStream { + self.events.subscribe() + } + + /// Cooperative request followed by executor-specific termination. + pub fn cancel(&self) { + self.cancellation.cancel(); + } + + pub async fn wait(self) -> Result { + Ok(self.join.await?) + } +} + +#[derive(Clone)] +pub struct PVisor { + executors: Arc>>, + event_sink: Arc, +} + +impl Default for PVisor { + fn default() -> Self { + Self::new() + } +} + +impl PVisor { + pub fn new() -> Self { + Self { + executors: Arc::new(vec![Arc::new(ProcessExecutor)]), + event_sink: Arc::new(NoopEventSink), + } + } + + pub fn with_event_sink(mut self, event_sink: Arc) -> Self { + self.event_sink = event_sink; + self + } + + pub fn with_executors( + executors: Vec>, + event_sink: Arc, + ) -> Self { + Self { + executors: Arc::new(executors), + event_sink, + } + } + + pub async fn submit(&self, spec: RunSpec) -> Result { + validate_spec(&spec)?; + let executor = self + .executors + .iter() + .find(|executor| executor.supports(&spec.invocation)) + .cloned() + .ok_or(PVisorError::UnsupportedInvocation)?; + let descriptor = executor.descriptor(); + if spec.runtime.policy_mode == PolicyMode::Enforce && !descriptor.enforces_capabilities { + return Err(PVisorError::UnsupportedPolicy(descriptor.name)); + } + + let run_id = spec.run_id.clone(); + let attempt_id = AttemptId::new(format!("attempt-{}", uuid::Uuid::new_v4())); + let now = unix_now_ms(); + let initial = RunStatus { + run_id: run_id.clone(), + state: RunState::Created, + attempt: AttemptInfo { + attempt_id: attempt_id.clone(), + number: 0, + executor: descriptor.clone(), + started_at_unix_ms: None, + finished_at_unix_ms: None, + }, + updated_at_unix_ms: now, + message: None, + }; + let (status_tx, status_rx) = watch::channel(initial); + let (live_tx, _) = broadcast::channel(256); + let events = RunEventPublisher::new( + run_id.clone(), + attempt_id.clone(), + "persisting-pvisor", + Arc::clone(&self.event_sink), + live_tx, + ); + events + .publish( + "run.created", + "runtime", + json!({ + "agent": spec.agent, + "task_id": spec.task_id, + "executor": descriptor, + "policy_mode": spec.runtime.policy_mode, + }), + ) + .await + .map_err(PVisorError::EventSink)?; + + let cancellation = CancellationToken::new(); + let context = AttemptContext::new( + Arc::new(spec), + attempt_id.clone(), + cancellation.clone(), + status_tx, + events.clone(), + ); + let join = tokio::spawn(async move { + let mut result = executor.execute(context.clone()).await; + context.transition(result.state, None).await; + let kind = match result.state { + RunState::Completed => "run.completed", + RunState::Cancelled => "run.cancelled", + _ => "run.failed", + }; + if let Err(error) = context + .events() + .publish( + kind, + "runtime", + json!({ + "state": result.state, + "exit_code": result.exit_code, + "failure": result.failure, + "started_at_unix_ms": result.started_at_unix_ms, + "finished_at_unix_ms": result.finished_at_unix_ms, + }), + ) + .await + { + result + .warnings + .push(format!("terminal event sink failed: {error:#}")); + } + result + }); + + Ok(RunHandle { + run_id, + attempt_id, + status: status_rx, + cancellation, + events, + join, + }) + } +} + +fn validate_spec(spec: &RunSpec) -> Result<(), PVisorError> { + if spec.schema_version != RUNTIME_SCHEMA_VERSION { + return Err(PVisorError::InvalidSpec(format!( + "unsupported schema_version {}; expected {}", + spec.schema_version, RUNTIME_SCHEMA_VERSION + ))); + } + if spec.run_id.is_empty() { + return Err(PVisorError::InvalidSpec("run_id must not be empty".into())); + } + if spec.agent.name.trim().is_empty() { + return Err(PVisorError::InvalidSpec( + "agent.name must not be empty".into(), + )); + } + let persisting_proto::RunInvocation::Process(process) = &spec.invocation; + if process.program.trim().is_empty() { + return Err(PVisorError::InvalidSpec( + "process program must not be empty".into(), + )); + } + if process.stdin == persisting_proto::StdioMode::Capture { + return Err(PVisorError::InvalidSpec( + "captured stdin is not supported in pVisor v1".into(), + )); + } + if spec.runtime.max_output_bytes == 0 { + return Err(PVisorError::InvalidSpec( + "runtime.max_output_bytes must be greater than zero".into(), + )); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::MemoryEventSink; + use persisting_proto::{NetworkCapability, RunFailureKind, RunInvocation, StdioMode}; + + #[cfg(unix)] + #[tokio::test] + async fn process_run_completes_and_emits_lifecycle() { + let sink = Arc::new(MemoryEventSink::default()); + let runtime = PVisor::new().with_event_sink(sink.clone()); + let mut spec = RunSpec::process("run-success", "test-agent", "/bin/sh"); + let RunInvocation::Process(process) = &mut spec.invocation; + process.args = vec!["-c".into(), "printf pvisor".into()]; + process.stdout = StdioMode::Capture; + process.stderr = StdioMode::Capture; + + let handle = runtime.submit(spec).await.unwrap(); + assert_eq!(handle.status().state, RunState::Created); + let result = handle.wait().await.unwrap(); + assert_eq!(result.state, RunState::Completed); + assert_eq!(result.output.stdout.as_deref(), Some("pvisor")); + + let kinds: Vec<_> = sink.events().into_iter().map(|event| event.kind).collect(); + assert_eq!(kinds.first().map(String::as_str), Some("run.created")); + assert_eq!(kinds.last().map(String::as_str), Some("run.completed")); + assert!(kinds.iter().any(|kind| kind == "run.state_changed")); + } + + #[cfg(unix)] + #[tokio::test] + async fn process_run_cancels_the_process_tree() { + let runtime = PVisor::new(); + let mut spec = RunSpec::process("run-cancel", "test-agent", "/bin/sh"); + let RunInvocation::Process(process) = &mut spec.invocation; + process.args = vec!["-c".into(), "sleep 30 & wait".into()]; + process.stdout = StdioMode::Capture; + process.stderr = StdioMode::Capture; + + let handle = runtime.submit(spec).await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(30)).await; + handle.cancel(); + let result = tokio::time::timeout(std::time::Duration::from_secs(3), handle.wait()) + .await + .expect("process tree did not terminate") + .unwrap(); + assert_eq!(result.state, RunState::Cancelled); + } + + #[cfg(unix)] + #[tokio::test] + async fn process_deadline_is_a_typed_failure() { + let runtime = PVisor::new(); + let mut spec = RunSpec::process("run-timeout", "test-agent", "/bin/sleep"); + let RunInvocation::Process(process) = &mut spec.invocation; + process.args = vec!["30".into()]; + process.stdout = StdioMode::Null; + process.stderr = StdioMode::Null; + spec.runtime.timeout_ms = Some(20); + + let result = runtime.submit(spec).await.unwrap().wait().await.unwrap(); + assert_eq!(result.state, RunState::Failed); + assert_eq!( + result.failure.unwrap().kind, + RunFailureKind::DeadlineExceeded + ); + } + + #[tokio::test] + async fn host_process_refuses_enforced_policy() { + let runtime = PVisor::new(); + let mut spec = RunSpec::process("run-enforce", "test-agent", "echo"); + spec.runtime.policy_mode = PolicyMode::Enforce; + spec.capabilities.network = NetworkCapability::Deny; + let error = match runtime.submit(spec).await { + Ok(_) => panic!("host process must not claim capability enforcement"), + Err(error) => error, + }; + assert!(matches!(error, PVisorError::UnsupportedPolicy(_))); + } +} diff --git a/justfile b/justfile index 7f84ec0..bc5abac 100644 --- a/justfile +++ b/justfile @@ -193,7 +193,7 @@ ci: # ── Rust 测试 ───────────────────────────────────────────────────────────────── -# 单 crate：engine | proto | core | capture | cli | compute | dlcapt +# 单 crate：engine | proto | core | capture | cli | compute | pvisor | dlcapt test-crate crate: #!/usr/bin/env bash set -euo pipefail @@ -204,8 +204,9 @@ test-crate crate: capture) cargo test -p persisting-capture ;; cli) cargo test -p persisting-cli ;; compute) cargo test -p persisting-compute ;; + pvisor) cargo test -p persisting-pvisor ;; dlcapt) cargo test -p persisting-dlcapt ;; - *) echo "unknown crate: {{ crate }} (engine|proto|core|capture|cli|compute|dlcapt)" >&2; exit 2 ;; + *) echo "unknown crate: {{ crate }} (engine|proto|core|capture|cli|compute|pvisor|dlcapt)" >&2; exit 2 ;; esac test-rust: From 866026cb0ffebb9a9fc554837c0b6845cd458357 Mon Sep 17 00:00:00 2001 From: Reiase Date: Sun, 26 Jul 2026 22:58:46 +0800 Subject: [PATCH 2/2] fmt code --- .../src/trajectory/judge_columns.rs | 47 ++++++++----------- docs/mkdocs.yml | 1 + 2 files changed, 20 insertions(+), 28 deletions(-) diff --git a/crates/persisting-engine/src/trajectory/judge_columns.rs b/crates/persisting-engine/src/trajectory/judge_columns.rs index c099404..7978716 100644 --- a/crates/persisting-engine/src/trajectory/judge_columns.rs +++ b/crates/persisting-engine/src/trajectory/judge_columns.rs @@ -14,12 +14,8 @@ use std::sync::Arc; use anyhow::{Context, Result}; use futures::TryStreamExt; -use lance::dataset::{ - MergeInsertBuilder, NewColumnTransform, WhenMatched, WhenNotMatched, -}; -use lance::deps::arrow_array::{ - Array, Int64Array, RecordBatch, RecordBatchIterator, StringArray, -}; +use lance::dataset::{MergeInsertBuilder, NewColumnTransform, WhenMatched, WhenNotMatched}; +use lance::deps::arrow_array::{Array, Int64Array, RecordBatch, RecordBatchIterator, StringArray}; use lance::deps::arrow_schema::{DataType, Field, Schema as ArrowSchema}; use lance::Dataset; @@ -130,8 +126,7 @@ pub async fn read_judge_rows(session: &TrajectorySession) -> Result>(), + verdicts.iter().map(|v| v.as_deref()).collect::>(), )), Arc::new(StringArray::from( - rationales - .iter() - .map(|v| v.as_deref()) - .collect::>(), + rationales.iter().map(|v| v.as_deref()).collect::>(), )), Arc::new(StringArray::from( units.iter().map(|v| v.as_deref()).collect::>(), @@ -384,17 +373,16 @@ async fn apply_rubric_rows(ds: &mut Dataset, prefix: &str, rows: &[&JudgeRow]) - .context("build judge merge RecordBatch")?; let reader = Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema)); - let (updated, _stats) = MergeInsertBuilder::try_new(Arc::new(ds.clone()), vec![ - TRAJECTORY_SEQ_COL.to_string(), - ]) - .context("MergeInsertBuilder")? - .when_matched(WhenMatched::UpdateAll) - .when_not_matched(WhenNotMatched::DoNothing) - .try_build() - .context("build merge insert job")? - .execute_reader(reader) - .await - .context("merge insert judge columns")?; + let (updated, _stats) = + MergeInsertBuilder::try_new(Arc::new(ds.clone()), vec![TRAJECTORY_SEQ_COL.to_string()]) + .context("MergeInsertBuilder")? + .when_matched(WhenMatched::UpdateAll) + .when_not_matched(WhenNotMatched::DoNothing) + .try_build() + .context("build merge insert job")? + .execute_reader(reader) + .await + .context("merge insert judge columns")?; *ds = Arc::try_unwrap(updated).unwrap_or_else(|arc| (*arc).clone()); Ok(()) @@ -408,6 +396,9 @@ mod tests { fn sanitize_rubric_for_column_prefix() { assert_eq!(layer_field_name("default"), "judge_default"); assert_eq!(layer_field_name("task/success"), "judge_task_success"); - assert_eq!(rubric_from_score_column("judge_default_score").as_deref(), Some("default")); + assert_eq!( + rubric_from_score_column("judge_default_score").as_deref(), + Some("default") + ); } } diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 33652f1..3438163 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -148,6 +148,7 @@ nav: - search command reference: design/cli-search.md - Architecture & Internals: - Overview and maturity: design/index.md + - Agent infrastructure: design/agent-infrastructure.md - Queue persistence: design/architecture.md - Capture and trajectory: - Capture pipeline: design/capture.md