diff --git a/.gitignore b/.gitignore index 5bcb47a..7e472ec 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,8 @@ __pycache__/ *.egg-info/ .venv/ build/ +target/ +workspace/ dist/ .pytest_cache/ .ruff_cache/ diff --git a/CHANGELOG.md b/CHANGELOG.md index e38b0da..24b9e7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +## 0.2.0a0 — unreleased + +- Add backend-neutral CPU, full-CUDA, and sequential CPU offload placement policy with + reserve/cap/workspace math, real accelerator probe requirements, precision checks, and + bounded AUTO OOM recovery. +- Extend runtime manifests with backward-compatible placement policy and capability + metadata without changing C ABI v1. +- Add optional Torch/Accelerate adapter with observed sequential-offload verification. +- Add process-local Rust host lifecycle, identity-bound snapshot reuse, bounded host-owned + result buffers, normalized diagnostic statuses, measurements, unload, and quarantine. +- Add executable deterministic Rust HMM provider baseline and provider-host benchmark. +- Add bounded investigator context packing, explicit workspace access, candidate transaction, + identity envelope, quarantine, and morning-report contracts. + ## 0.1.0a0 — unreleased - Define Machine-Native Experimental Learning and Verified Experience Distillation. diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..be79b0f --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,31 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 3 + +[[package]] +name = "mnel-provider-api" +version = "0.1.0-alpha.0" + +[[package]] +name = "mnel-provider-classical" +version = "0.1.0-alpha.0" +dependencies = [ + "mnel-provider-api", + "mnel-provider-sdk", +] + +[[package]] +name = "mnel-provider-host" +version = "0.1.0-alpha.0" +dependencies = [ + "mnel-provider-api", + "mnel-provider-classical", + "mnel-provider-sdk", +] + +[[package]] +name = "mnel-provider-sdk" +version = "0.1.0-alpha.0" +dependencies = [ + "mnel-provider-api", +] diff --git a/Cargo.toml b/Cargo.toml index 3d76179..a49a0d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "crates/mnel-provider-api", + "crates/mnel-provider-classical", "crates/mnel-provider-sdk", "crates/mnel-provider-host", ] diff --git a/README.md b/README.md index 686621d..7ccaa5e 100644 --- a/README.md +++ b/README.md @@ -37,12 +37,14 @@ experience, negative memory, causal attribution, transfer-gated principles, reus strategies, and append-only candidate lineage rather than relying exclusively on conventional neural-weight training. -> **Current status:** functional `0.1.0a0` foundation. The repository implements the -> core local lifecycle, deterministic evidence ledger, hard-gate evaluator, recursion -> governor, investigator contracts, a diagnostic-only learned micro-provider registry, -> and a testable Rust-first provider runtime contract. It does not yet train or execute -> learned micro-providers, provide unattended model execution, distributed scheduling, -> protected final custody, formal MNCS/MNCDS conformance, or automatic RAVEL promotion. +> **Current status:** functional `0.2.0a0` iteration. The repository now includes a +> backend-neutral accelerator placement policy, optional Torch/Accelerate adapter, +> process-local persistent Rust host, reusable identity-bound snapshots, bounded and +> normalized diagnostic results, failure quarantine, an executable Rust HMM baseline, +> deterministic runtime measurements, and bounded investigator context/workspace +> contracts. It still does not provide dynamic library loading, process isolation, +> unattended model execution, distributed scheduling, protected final custody, formal +> MNCS/MNCDS conformance, or automatic RAVEL promotion. ## Core rule @@ -86,6 +88,14 @@ copy their authority or silently create substitute implementations. - accepted Rust-first runtime architecture decision and versioned C ABI; - safe Rust provider SDK, host admission policy, reusable snapshot cache, and runtime manifest validation; +- backend-neutral CPU/full-CUDA/sequential-CPU-offload policy with reserve/cap/workspace + accounting, real-probe requirements, precision checks, and bounded AUTO OOM recovery; +- explicit distinction between persistent provider lifetime and physical weight placement; +- process-local Rust host lifecycle, snapshot reuse, output-buffer limits, normalized + status handling, measurement collection, and deterministic quarantine; +- executable `mnel-provider-classical` HMM diagnostic provider and host integration tests; +- eligible-context packing, read-only/proposal workspace models, identity envelopes, + candidate transactions, quarantine queues, and deterministic morning-report records; - deterministic reference workflow, JSON schemas, mutation-oriented tests, and CI. ## Install @@ -132,8 +142,15 @@ mnel learned-provider match \ --limit 4 ``` -These commands inspect declarations and matching only. They do not download, train, or -execute any model. +The catalog commands do not download or train models. The Rust reference provider can be +measured locally with: + +```bash +cargo run -p mnel-provider-host --example provider-benchmark +``` + +The benchmark reports one local observation of host admission and warm invocation timing; +hardware-independent placement policy tests use fake accelerator diagnostics. Run the deterministic reference lifecycle: @@ -201,6 +218,15 @@ path. See [ADR 0001](docs/decisions/0001-rust-provider-runtime.md) and the [learned-provider runtime contract](docs/LEARNED_PROVIDER_RUNTIME.md). +### Placement and residency + +Provider admission is persistent, but GPU residency is a separate policy decision. +`resident-on-admission` means the provider is loaded and reusable; it does not mean all +weights must remain permanently on a GPU. With sequential CPU offload, weights remain in +system RAM while individual modules execute temporarily on CUDA, trading VRAM for host +memory and transfer overhead. Explicit CPU/CUDA/offload choices are honored or rejected; +only `auto` may use bounded full-CUDA → sequential-offload → CPU recovery. + ## Investigator roles - **Investigator** — proposes falsifiable hypotheses and bounded interventions. @@ -239,9 +265,10 @@ state, and failure modes. src/mnel/ Python control plane and executable foundation crates/mnel-provider-api/ versioned provider ABI vocabulary crates/mnel-provider-sdk/ safe Rust provider authoring surface -crates/mnel-provider-host/ admission policy and reusable snapshot storage +crates/mnel-provider-host/ persistent process-local host, policy, and snapshots +crates/mnel-provider-classical/ executable deterministic HMM diagnostic baseline include/ language-neutral provider ABI header -schemas/ machine-readable record vocabulary +schemas/ machine-readable record and runtime vocabulary docs/ architecture, decisions, method, boundaries, roadmap examples/reference-study/ deterministic lifecycle example examples/learned-providers/ architecture catalog and runtime manifest example diff --git a/crates/mnel-provider-classical/Cargo.toml b/crates/mnel-provider-classical/Cargo.toml new file mode 100644 index 0000000..5c58084 --- /dev/null +++ b/crates/mnel-provider-classical/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "mnel-provider-classical" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Deterministic Rust classical provider baseline for MNEL" + +[dependencies] +mnel-provider-api = { path = "../mnel-provider-api" } +mnel-provider-sdk = { path = "../mnel-provider-sdk" } + +[lints] +workspace = true diff --git a/crates/mnel-provider-classical/src/lib.rs b/crates/mnel-provider-classical/src/lib.rs new file mode 100644 index 0000000..437b3ee --- /dev/null +++ b/crates/mnel-provider-classical/src/lib.rs @@ -0,0 +1,137 @@ +//! A small executable HMM-style diagnostic provider. +//! +//! The provider is intentionally boring: it consumes a compact state-sequence snapshot, +//! computes a bounded negative log likelihood against a fixed transition model, and +//! returns an anomaly observation. It has no evaluator or promotion authority. + +use mnel_provider_api::OUTPUT_ANOMALY_SCORE; +use mnel_provider_sdk::{DiagnosticResult, Invocation, LearnedProvider, ProviderError}; + +const STATE_COUNT: usize = 4; +const TRANSITION_PROBABILITIES: [[f64; STATE_COUNT]; STATE_COUNT] = [ + [0.70, 0.20, 0.08, 0.02], + [0.05, 0.75, 0.15, 0.05], + [0.10, 0.10, 0.70, 0.10], + [0.20, 0.10, 0.10, 0.60], +]; + +#[derive(Clone, Copy, Debug, Default)] +pub struct HiddenMarkovProvider; + +impl HiddenMarkovProvider { + pub const fn new() -> Self { + Self + } +} + +impl LearnedProvider for HiddenMarkovProvider { + fn infer(&self, invocation: &Invocation<'_>) -> Result { + let snapshot = invocation + .snapshots() + .first() + .ok_or(ProviderError::MissingSnapshots)?; + let sequence = snapshot.payload; + let operation_limit = invocation.budget().operation_limit as usize; + if sequence.len() > operation_limit { + return Err(ProviderError::BudgetExceeded); + } + if sequence + .iter() + .any(|state| usize::from(*state) >= STATE_COUNT) + { + return Err(ProviderError::OutOfDistribution); + } + + let mut negative_log_likelihood = 0.0_f64; + let mut transitions = 0_usize; + for pair in sequence.windows(2) { + let probability = TRANSITION_PROBABILITIES[usize::from(pair[0])][usize::from(pair[1])]; + negative_log_likelihood -= probability.ln(); + transitions += 1; + } + let score = if transitions == 0 { + 0.0 + } else { + negative_log_likelihood / transitions as f64 + }; + let calibration_band = if score < 0.6 { + 0 + } else if score < 1.5 { + 1 + } else { + 2 + }; + let payload = format!("transitions={transitions};mean_negative_log_likelihood={score:.6}") + .into_bytes(); + DiagnosticResult { + output_kind: OUTPUT_ANOMALY_SCORE, + value: score, + calibration_band, + out_of_distribution: false, + payload, + } + .validate() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use mnel_provider_api::Digest32; + use mnel_provider_sdk::{InvocationIdentity, ResourceBudget, SnapshotRef}; + + fn digest(value: u8) -> Digest32 { + Digest32 { bytes: [value; 32] } + } + + fn invocation(payload: &[u8], limit: u64) -> Invocation<'_> { + let result = Invocation::new( + InvocationIdentity { + declaration: digest(1), + model: digest(2), + calibration: digest(3), + query: digest(4), + }, + ResourceBudget { + wall_time_ns: 1_000_000, + operation_limit: limit, + memory_bytes: 1024, + }, + vec![SnapshotRef { + schema_version: 1, + identity: digest(5), + feature_extractor_identity: digest(6), + payload, + }], + ); + match result { + Ok(invocation) => invocation, + Err(_) => panic!("test invocation should be bounded and non-empty"), + } + } + + #[test] + fn produces_a_bounded_diagnostic_score() { + let result = match HiddenMarkovProvider::new().infer(&invocation(&[0, 0, 1, 1, 2], 10)) { + Ok(result) => result, + Err(_) => panic!("provider should infer"), + }; + assert_eq!(result.output_kind, OUTPUT_ANOMALY_SCORE); + assert!(result.value.is_finite()); + assert!(!result.payload.is_empty()); + } + + #[test] + fn respects_operation_budget_and_ood_states() { + let budget_error = match HiddenMarkovProvider::new().infer(&invocation(&[0, 1, 2], 2)) { + Ok(_) => panic!("operation budget should be enforced"), + Err(error) => error, + }; + assert_eq!(budget_error, ProviderError::BudgetExceeded); + let ood_error = match HiddenMarkovProvider::new().infer(&invocation(&[0, 9], 10)) { + Ok(_) => panic!("unknown state should abstain as OOD"), + Err(error) => error, + }; + assert_eq!(ood_error, ProviderError::OutOfDistribution); + } +} diff --git a/crates/mnel-provider-host/Cargo.toml b/crates/mnel-provider-host/Cargo.toml index 2f1db7b..56cde28 100644 --- a/crates/mnel-provider-host/Cargo.toml +++ b/crates/mnel-provider-host/Cargo.toml @@ -9,6 +9,10 @@ description = "Persistent policy host for MNEL learned micro-providers" [dependencies] mnel-provider-api = { path = "../mnel-provider-api" } +mnel-provider-sdk = { path = "../mnel-provider-sdk" } + +[dev-dependencies] +mnel-provider-classical = { path = "../mnel-provider-classical" } [lints] workspace = true diff --git a/crates/mnel-provider-host/examples/provider-benchmark.rs b/crates/mnel-provider-host/examples/provider-benchmark.rs new file mode 100644 index 0000000..8accae4 --- /dev/null +++ b/crates/mnel-provider-host/examples/provider-benchmark.rs @@ -0,0 +1,77 @@ +use std::sync::Arc; +use std::time::Instant; + +use mnel_provider_api::{Digest32, ABI_VERSION_V1}; +use mnel_provider_classical::HiddenMarkovProvider; +use mnel_provider_host::placement::{PlacementCapabilities, PlacementPolicy}; +use mnel_provider_host::{ + CachedSnapshot, ExecutionTier, ImplementationLanguage, ProviderHost, ProviderManifest, +}; +use mnel_provider_sdk::{InvocationIdentity, ResourceBudget}; + +fn digest(value: u8) -> Digest32 { + Digest32 { bytes: [value; 32] } +} + +fn main() { + let manifest = ProviderManifest { + provider_id: "state.hidden-markov-model".to_owned(), + provider_version: "0.1.0".to_owned(), + declaration_identity: digest(1), + artifact_identity: digest(2), + language: ImplementationLanguage::Rust, + tier: ExecutionTier::NativeTrusted, + abi_version: ABI_VERSION_V1, + persistent_host: true, + language_exception: None, + placement_policy: PlacementPolicy::default(), + placement_capabilities: PlacementCapabilities::default(), + }; + let mut host = ProviderHost::new(3, 1024); + let cold_started = Instant::now(); + if let Err(error) = host.admit(manifest, Arc::new(HiddenMarkovProvider::new())) { + panic!("benchmark provider admission should succeed: {error:?}"); + } + let cold_admission_ns = cold_started.elapsed().as_nanos(); + let snapshot_id = digest(9); + host.register_snapshot(CachedSnapshot { + identity: snapshot_id, + feature_extractor_identity: digest(10), + payload: Arc::from([0_u8, 0, 1, 1, 2, 2, 3, 3]), + }); + let identities = InvocationIdentity { + declaration: digest(1), + model: digest(2), + calibration: digest(3), + query: digest(4), + }; + let budget = ResourceBudget { + wall_time_ns: 1_000_000, + operation_limit: 100, + memory_bytes: 1024, + }; + let mut samples = Vec::new(); + let mut process_rss_bytes = None; + for _ in 0..32 { + let result = match host.invoke( + "state.hidden-markov-model", + identities, + &[snapshot_id], + budget, + ) { + Ok(result) => result, + Err(error) => panic!("benchmark invocation should succeed: {error:?}"), + }; + samples.push(result.measurement.elapsed_ns); + process_rss_bytes = result.measurement.process_rss_bytes; + } + samples.sort_unstable(); + let p50 = samples[samples.len() / 2]; + let p95 = samples[(samples.len() * 95 / 100).min(samples.len() - 1)]; + let p99 = samples[(samples.len() * 99 / 100).min(samples.len() - 1)]; + println!( + "{{\"provider\":\"state.hidden-markov-model\",\"cold_admission_ns\":{cold_admission_ns},\"warm_p50_ns\":{p50},\"warm_p95_ns\":{p95},\"warm_p99_ns\":{p99},\"samples\":{},\"snapshot_reuse\":true,\"bytes_copied_per_invocation\":0,\"process_rss_bytes\":{rss},\"execution_mode\":\"cpu\",\"precision\":\"float32\",\"placement_reason\":\"native Rust CPU backend; no accelerator adapter attached\"}}", + samples.len(), + rss = process_rss_bytes.map_or_else(|| "null".to_owned(), |value| value.to_string()) + ); +} diff --git a/crates/mnel-provider-host/src/lib.rs b/crates/mnel-provider-host/src/lib.rs index 966978b..589efd6 100644 --- a/crates/mnel-provider-host/src/lib.rs +++ b/crates/mnel-provider-host/src/lib.rs @@ -5,8 +5,22 @@ use std::collections::BTreeMap; use std::sync::Arc; +use std::time::Instant; -use mnel_provider_api::{Digest32, ABI_VERSION_V1}; +use mnel_provider_api::{ + Digest32, ProviderStatusV1, ABI_VERSION_V1, AUTHORITY_DIAGNOSTIC_ONLY, OUTPUT_ANOMALY_SCORE, + PROVIDER_STATUS_ABSTAINED, PROVIDER_STATUS_BUDGET_EXCEEDED, PROVIDER_STATUS_COMPLETED, + PROVIDER_STATUS_INVALID_INPUT, PROVIDER_STATUS_OUT_OF_DISTRIBUTION, + PROVIDER_STATUS_RUNTIME_ERROR, RESULT_FLAG_OUT_OF_DISTRIBUTION, RESULT_FLAG_TRUNCATED_PAYLOAD, + VERDICT_SEMANTICS_NOT_A_VERDICT, +}; +use mnel_provider_sdk::{ + DiagnosticResult, Invocation, InvocationIdentity, LearnedProvider, ProviderError, + ResourceBudget, SnapshotRef, +}; + +pub mod placement; +use placement::{PlacementCapabilities, PlacementError, PlacementPolicy}; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum ImplementationLanguage { @@ -44,6 +58,8 @@ pub struct ProviderManifest { pub abi_version: u32, pub persistent_host: bool, pub language_exception: Option, + pub placement_policy: PlacementPolicy, + pub placement_capabilities: PlacementCapabilities, } impl ProviderManifest { @@ -57,6 +73,31 @@ impl ProviderManifest { if !self.persistent_host { return Err(AdmissionError::ProcessPerInvocationForbidden); } + if self.placement_policy.execution_device == placement::ExecutionDevice::Cpu + && self.placement_policy.offload == placement::OffloadMode::SequentialCpu + { + return Err(AdmissionError::InvalidPlacement( + PlacementError::InvalidPolicy, + )); + } + if self.placement_capabilities.cpu_precisions.is_empty() + || self.placement_capabilities.cuda_precisions.is_empty() + { + return Err(AdmissionError::InvalidPlacement( + PlacementError::MissingCpuPrecision, + )); + } + if self.placement_policy.max_vram_bytes == Some(0) + || self.placement_policy.host_memory_budget_bytes == Some(0) + || self + .placement_policy + .host_memory_budget_bytes + .is_some_and(|budget| self.placement_policy.model_storage_bytes > budget) + { + return Err(AdmissionError::InvalidPlacement( + PlacementError::InvalidPolicy, + )); + } match self.tier { ExecutionTier::NativeTrusted => { if self.language != ImplementationLanguage::Rust { @@ -85,6 +126,12 @@ impl ProviderManifest { } } } + if self.language_exception.is_some() + && !(self.tier == ExecutionTier::NativeTrusted + && self.language != ImplementationLanguage::Rust) + { + return Err(AdmissionError::UnnecessaryLanguageException); + } Ok(()) } } @@ -98,6 +145,7 @@ pub enum AdmissionError { IncompleteLanguageException, TierLanguageMismatch, UnnecessaryLanguageException, + InvalidPlacement(PlacementError), DuplicateProvider, } @@ -116,6 +164,9 @@ pub struct SnapshotCache { impl SnapshotCache { pub fn insert(&mut self, snapshot: CachedSnapshot) -> Arc<[u8]> { let key = snapshot.identity.bytes; + if let Some(existing) = self.entries.get(&key) { + return Arc::clone(&existing.payload); + } let payload = Arc::clone(&snapshot.payload); self.entries.insert(key, snapshot); payload @@ -145,13 +196,311 @@ impl ProviderCatalog { if self.manifests.contains_key(&manifest.provider_id) { return Err(AdmissionError::DuplicateProvider); } - self.manifests.insert(manifest.provider_id.clone(), manifest); + self.manifests + .insert(manifest.provider_id.clone(), manifest); Ok(()) } pub fn get(&self, provider_id: &str) -> Option<&ProviderManifest> { self.manifests.get(provider_id) } + + pub fn remove(&mut self, provider_id: &str) -> Option { + self.manifests.remove(provider_id) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ProviderState { + Active, + Quarantined { + failures: u32, + reason: ProviderError, + }, +} + +struct HostedProvider { + provider: Arc, + state: ProviderState, + failures: u32, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct InvocationMeasurement { + pub elapsed_ns: u64, + pub bytes_copied: u64, + pub snapshots_reused: u32, + pub output_bytes: u64, + pub execution_mode: String, + pub precision: String, + pub placement_reason: String, + pub process_rss_bytes: Option, +} + +#[derive(Clone, Debug)] +pub struct HostedInvocation { + pub status: ProviderStatusV1, + pub output_kind: u32, + pub scalar_value: f64, + pub calibration_band: u32, + pub flags: u64, + pub payload: Vec, + pub authority: u32, + pub verdict_semantics: u32, + pub state: ProviderState, + pub measurement: InvocationMeasurement, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum HostError { + Admission(AdmissionError), + AcceleratorBackendUnavailable, + UnknownProvider, + ProviderQuarantined, + MissingSnapshot, + InvalidInvocation(ProviderError), + SnapshotIdentityMismatch, +} + +pub struct ProviderHost { + catalog: ProviderCatalog, + snapshots: SnapshotCache, + providers: BTreeMap, + max_failures: u32, + max_output_bytes: usize, +} + +impl ProviderHost { + pub fn new(max_failures: u32, max_output_bytes: usize) -> Self { + Self { + catalog: ProviderCatalog::default(), + snapshots: SnapshotCache::default(), + providers: BTreeMap::new(), + max_failures: max_failures.max(1), + max_output_bytes: max_output_bytes.max(1), + } + } + + pub fn admit( + &mut self, + manifest: ProviderManifest, + provider: Arc, + ) -> Result<(), HostError> { + if manifest.placement_policy.execution_device == placement::ExecutionDevice::Cuda + || manifest.placement_policy.offload == placement::OffloadMode::SequentialCpu + { + return Err(HostError::AcceleratorBackendUnavailable); + } + let provider_id = manifest.provider_id.clone(); + self.catalog + .admit(manifest.clone()) + .map_err(HostError::Admission)?; + self.providers.insert( + provider_id, + HostedProvider { + provider, + state: ProviderState::Active, + failures: 0, + }, + ); + Ok(()) + } + + pub fn register_snapshot(&mut self, snapshot: CachedSnapshot) -> Arc<[u8]> { + self.snapshots.insert(snapshot) + } + + pub fn snapshot_count(&self) -> usize { + self.snapshots.len() + } + + pub fn state(&self, provider_id: &str) -> Option<&ProviderState> { + self.providers + .get(provider_id) + .map(|provider| &provider.state) + } + + pub fn unload(&mut self, provider_id: &str) -> Result<(), HostError> { + if self.providers.remove(provider_id).is_none() { + return Err(HostError::UnknownProvider); + } + self.catalog.remove(provider_id); + Ok(()) + } + + pub fn invoke( + &mut self, + provider_id: &str, + identities: InvocationIdentity, + snapshot_ids: &[Digest32], + budget: ResourceBudget, + ) -> Result { + let hosted = self + .providers + .get(provider_id) + .ok_or(HostError::UnknownProvider)?; + if hosted.state != ProviderState::Active { + return Err(HostError::ProviderQuarantined); + } + let provider = Arc::clone(&hosted.provider); + let mut payloads = Vec::with_capacity(snapshot_ids.len()); + for identity in snapshot_ids { + let snapshot = self + .snapshots + .get(identity) + .ok_or(HostError::MissingSnapshot)?; + if snapshot.identity != *identity { + return Err(HostError::SnapshotIdentityMismatch); + } + payloads.push(Arc::clone(&snapshot.payload)); + } + let snapshot_refs = payloads + .iter() + .zip(snapshot_ids.iter()) + .map(|(payload, identity)| SnapshotRef { + schema_version: 1, + identity: *identity, + feature_extractor_identity: Digest32::ZERO, + payload, + }) + .collect::>(); + let invocation = Invocation::new(identities, budget, snapshot_refs) + .map_err(HostError::InvalidInvocation)?; + let started = Instant::now(); + let outcome = provider.infer(&invocation); + let elapsed_ns = started.elapsed().as_nanos() as u64; + let result = match outcome { + Ok(result) => self.normalize_success( + provider_id, + result, + budget, + snapshot_ids.len() as u32, + elapsed_ns, + ), + Err(error) => { + self.normalize_error(provider_id, error, snapshot_ids.len() as u32, elapsed_ns) + } + }; + Ok(result) + } + + fn normalize_success( + &mut self, + provider_id: &str, + result: DiagnosticResult, + budget: ResourceBudget, + snapshots_reused: u32, + elapsed_ns: u64, + ) -> HostedInvocation { + let output_limit = self.max_output_bytes.min(budget.memory_bytes as usize); + let mut flags = 0; + let mut payload = result.payload; + let original_output_bytes = payload.len() as u64; + let mut status = PROVIDER_STATUS_COMPLETED; + if elapsed_ns > budget.wall_time_ns { + status = PROVIDER_STATUS_BUDGET_EXCEEDED; + self.record_failure(provider_id, ProviderError::BudgetExceeded); + } + if payload.len() > output_limit { + payload.truncate(output_limit); + flags |= RESULT_FLAG_TRUNCATED_PAYLOAD; + status = PROVIDER_STATUS_BUDGET_EXCEEDED; + self.record_failure(provider_id, ProviderError::BudgetExceeded); + } + if result.out_of_distribution { + flags |= RESULT_FLAG_OUT_OF_DISTRIBUTION; + } + let state = self.provider_state(provider_id); + HostedInvocation { + status, + output_kind: result.output_kind, + scalar_value: result.value, + calibration_band: result.calibration_band, + flags, + payload, + authority: AUTHORITY_DIAGNOSTIC_ONLY, + verdict_semantics: VERDICT_SEMANTICS_NOT_A_VERDICT, + state, + measurement: measurement(elapsed_ns, original_output_bytes, snapshots_reused), + } + } + + fn normalize_error( + &mut self, + provider_id: &str, + error: ProviderError, + snapshots_reused: u32, + elapsed_ns: u64, + ) -> HostedInvocation { + self.record_failure(provider_id, error); + let (status, flags) = match error { + ProviderError::Abstained => (PROVIDER_STATUS_ABSTAINED, 0), + ProviderError::InvalidBudget | ProviderError::EmptySnapshot => { + (PROVIDER_STATUS_INVALID_INPUT, 0) + } + ProviderError::BudgetExceeded => (PROVIDER_STATUS_BUDGET_EXCEEDED, 0), + ProviderError::OutOfDistribution => ( + PROVIDER_STATUS_OUT_OF_DISTRIBUTION, + RESULT_FLAG_OUT_OF_DISTRIBUTION, + ), + ProviderError::MissingSnapshots + | ProviderError::NonFiniteResult + | ProviderError::RuntimeFailure => (PROVIDER_STATUS_RUNTIME_ERROR, 0), + }; + HostedInvocation { + status, + output_kind: OUTPUT_ANOMALY_SCORE, + scalar_value: 0.0, + calibration_band: 0, + flags, + payload: Vec::new(), + authority: AUTHORITY_DIAGNOSTIC_ONLY, + verdict_semantics: VERDICT_SEMANTICS_NOT_A_VERDICT, + state: self.provider_state(provider_id), + measurement: measurement(elapsed_ns, 0, snapshots_reused), + } + } + + fn record_failure(&mut self, provider_id: &str, reason: ProviderError) { + if let Some(provider) = self.providers.get_mut(provider_id) { + provider.failures = provider.failures.saturating_add(1); + if provider.failures >= self.max_failures { + provider.state = ProviderState::Quarantined { + failures: provider.failures, + reason, + }; + } + } + } + + fn provider_state(&self, provider_id: &str) -> ProviderState { + self.providers.get(provider_id).map_or( + ProviderState::Quarantined { + failures: 0, + reason: ProviderError::RuntimeFailure, + }, + |provider| provider.state.clone(), + ) + } +} + +fn measurement(elapsed_ns: u64, output_bytes: u64, snapshots_reused: u32) -> InvocationMeasurement { + InvocationMeasurement { + elapsed_ns, + bytes_copied: 0, + snapshots_reused, + output_bytes, + execution_mode: "cpu".to_owned(), + precision: "float32".to_owned(), + placement_reason: "native Rust CPU backend; no accelerator adapter attached".to_owned(), + process_rss_bytes: process_rss_bytes(), + } +} + +fn process_rss_bytes() -> Option { + let contents = std::fs::read_to_string("/proc/self/statm").ok()?; + let resident_pages = contents.split_whitespace().nth(1)?.parse::().ok()?; + Some(resident_pages.saturating_mul(4096)) } #[cfg(test)] @@ -162,6 +511,13 @@ mod tests { Digest32 { bytes: [value; 32] } } + fn ok(result: Result) -> T { + match result { + Ok(value) => value, + Err(error) => panic!("unexpected test error: {error:?}"), + } + } + fn manifest(language: ImplementationLanguage, tier: ExecutionTier) -> ProviderManifest { ProviderManifest { provider_id: "state.hidden-markov-model".to_owned(), @@ -173,6 +529,8 @@ mod tests { abi_version: ABI_VERSION_V1, persistent_host: true, language_exception: None, + placement_policy: PlacementPolicy::default(), + placement_capabilities: PlacementCapabilities::default(), } } @@ -184,6 +542,17 @@ mod tests { ); } + #[test] + fn catalog_rejects_duplicate_admission() { + let mut catalog = ProviderCatalog::default(); + let candidate = manifest(ImplementationLanguage::Rust, ExecutionTier::NativeTrusted); + assert_eq!(catalog.admit(candidate.clone()), Ok(())); + assert_eq!( + catalog.admit(candidate), + Err(AdmissionError::DuplicateProvider) + ); + } + #[test] fn non_rust_native_requires_evidence_backed_exception() { let mut candidate = manifest(ImplementationLanguage::Cpp, ExecutionTier::NativeTrusted); @@ -220,4 +589,180 @@ mod tests { assert!(Arc::ptr_eq(&payload, &returned)); assert_eq!(cache.len(), 1); } + + #[test] + fn snapshot_identity_is_append_only_for_repeated_registration() { + let mut cache = SnapshotCache::default(); + let identity = digest(15); + cache.insert(CachedSnapshot { + identity, + feature_extractor_identity: digest(16), + payload: Arc::from([1_u8]), + }); + cache.insert(CachedSnapshot { + identity, + feature_extractor_identity: digest(17), + payload: Arc::from([9_u8]), + }); + assert_eq!( + cache.get(&identity).map(|snapshot| &*snapshot.payload), + Some(&[1_u8][..]) + ); + assert_eq!(cache.len(), 1); + } + + #[test] + fn process_local_host_reuses_snapshot_and_normalizes_diagnostic_output() { + use mnel_provider_classical::HiddenMarkovProvider; + use mnel_provider_sdk::{InvocationIdentity, ResourceBudget}; + + let provider_id = "state.hidden-markov-model"; + let mut host = ProviderHost::new(3, 128); + ok(host.admit( + manifest(ImplementationLanguage::Rust, ExecutionTier::NativeTrusted), + Arc::new(HiddenMarkovProvider::new()), + )); + let snapshot_id = digest(9); + host.register_snapshot(CachedSnapshot { + identity: snapshot_id, + feature_extractor_identity: digest(10), + payload: Arc::from([0_u8, 0, 1, 1, 2]), + }); + let result = ok(host.invoke( + provider_id, + InvocationIdentity { + declaration: digest(1), + model: digest(2), + calibration: digest(3), + query: digest(4), + }, + &[snapshot_id], + ResourceBudget { + wall_time_ns: 1_000_000, + operation_limit: 100, + memory_bytes: 128, + }, + )); + assert_eq!(result.status, PROVIDER_STATUS_COMPLETED); + assert_eq!(result.authority, AUTHORITY_DIAGNOSTIC_ONLY); + assert_eq!(result.verdict_semantics, VERDICT_SEMANTICS_NOT_A_VERDICT); + assert_eq!(result.measurement.bytes_copied, 0); + assert_eq!(result.measurement.snapshots_reused, 1); + assert_eq!(host.snapshot_count(), 1); + assert_eq!(host.unload(provider_id), Ok(())); + assert_eq!(host.state(provider_id), None); + } + + #[test] + fn cpu_host_rejects_explicit_accelerator_mode_instead_of_silently_falling_back() { + let mut candidate = manifest(ImplementationLanguage::Rust, ExecutionTier::NativeTrusted); + candidate.placement_policy.execution_device = placement::ExecutionDevice::Cuda; + let mut host = ProviderHost::new(3, 128); + assert_eq!( + host.admit(candidate, Arc::new(VerboseProvider)), + Err(HostError::AcceleratorBackendUnavailable) + ); + } + + struct VerboseProvider; + + impl LearnedProvider for VerboseProvider { + fn infer(&self, _invocation: &Invocation<'_>) -> Result { + Ok(DiagnosticResult { + output_kind: OUTPUT_ANOMALY_SCORE, + value: 0.5, + calibration_band: 0, + out_of_distribution: false, + payload: vec![7; 32], + }) + } + } + + #[test] + fn host_owned_result_buffer_is_bounded() { + let mut host = ProviderHost::new(3, 8); + let provider_id = "state.hidden-markov-model"; + ok(host.admit( + manifest(ImplementationLanguage::Rust, ExecutionTier::NativeTrusted), + Arc::new(VerboseProvider), + )); + let snapshot_id = digest(11); + host.register_snapshot(CachedSnapshot { + identity: snapshot_id, + feature_extractor_identity: digest(12), + payload: Arc::from([0_u8]), + }); + let result = ok(host.invoke( + provider_id, + InvocationIdentity { + declaration: digest(1), + model: digest(2), + calibration: digest(3), + query: digest(4), + }, + &[snapshot_id], + ResourceBudget { + wall_time_ns: 1_000_000, + operation_limit: 10, + memory_bytes: 8, + }, + )); + assert_eq!(result.status, PROVIDER_STATUS_BUDGET_EXCEEDED); + assert_eq!(result.payload.len(), 8); + assert_eq!(result.flags, RESULT_FLAG_TRUNCATED_PAYLOAD); + } + + struct FailingProvider; + + impl LearnedProvider for FailingProvider { + fn infer(&self, _invocation: &Invocation<'_>) -> Result { + Err(ProviderError::RuntimeFailure) + } + } + + #[test] + fn repeated_provider_failures_quarantine_without_affecting_other_catalog_state() { + let mut host = ProviderHost::new(2, 128); + let provider_id = "state.hidden-markov-model"; + ok(host.admit( + manifest(ImplementationLanguage::Rust, ExecutionTier::NativeTrusted), + Arc::new(FailingProvider), + )); + let snapshot_id = digest(13); + host.register_snapshot(CachedSnapshot { + identity: snapshot_id, + feature_extractor_identity: digest(14), + payload: Arc::from([0_u8]), + }); + let identities = InvocationIdentity { + declaration: digest(1), + model: digest(2), + calibration: digest(3), + query: digest(4), + }; + let budget = ResourceBudget { + wall_time_ns: 1_000_000, + operation_limit: 10, + memory_bytes: 128, + }; + assert_eq!( + ok(host.invoke(provider_id, identities, &[snapshot_id], budget)).status, + PROVIDER_STATUS_RUNTIME_ERROR + ); + assert_eq!( + ok(host.invoke(provider_id, identities, &[snapshot_id], budget)).status, + PROVIDER_STATUS_RUNTIME_ERROR + ); + assert_eq!( + host.state(provider_id), + Some(&ProviderState::Quarantined { + failures: 2, + reason: ProviderError::RuntimeFailure, + }) + ); + match host.invoke(provider_id, identities, &[snapshot_id], budget) { + Ok(_) => panic!("quarantined providers must not be invoked"), + Err(error) => assert_eq!(error, HostError::ProviderQuarantined), + } + } } diff --git a/crates/mnel-provider-host/src/placement.rs b/crates/mnel-provider-host/src/placement.rs new file mode 100644 index 0000000..e01a559 --- /dev/null +++ b/crates/mnel-provider-host/src/placement.rs @@ -0,0 +1,403 @@ +//! Backend-neutral accelerator placement policy. +//! +//! This module records policy and decisions only. A CUDA, vendor, or Torch adapter +//! supplies diagnostics and applies the decision behind an explicit capability boundary. + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ExecutionDevice { + Auto, + Cpu, + Cuda, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum OffloadMode { + Auto, + None, + SequentialCpu, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Precision { + Auto, + Float32, + Float16, + Bfloat16, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ExecutionMode { + Cpu, + FullCuda, + SequentialCpuOffload, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct PlacementPolicy { + pub execution_device: ExecutionDevice, + pub offload: OffloadMode, + pub precision: Precision, + pub gpu_reserve_bytes: u64, + pub max_vram_bytes: Option, + pub model_storage_bytes: u64, + pub workspace_bytes: u64, + pub host_memory_budget_bytes: Option, +} + +impl Default for PlacementPolicy { + fn default() -> Self { + Self { + execution_device: ExecutionDevice::Auto, + offload: OffloadMode::Auto, + precision: Precision::Auto, + gpu_reserve_bytes: 256 * 1024 * 1024, + max_vram_bytes: None, + model_storage_bytes: 0, + workspace_bytes: 256 * 1024 * 1024, + host_memory_budget_bytes: None, + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct PlacementCapabilities { + pub supports_sequential_cpu_offload: bool, + pub cpu_precisions: Vec, + pub cuda_precisions: Vec, +} + +impl Default for PlacementCapabilities { + fn default() -> Self { + Self { + supports_sequential_cpu_offload: false, + cpu_precisions: vec![Precision::Float32], + cuda_precisions: vec![Precision::Float32, Precision::Float16, Precision::Bfloat16], + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AcceleratorDiagnostics { + pub accelerator_available: bool, + pub execution_probe_succeeded: bool, + pub free_vram_bytes: Option, + pub accelerator_identity: Option, + pub float16_probe_succeeded: Option, + pub bfloat16_probe_succeeded: Option, + pub probe_error: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct PlacementDecision { + pub execution_mode: ExecutionMode, + pub execution_device: ExecutionDevice, + pub offload: OffloadMode, + pub precision: Precision, + pub reason: String, + pub configured_gpu_reserve_bytes: u64, + pub configured_max_vram_bytes: Option, + pub effective_gpu_budget_bytes: u64, + pub estimated_model_bytes: u64, + pub estimated_workspace_bytes: u64, + pub full_cuda_required_bytes: u64, + pub host_memory_required_bytes: u64, + pub sequential_offload_supported: bool, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum PlacementError { + InvalidPolicy, + MissingCpuPrecision, + MissingCudaPrecision, + HostMemoryBudgetExceeded, + CudaUnavailable, + UnsupportedPrecision, + PrecisionProbeFailed, + SequentialOffloadUnsupported, + FullCudaBudgetExceeded, +} + +fn precision_bytes(model_storage_bytes: u64, precision: Precision) -> u64 { + match precision { + Precision::Float16 | Precision::Bfloat16 => model_storage_bytes.div_ceil(2), + Precision::Auto | Precision::Float32 => model_storage_bytes, + } +} + +fn cuda_precision( + policy: &PlacementPolicy, + diagnostics: &AcceleratorDiagnostics, + capabilities: &PlacementCapabilities, +) -> Result { + let precision = match policy.precision { + Precision::Auto => { + if capabilities.cuda_precisions.contains(&Precision::Bfloat16) + && diagnostics.bfloat16_probe_succeeded == Some(true) + { + Precision::Bfloat16 + } else if capabilities.cuda_precisions.contains(&Precision::Float16) + && diagnostics.float16_probe_succeeded == Some(true) + { + Precision::Float16 + } else { + Precision::Float32 + } + } + requested => requested, + }; + if !capabilities.cuda_precisions.contains(&precision) || precision == Precision::Auto { + return Err(PlacementError::UnsupportedPrecision); + } + if (precision == Precision::Float16 && diagnostics.float16_probe_succeeded != Some(true)) + || (precision == Precision::Bfloat16 && diagnostics.bfloat16_probe_succeeded != Some(true)) + { + return Err(PlacementError::PrecisionProbeFailed); + } + Ok(precision) +} + +pub fn effective_gpu_budget_bytes( + free_vram_bytes: Option, + reserve_bytes: u64, + max_vram_bytes: Option, +) -> u64 { + let Some(free) = free_vram_bytes else { + return 0; + }; + let capped = max_vram_bytes.map_or(free, |maximum| free.min(maximum)); + capped.saturating_sub(reserve_bytes) +} + +pub fn decide_placement( + policy: &PlacementPolicy, + diagnostics: &AcceleratorDiagnostics, + capabilities: &PlacementCapabilities, +) -> Result { + if policy.execution_device == ExecutionDevice::Cpu + && policy.offload == OffloadMode::SequentialCpu + { + return Err(PlacementError::InvalidPolicy); + } + if capabilities.cpu_precisions.is_empty() { + return Err(PlacementError::MissingCpuPrecision); + } + if capabilities.cuda_precisions.is_empty() { + return Err(PlacementError::MissingCudaPrecision); + } + if policy.max_vram_bytes == Some(0) || policy.host_memory_budget_bytes == Some(0) { + return Err(PlacementError::InvalidPolicy); + } + if policy + .host_memory_budget_bytes + .is_some_and(|budget| policy.model_storage_bytes > budget) + { + return Err(PlacementError::HostMemoryBudgetExceeded); + } + + let gpu_budget = effective_gpu_budget_bytes( + diagnostics.free_vram_bytes, + policy.gpu_reserve_bytes, + policy.max_vram_bytes, + ); + let usable = diagnostics.accelerator_available && diagnostics.execution_probe_succeeded; + + let make_decision = |mode: ExecutionMode, precision: Precision, reason: &str| { + let estimated_model = precision_bytes(policy.model_storage_bytes, precision); + PlacementDecision { + execution_mode: mode, + execution_device: if mode == ExecutionMode::Cpu { + ExecutionDevice::Cpu + } else { + ExecutionDevice::Cuda + }, + offload: if mode == ExecutionMode::SequentialCpuOffload { + OffloadMode::SequentialCpu + } else { + OffloadMode::None + }, + precision, + reason: reason.to_owned(), + configured_gpu_reserve_bytes: policy.gpu_reserve_bytes, + configured_max_vram_bytes: policy.max_vram_bytes, + effective_gpu_budget_bytes: gpu_budget, + estimated_model_bytes: estimated_model, + estimated_workspace_bytes: policy.workspace_bytes, + full_cuda_required_bytes: estimated_model.saturating_add(policy.workspace_bytes), + host_memory_required_bytes: policy.model_storage_bytes, + sequential_offload_supported: capabilities.supports_sequential_cpu_offload, + } + }; + + if policy.execution_device == ExecutionDevice::Cpu { + let precision = if policy.precision == Precision::Auto { + Precision::Float32 + } else { + policy.precision + }; + if !capabilities.cpu_precisions.contains(&precision) { + return Err(PlacementError::UnsupportedPrecision); + } + return Ok(make_decision( + ExecutionMode::Cpu, + precision, + "CPU was explicitly requested", + )); + } + + if !usable { + if policy.execution_device == ExecutionDevice::Cuda + || policy.offload == OffloadMode::SequentialCpu + { + return Err(PlacementError::CudaUnavailable); + } + return Ok(make_decision( + ExecutionMode::Cpu, + Precision::Float32, + "AUTO selected CPU because accelerator execution is unavailable", + )); + } + + let precision = cuda_precision(policy, diagnostics, capabilities)?; + let required = precision_bytes(policy.model_storage_bytes, precision) + .saturating_add(policy.workspace_bytes); + let fits = required <= gpu_budget; + if policy.offload == OffloadMode::SequentialCpu { + if !capabilities.supports_sequential_cpu_offload { + return Err(PlacementError::SequentialOffloadUnsupported); + } + return Ok(make_decision( + ExecutionMode::SequentialCpuOffload, + precision, + "sequential CPU offload was explicitly requested", + )); + } + if policy.offload == OffloadMode::None { + if !fits { + if policy.execution_device == ExecutionDevice::Auto { + return Ok(make_decision( + ExecutionMode::Cpu, + Precision::Float32, + "AUTO selected CPU because full CUDA exceeds budget and offload is disabled", + )); + } + return Err(PlacementError::FullCudaBudgetExceeded); + } + return Ok(make_decision( + ExecutionMode::FullCuda, + precision, + "full CUDA fits the effective GPU budget", + )); + } + if fits { + return Ok(make_decision( + ExecutionMode::FullCuda, + precision, + "full CUDA fits the effective GPU budget", + )); + } + if capabilities.supports_sequential_cpu_offload { + return Ok(make_decision( + ExecutionMode::SequentialCpuOffload, + precision, + "full CUDA exceeds budget; using CPU-backed sequential execution", + )); + } + if policy.execution_device == ExecutionDevice::Auto { + return Ok(make_decision( + ExecutionMode::Cpu, + Precision::Float32, + "AUTO selected CPU because full CUDA exceeds budget and offload is unsupported", + )); + } + Err(PlacementError::FullCudaBudgetExceeded) +} + +pub fn fallback_after_oom( + policy: &PlacementPolicy, + current_mode: ExecutionMode, + capabilities: &PlacementCapabilities, +) -> Option { + if policy.execution_device != ExecutionDevice::Auto { + return None; + } + if current_mode == ExecutionMode::FullCuda && capabilities.supports_sequential_cpu_offload { + return Some(PlacementPolicy { + execution_device: ExecutionDevice::Auto, + offload: OffloadMode::SequentialCpu, + ..policy.clone() + }); + } + if matches!( + current_mode, + ExecutionMode::FullCuda | ExecutionMode::SequentialCpuOffload + ) { + return Some(PlacementPolicy { + execution_device: ExecutionDevice::Cpu, + offload: OffloadMode::None, + precision: Precision::Float32, + ..policy.clone() + }); + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cuda(free_mib: u64) -> AcceleratorDiagnostics { + AcceleratorDiagnostics { + accelerator_available: true, + execution_probe_succeeded: true, + free_vram_bytes: Some(free_mib * 1024 * 1024), + accelerator_identity: Some("fake-cuda".to_owned()), + float16_probe_succeeded: Some(true), + bfloat16_probe_succeeded: Some(false), + probe_error: None, + } + } + + #[test] + fn reserve_and_cap_math_is_deterministic() { + assert_eq!( + effective_gpu_budget_bytes( + Some(2048 * 1024 * 1024), + 256 * 1024 * 1024, + Some(1024 * 1024 * 1024) + ), + 768 * 1024 * 1024 + ); + } + + #[test] + fn auto_selects_sequential_offload_when_full_cuda_does_not_fit() { + let policy = PlacementPolicy { + model_storage_bytes: 2048 * 1024 * 1024, + workspace_bytes: 512 * 1024 * 1024, + ..PlacementPolicy::default() + }; + let capabilities = PlacementCapabilities { + supports_sequential_cpu_offload: true, + ..PlacementCapabilities::default() + }; + let decision = match decide_placement(&policy, &cuda(1024), &capabilities) { + Ok(decision) => decision, + Err(_) => panic!("placement should succeed"), + }; + assert_eq!(decision.execution_mode, ExecutionMode::SequentialCpuOffload); + } + + #[test] + fn explicit_cuda_budget_failure_is_not_silently_recovered() { + let policy = PlacementPolicy { + execution_device: ExecutionDevice::Cuda, + offload: OffloadMode::None, + model_storage_bytes: 2048 * 1024 * 1024, + ..PlacementPolicy::default() + }; + assert_eq!( + decide_placement(&policy, &cuda(1024), &PlacementCapabilities::default()), + Err(PlacementError::FullCudaBudgetExceeded) + ); + } +} diff --git a/crates/mnel-provider-sdk/src/lib.rs b/crates/mnel-provider-sdk/src/lib.rs index ae0a345..ac8af51 100644 --- a/crates/mnel-provider-sdk/src/lib.rs +++ b/crates/mnel-provider-sdk/src/lib.rs @@ -99,6 +99,14 @@ impl<'a> Invocation<'a> { pub fn snapshots(&self) -> &[SnapshotRef<'a>] { &self.snapshot_lifetimes } + + pub fn identities(&self) -> InvocationIdentity { + self.identities + } + + pub fn budget(&self) -> ResourceBudget { + self.budget + } } #[derive(Clone, Debug, PartialEq)] @@ -119,7 +127,7 @@ impl DiagnosticResult { } } -pub trait LearnedProvider { +pub trait LearnedProvider: Send + Sync { fn infer(&self, invocation: &Invocation<'_>) -> Result; } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 1674e10..abc6807 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -66,6 +66,12 @@ useful diagnostic event. An investigator or skeptic must translate an observatio a falsifiable hypothesis, and Forge must answer bounded claims through its normal verifier boundary. +Provider lifetime and physical placement are separate. The persistent Rust host keeps an +admitted provider reusable, while its placement policy may select CPU, full CUDA, or +sequential CPU offload. The latter keeps weights in system RAM and temporarily executes +modules on CUDA; it does not reload a provider for each query. Placement decisions are +resource-accounted and diagnostic, never evaluator decisions. + ### Probe plane Forge supplies small, identity-bearing questions and witnesses. The stable interface is @@ -122,6 +128,11 @@ Parallel workers must never mutate a shared active candidate in place. Each job Workers append results. A separate selector may accept one child transaction; rejected children remain available as negative memory. +The local investigator harness now packs eligible records under explicit visibility and +byte ceilings, separates read-only from proposal workspaces, binds runtime/model/tool +identities, and creates proposal-only candidate transaction identities. These contracts +do not grant filesystem or network authority and do not implement unattended execution. + ## Model independence MNEL is designed so Gemma, Qwen, a multimodal worker, a JEPA-derived predictor, a graph diff --git a/docs/LEARNED_MICRO_PROVIDERS.md b/docs/LEARNED_MICRO_PROVIDERS.md index 5160aa2..5281c16 100644 --- a/docs/LEARNED_MICRO_PROVIDERS.md +++ b/docs/LEARNED_MICRO_PROVIDERS.md @@ -201,13 +201,16 @@ remain valuable when it discovers a distinct class of omitted questions. ## Current implementation boundary -This foundation implements declarations, canonical identities, deterministic matching, +The 0.2 iteration implements declarations, canonical identities, deterministic matching, diversity selection, a diagnostic observation record, CLI inspection, schema, examples, -and tests. +and tests. The Rust host also executes a small deterministic HMM baseline against an +identity-bound transition snapshot. That baseline is a runtime contract test and +diagnostic reference; it is not evidence that the HMM or any catalog architecture +improves MNEL. It does not yet: -- train or execute any model; +- train or execute the learned neural/catalog models; - download third-party weights; - add PyTorch, ONNX, or another runtime dependency; - build Forge diagnostic snapshots; diff --git a/docs/LEARNED_PROVIDER_RUNTIME.md b/docs/LEARNED_PROVIDER_RUNTIME.md index 9585044..70840d8 100644 --- a/docs/LEARNED_PROVIDER_RUNTIME.md +++ b/docs/LEARNED_PROVIDER_RUNTIME.md @@ -33,6 +33,38 @@ The provider host does not evaluate MNEL gates and cannot promote an observation verdict. Its job is admission, dispatch, bounded execution, normalization, and measurement. +## Placement is separate from provider lifetime + +The runtime manifest carries two different facts: + +1. `weight_residency: resident-on-admission` means an admitted provider remains loaded + and reusable; weights are not reloaded for each invocation. +2. `placement` describes where execution and physical weight storage should occur. + +The policy vocabulary is: + +| Field | Values | +|---|---| +| execution device | `auto`, `cpu`, `cuda` | +| offload | `auto`, `none`, `sequential-cpu` | +| precision | `auto`, `float32`, `float16`, `bfloat16` | + +Sequential CPU offload is therefore compatible with persistent admission. Weights remain +in system RAM and supported modules are temporarily moved to CUDA for execution. This +reduces persistent VRAM at the cost of host memory and transfer time; it is not a +process-per-invocation reload strategy. + +The Python control-plane implementation is in `mnel.placement`. The Rust host mirrors the +policy vocabulary in `mnel-provider-host::placement`. Both are backend-neutral. The +optional `mnel.torch_runtime` adapter performs actual Torch probes and applies a decision +through Accelerate when sequential offload is selected. + +AUTO placement requires a real accelerator execution probe, measures currently free VRAM, +subtracts GPU reserve, applies an optional cap, and includes model plus workspace estimates. +It selects full CUDA when it fits, sequential offload when supported, and CPU otherwise. +Only AUTO may recover from a bounded CUDA OOM sequence; explicit operator choices fail +instead of silently changing execution mode. + ## Language and execution tiers | Tier | Default language | Purpose | Admission rule | @@ -103,9 +135,11 @@ An admitted host must: 8. record warm and cold performance separately; and 9. unload or quarantine a provider after integrity, calibration, or budget failures. -The initial Rust host crate establishes admission policy and reusable snapshot storage. -Dynamic loading, operating-system sandboxing, and model-runtime selection remain future -implementation work. +The Rust host now provides a process-local lifecycle for admitted `LearnedProvider` trait +objects, reusable snapshot storage, bounded result normalization, timing/copy measurements, +clean unload, and deterministic failure quarantine. Dynamic shared-library loading, OS +sandboxing, and a production accelerator backend remain future work. The C ABI v1 remains +unchanged. ## Snapshot transport @@ -133,8 +167,8 @@ artifact and does not establish a general language preference. ## Delivery sequence 1. Freeze and test the v1 manifest and ABI vocabulary. -2. Implement a process-local Rust reference provider for a classical baseline. -3. Add the persistent loader and host-owned output buffer enforcement. +2. Use the executable Rust HMM reference provider as the classical baseline. +3. Keep dynamic loading and host-owned ABI output enforcement behind a reviewed boundary. 4. Add Forge snapshot producers and reuse measurements. 5. Export one Python-trained neural provider and compare it with the baseline. 6. Add WASM quarantine only after native measurements establish the overhead budget. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index e31ab50..ed27e0c 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -18,17 +18,18 @@ ## 0.2 — local investigator harness and provider runtime -- native adapter to `MNEL-local-harness`; -- eligible-context retrieval and packing; -- investigator portfolio scheduling; -- explicit read-only and proposal workspaces; -- Git worktree or snapshot-isolated candidate transactions; -- model, quantization, runtime, prompt, and tool-schema identities; -- deterministic morning reports and quarantine queues; -- process-local persistent Rust provider host; -- host-owned bounded result buffers and ABI loader validation; -- first native Rust classical provider baseline; -- warm/cold latency, copy-byte, resident-memory, and snapshot-reuse benchmarks. +- **Implemented:** eligible-context packing with visibility and byte ceilings; +- **Implemented:** explicit read-only/proposal workspace models and proposal-only candidate + transactions; +- **Implemented:** model, quantization, runtime, prompt, and tool-schema identity envelope; +- **Implemented:** deterministic morning-report and quarantine queue records; +- **Implemented:** process-local persistent Rust provider host with reusable provider state; +- **Implemented:** host-owned bounded result normalization, failure quarantine, and clean unload; +- **Implemented:** first native Rust HMM classical provider baseline; +- **Implemented:** warm/cold timing, copied-byte, output, placement, and snapshot-reuse + measurement harness; +- native adapter to `MNEL-local-harness` and Git worktree materialization remain open; +- ABI dynamic-loader validation remains open; ABI v1 itself is unchanged. ## 0.3 — Forge experiment lifecycle diff --git a/docs/THREAT_MODEL.md b/docs/THREAT_MODEL.md index 01870bc..3492eac 100644 --- a/docs/THREAT_MODEL.md +++ b/docs/THREAT_MODEL.md @@ -50,6 +50,14 @@ selection. Model, quantization, runtime, prompt, tool schema, compiler, or context-packing changes silently change behavior. Bind all of them in experiment records. +### False offload claims + +An accelerator may be discoverable while kernels or a requested dtype fail, and a runtime +may accept an offload option without moving weights as intended. Require a real execution +probe before CUDA placement, keep explicit choices fail-closed, record reserve/cap/workspace +math, and mark sequential offload verified only from completed inference plus observed hooks +and parameter residency. + ### Apparent independence Multiple local machines run the same operator-controlled stack. This is replication, @@ -57,7 +65,7 @@ not independent evaluation or protected custody. ## Current residual risks -The foundation does not provide process isolation, network enforcement, cgroups, +The foundation does not provide dynamic library loading, process isolation, network enforcement, cgroups, hardware attestation, authenticated Fabric transport, protected custody, or immutable remote verifier nodes. Those remain roadmap requirements before unattended operation on untrusted workloads. diff --git a/docs/decisions/0001-rust-provider-runtime.md b/docs/decisions/0001-rust-provider-runtime.md index b2fd2cf..b075f7d 100644 --- a/docs/decisions/0001-rust-provider-runtime.md +++ b/docs/decisions/0001-rust-provider-runtime.md @@ -60,11 +60,19 @@ This decision is enforced by repository artifacts rather than prose alone: - `mnel-provider-api` defines allocation-neutral ABI vocabulary. - `mnel-provider-sdk` provides a safe Rust authoring surface. - `mnel-provider-host` encodes native-language admission policy and snapshot reuse. +- `mnel-provider-host::placement` mirrors the backend-neutral CPU/CUDA/offload policy; + physical accelerator adapters remain outside the trusted ABI boundary. - `include/mnel_provider_v1.h` is the language-neutral ABI header. - `ProviderRuntimeManifest` mirrors the admission contract in the Python control plane. - `learned-provider-runtime-manifest.schema.json` makes the durable manifest testable. - CI runs Rust formatting, linting, and tests alongside the Python suite. +The first executable native baseline is a deterministic Rust HMM diagnostic provider. Its +host integration demonstrates persistent admission, identity-bound snapshot reuse, +bounded output normalization, timing measurements, and quarantine without changing the +v1 ABI. Sequential CPU offload is an optional external/backend capability, not a reason to +replace the Rust host with a Python daemon. + A future loader may not weaken these requirements. It must reject unsupported ABI versions, missing identities, unbounded queries, process-per-invocation providers, invalid tier/language combinations, or attempts to grant learned output evaluator diff --git a/examples/learned-providers/runtime-manifest.json b/examples/learned-providers/runtime-manifest.json index 8cb1d85..4577b86 100644 --- a/examples/learned-providers/runtime-manifest.json +++ b/examples/learned-providers/runtime-manifest.json @@ -10,7 +10,20 @@ "abi": "mnel-provider-c-abi/1", "persistent_host": true, "snapshot_transport": "identity-bound-compact-binary", - "weight_residency": "resident-on-admission" + "weight_residency": "resident-on-admission", + "placement": { + "execution_device": "auto", + "offload": "auto", + "precision": "auto", + "gpu_reserve_mib": 256, + "max_vram_mib": null, + "model_storage_bytes": 65536, + "workspace_bytes": 268435456, + "host_memory_budget_bytes": null, + "supports_sequential_cpu_offload": false, + "cpu_precisions": ["float32"], + "cuda_precisions": ["float32", "float16", "bfloat16"] + } }, "authority": "diagnostic-only", "verdict_semantics": "not-a-verdict" diff --git a/schemas/learned-provider-runtime-manifest.schema.json b/schemas/learned-provider-runtime-manifest.schema.json index e02521d..f6b709e 100644 --- a/schemas/learned-provider-runtime-manifest.schema.json +++ b/schemas/learned-provider-runtime-manifest.schema.json @@ -44,7 +44,47 @@ "abi": {"const": "mnel-provider-c-abi/1"}, "persistent_host": {"const": true}, "snapshot_transport": {"const": "identity-bound-compact-binary"}, - "weight_residency": {"const": "resident-on-admission"} + "weight_residency": {"const": "resident-on-admission"}, + "placement": { + "type": "object", + "additionalProperties": false, + "required": [ + "execution_device", + "offload", + "precision", + "gpu_reserve_mib", + "max_vram_mib", + "model_storage_bytes", + "workspace_bytes", + "host_memory_budget_bytes", + "supports_sequential_cpu_offload", + "cpu_precisions", + "cuda_precisions" + ], + "properties": { + "execution_device": {"enum": ["auto", "cpu", "cuda"]}, + "offload": {"enum": ["auto", "none", "sequential-cpu"]}, + "precision": {"enum": ["auto", "float32", "float16", "bfloat16"]}, + "gpu_reserve_mib": {"type": "integer", "minimum": 0}, + "max_vram_mib": {"type": ["integer", "null"], "minimum": 1}, + "model_storage_bytes": {"type": "integer", "minimum": 0}, + "workspace_bytes": {"type": "integer", "minimum": 0}, + "host_memory_budget_bytes": {"type": ["integer", "null"], "minimum": 1}, + "supports_sequential_cpu_offload": {"type": "boolean"}, + "cpu_precisions": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"enum": ["auto", "float32", "float16", "bfloat16"]} + }, + "cuda_precisions": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"enum": ["auto", "float32", "float16", "bfloat16"]} + } + } + } } }, "language_exception": { diff --git a/schemas/mnel-investigator-runtime.schema.json b/schemas/mnel-investigator-runtime.schema.json new file mode 100644 index 0000000..370a7bc --- /dev/null +++ b/schemas/mnel-investigator-runtime.schema.json @@ -0,0 +1,54 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/epi13/Machine-Native-Experimental-Learning/schemas/mnel-investigator-runtime.schema.json", + "title": "MNEL bounded investigator runtime records", + "oneOf": [ + {"$ref": "#/$defs/context"}, + {"$ref": "#/$defs/transaction"}, + {"$ref": "#/$defs/report"} + ], + "$defs": { + "context": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "snapshot_identity", "record_ids", "records", "encoded_bytes", "authority"], + "properties": { + "schema": {"const": "mnel-eligible-context/0.2"}, + "snapshot_identity": {"type": "string", "minLength": 1}, + "record_ids": {"type": "array", "items": {"type": "string"}, "uniqueItems": true}, + "records": {"type": "array", "items": {"type": "object"}}, + "encoded_bytes": {"type": "integer", "minimum": 0}, + "authority": {"const": "proposal-only"} + } + }, + "transaction": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "transaction_identity", "parent_candidate_id", "context_snapshot_identity", "workspace", "access", "proposal_only"], + "properties": { + "schema": {"const": "mnel-candidate-transaction/0.2"}, + "transaction_identity": {"type": "string", "minLength": 1}, + "parent_candidate_id": {"type": "string", "minLength": 1}, + "context_snapshot_identity": {"type": "string", "minLength": 1}, + "workspace": {"type": "string", "minLength": 1}, + "access": {"enum": ["read-only", "proposal"]}, + "proposal_only": {"const": true} + } + }, + "report": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "context_snapshot_identity", "packed_record_count", "provider_observation_count", "quarantined_provider_count", "proposal_count", "authority", "report_identity"], + "properties": { + "schema": {"const": "mnel-morning-report/0.2"}, + "context_snapshot_identity": {"type": "string", "minLength": 1}, + "packed_record_count": {"type": "integer", "minimum": 0}, + "provider_observation_count": {"type": "integer", "minimum": 0}, + "quarantined_provider_count": {"type": "integer", "minimum": 0}, + "proposal_count": {"type": "integer", "minimum": 0}, + "authority": {"const": "proposal-only"}, + "report_identity": {"type": "string", "minLength": 1} + } + } + } +} diff --git a/src/mnel/__init__.py b/src/mnel/__init__.py index 24d0d22..19d9f4a 100644 --- a/src/mnel/__init__.py +++ b/src/mnel/__init__.py @@ -21,22 +21,42 @@ ProviderRuntimeManifest, load_runtime_manifest, ) +from .placement import ( + AcceleratorDiagnostics, + ExecutionDevice, + ExecutionMode, + OffloadMode, + PlacementCapabilities, + PlacementDecision, + PlacementPolicy, + Precision, + decide_placement, +) __all__ = [ "DEFAULT_LEARNED_PROVIDER_REGISTRY", "EvidenceLedger", + "AcceleratorDiagnostics", "ExecutionTier", "HardGateEvaluator", "ImplementationLanguage", + "ExecutionDevice", + "ExecutionMode", "LearnedProviderDeclaration", "LearnedProviderObservation", "LearnedProviderQuery", "LearnedProviderRegistry", "NativeLanguageException", + "OffloadMode", + "PlacementCapabilities", + "PlacementDecision", + "PlacementPolicy", + "Precision", "ProviderRuntimeManifest", "RecursionGovernor", "VerifiedExperienceDistiller", "canonical_digest", + "decide_placement", "load_runtime_manifest", ] diff --git a/src/mnel/investigator_harness.py b/src/mnel/investigator_harness.py new file mode 100644 index 0000000..1a604f4 --- /dev/null +++ b/src/mnel/investigator_harness.py @@ -0,0 +1,214 @@ +"""Bounded local investigator context, workspace, and quarantine contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Any, Iterable + +from .core import Visibility, canonical_digest, canonical_json + + +class WorkspaceAccess(StrEnum): + READ_ONLY = "read-only" + PROPOSAL = "proposal" + + +class ContextPackingError(ValueError): + pass + + +@dataclass(frozen=True, slots=True) +class InvestigatorWorkspace: + path: str + access: WorkspaceAccess + + @classmethod + def read_only(cls, path: str | Path) -> "InvestigatorWorkspace": + return cls(str(Path(path).resolve()), WorkspaceAccess.READ_ONLY) + + @classmethod + def proposal(cls, path: str | Path) -> "InvestigatorWorkspace": + return cls(str(Path(path).resolve()), WorkspaceAccess.PROPOSAL) + + def assert_write_allowed(self) -> None: + if self.access is WorkspaceAccess.READ_ONLY: + raise PermissionError("read-only investigator workspace cannot be mutated") + + +@dataclass(frozen=True, slots=True) +class RuntimeIdentityEnvelope: + model_identity: str + quantization_identity: str + runtime_identity: str + prompt_identity: str + tool_schema_identity: str + + def __post_init__(self) -> None: + if not all( + item.strip() + for item in ( + self.model_identity, + self.quantization_identity, + self.runtime_identity, + self.prompt_identity, + self.tool_schema_identity, + ) + ): + raise ValueError("model, quantization, runtime, prompt, and tool identities are required") + + def to_dict(self) -> dict[str, str]: + return { + "model_identity": self.model_identity, + "quantization_identity": self.quantization_identity, + "runtime_identity": self.runtime_identity, + "prompt_identity": self.prompt_identity, + "tool_schema_identity": self.tool_schema_identity, + } + + +@dataclass(frozen=True, slots=True) +class PackedContext: + snapshot_identity: str + record_ids: tuple[str, ...] + records: tuple[dict[str, Any], ...] + encoded_bytes: int + + def to_dict(self) -> dict[str, Any]: + return { + "schema": "mnel-eligible-context/0.2", + "snapshot_identity": self.snapshot_identity, + "record_ids": list(self.record_ids), + "records": list(self.records), + "encoded_bytes": self.encoded_bytes, + "authority": "proposal-only", + } + + +def pack_eligible_context( + records: Iterable[dict[str, Any]], + *, + max_records: int = 32, + max_bytes: int = 256 * 1024, +) -> PackedContext: + """Pack visible records in stable identity order under an explicit byte ceiling.""" + + if max_records < 1 or max_bytes < 1: + raise ContextPackingError("context limits must be positive") + normalized: list[dict[str, Any]] = [] + for record in records: + record_id = record.get("record_id") or record.get("observation_id") or record.get("id") + visibility = record.get("visibility", Visibility.DEVELOPMENT.value) + if not isinstance(record_id, str) or not record_id.strip(): + raise ContextPackingError("every context record requires an identity") + if visibility in {Visibility.TRANSFER_HIDDEN.value, Visibility.FUTURE_FINAL.value}: + raise ContextPackingError(f"hidden record cannot enter investigator context: {record_id}") + normalized.append(dict(record)) + normalized.sort(key=lambda record: str(record.get("record_id") or record.get("observation_id") or record["id"])) + selected = normalized[:max_records] + encoded = canonical_json(selected) + if len(encoded) > max_bytes: + raise ContextPackingError("eligible context exceeds its explicit byte ceiling") + record_ids = tuple( + str(record.get("record_id") or record.get("observation_id") or record["id"]) + for record in selected + ) + return PackedContext( + snapshot_identity=canonical_digest( + {"record_ids": record_ids, "encoded": encoded.decode("utf-8")} + ), + record_ids=record_ids, + records=tuple(selected), + encoded_bytes=len(encoded), + ) + + +@dataclass(frozen=True, slots=True) +class CandidateTransaction: + transaction_identity: str + parent_candidate_id: str + context_snapshot_identity: str + workspace: str + access: WorkspaceAccess = WorkspaceAccess.PROPOSAL + proposal_only: bool = True + + @classmethod + def create( + cls, + *, + parent_candidate_id: str, + context_snapshot_identity: str, + workspace: str | Path, + ) -> "CandidateTransaction": + if not parent_candidate_id.strip() or not context_snapshot_identity.strip(): + raise ValueError("candidate transactions require parent and context identities") + value = { + "parent_candidate_id": parent_candidate_id, + "context_snapshot_identity": context_snapshot_identity, + "workspace": str(Path(workspace).resolve()), + "access": WorkspaceAccess.PROPOSAL.value, + "proposal_only": True, + } + return cls( + transaction_identity=canonical_digest(value), + parent_candidate_id=parent_candidate_id, + context_snapshot_identity=context_snapshot_identity, + workspace=value["workspace"], + ) + + def to_dict(self) -> dict[str, Any]: + return { + "schema": "mnel-candidate-transaction/0.2", + "transaction_identity": self.transaction_identity, + "parent_candidate_id": self.parent_candidate_id, + "context_snapshot_identity": self.context_snapshot_identity, + "workspace": self.workspace, + "access": self.access.value, + "proposal_only": self.proposal_only, + } + + +@dataclass(frozen=True, slots=True) +class QuarantineEntry: + provider_id: str + reason: str + evidence_identity: str + + +class QuarantineQueue: + """Explicit queue for failures; enqueueing never changes evaluator state.""" + + def __init__(self) -> None: + self._entries: list[QuarantineEntry] = [] + + def enqueue(self, entry: QuarantineEntry) -> None: + if not entry.provider_id.strip() or not entry.reason.strip() or not entry.evidence_identity.strip(): + raise ValueError("quarantine entries require provider, reason, and evidence identities") + self._entries.append(entry) + + def list(self) -> tuple[QuarantineEntry, ...]: + return tuple(sorted(self._entries, key=lambda entry: (entry.provider_id, entry.evidence_identity))) + + +@dataclass(frozen=True, slots=True) +class MorningReport: + context_snapshot_identity: str + packed_record_count: int + provider_observation_count: int + quarantined_provider_count: int + proposal_count: int + authority: str = "proposal-only" + + def to_dict(self) -> dict[str, Any]: + value = { + "schema": "mnel-morning-report/0.2", + "context_snapshot_identity": self.context_snapshot_identity, + "packed_record_count": self.packed_record_count, + "provider_observation_count": self.provider_observation_count, + "quarantined_provider_count": self.quarantined_provider_count, + "proposal_count": self.proposal_count, + "authority": self.authority, + } + value["report_identity"] = canonical_digest(value) + return value diff --git a/src/mnel/placement.py b/src/mnel/placement.py new file mode 100644 index 0000000..319e687 --- /dev/null +++ b/src/mnel/placement.py @@ -0,0 +1,316 @@ +"""Backend-neutral accelerator placement policy for learned providers. + +The policy deliberately does not import Torch, Accelerate, or a vendor runtime. A +backend supplies capability diagnostics and an adapter applies the returned decision. +This keeps deterministic policy tests cheap and keeps the Rust host architecture +independent from any one accelerator stack. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from enum import StrEnum +from typing import Any + +MIB = 1024 * 1024 +DEFAULT_GPU_RESERVE_MIB = 256 +DEFAULT_WORKSPACE_MIB = 256 + + +class PlacementError(RuntimeError): + """The requested placement cannot be satisfied safely.""" + + +class ExecutionDevice(StrEnum): + AUTO = "auto" + CPU = "cpu" + CUDA = "cuda" + + +class OffloadMode(StrEnum): + AUTO = "auto" + NONE = "none" + SEQUENTIAL_CPU = "sequential-cpu" + + +class Precision(StrEnum): + AUTO = "auto" + FLOAT32 = "float32" + FLOAT16 = "float16" + BFLOAT16 = "bfloat16" + + +class ExecutionMode(StrEnum): + CPU = "cpu" + FULL_CUDA = "full-cuda" + SEQUENTIAL_CPU_OFFLOAD = "sequential-cpu-offload" + + +@dataclass(frozen=True, slots=True) +class PlacementPolicy: + execution_device: ExecutionDevice = ExecutionDevice.AUTO + offload: OffloadMode = OffloadMode.AUTO + precision: Precision = Precision.AUTO + gpu_reserve_mib: int = DEFAULT_GPU_RESERVE_MIB + max_vram_mib: int | None = None + model_storage_bytes: int = 0 + workspace_bytes: int = DEFAULT_WORKSPACE_MIB * MIB + host_memory_budget_bytes: int | None = None + + def validate(self) -> None: + if self.execution_device is ExecutionDevice.CPU and self.offload is OffloadMode.SEQUENTIAL_CPU: + raise PlacementError("sequential CPU offload requires auto or cuda execution device") + if self.gpu_reserve_mib < 0: + raise PlacementError("GPU reserve cannot be negative") + if self.max_vram_mib is not None and self.max_vram_mib < 1: + raise PlacementError("maximum VRAM must be positive") + if self.model_storage_bytes < 0 or self.workspace_bytes < 0: + raise PlacementError("model and workspace estimates cannot be negative") + if self.host_memory_budget_bytes is not None and self.host_memory_budget_bytes < 1: + raise PlacementError("host memory budget must be positive") + + +@dataclass(frozen=True, slots=True) +class PlacementCapabilities: + supports_sequential_cpu_offload: bool = False + cpu_precisions: tuple[Precision, ...] = (Precision.FLOAT32,) + cuda_precisions: tuple[Precision, ...] = ( + Precision.FLOAT32, + Precision.FLOAT16, + Precision.BFLOAT16, + ) + + def validate(self) -> None: + if not self.cpu_precisions: + raise PlacementError("backend must declare at least one CPU precision") + if not self.cuda_precisions: + raise PlacementError("backend must declare at least one CUDA precision") + + +@dataclass(frozen=True, slots=True) +class AcceleratorDiagnostics: + accelerator_available: bool = False + execution_probe_succeeded: bool = False + free_vram_bytes: int | None = None + accelerator_identity: str | None = None + float16_probe_succeeded: bool | None = None + bfloat16_probe_succeeded: bool | None = None + probe_error: str | None = None + + def as_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True, slots=True) +class PlacementDecision: + execution_mode: ExecutionMode + execution_device: ExecutionDevice + offload: OffloadMode + precision: Precision + reason: str + configured_gpu_reserve_bytes: int + configured_max_vram_bytes: int | None + effective_gpu_budget_bytes: int + estimated_model_bytes: int + estimated_workspace_bytes: int + full_cuda_required_bytes: int + host_memory_required_bytes: int + sequential_offload_supported: bool + + def as_dict(self) -> dict[str, Any]: + value = asdict(self) + for key in ("execution_mode", "execution_device", "offload", "precision"): + value[key] = value[key].value + return value + + +def effective_gpu_budget_bytes( + free_vram_bytes: int | None, + reserve_mib: int, + max_vram_mib: int | None, +) -> int: + """Return free VRAM remaining after the operator reserve and optional cap.""" + + if free_vram_bytes is None: + return 0 + capped = free_vram_bytes + if max_vram_mib is not None: + capped = min(capped, max_vram_mib * MIB) + return max(0, capped - reserve_mib * MIB) + + +def _precision_bytes(model_storage_bytes: int, precision: Precision) -> int: + if precision in {Precision.FLOAT16, Precision.BFLOAT16}: + return (model_storage_bytes + 1) // 2 + return model_storage_bytes + + +def _choose_cuda_precision( + policy: PlacementPolicy, + diagnostics: AcceleratorDiagnostics, + capabilities: PlacementCapabilities, +) -> Precision: + if policy.precision is not Precision.AUTO: + if policy.precision not in capabilities.cuda_precisions: + raise PlacementError(f"CUDA backend does not declare {policy.precision.value} support") + if policy.precision is Precision.FLOAT16 and diagnostics.float16_probe_succeeded is not True: + raise PlacementError("float16 was requested but its execution probe failed") + if policy.precision is Precision.BFLOAT16 and diagnostics.bfloat16_probe_succeeded is not True: + raise PlacementError("bfloat16 was requested but its execution probe failed") + return policy.precision + if ( + Precision.BFLOAT16 in capabilities.cuda_precisions + and diagnostics.bfloat16_probe_succeeded is True + ): + return Precision.BFLOAT16 + if Precision.FLOAT16 in capabilities.cuda_precisions and diagnostics.float16_probe_succeeded is True: + return Precision.FLOAT16 + if Precision.FLOAT32 in capabilities.cuda_precisions: + return Precision.FLOAT32 + raise PlacementError("CUDA backend has no usable precision") + + +def decide_placement( + policy: PlacementPolicy, + diagnostics: AcceleratorDiagnostics, + capabilities: PlacementCapabilities, +) -> PlacementDecision: + """Choose CPU, full CUDA, or true sequential CPU offload deterministically.""" + + policy.validate() + capabilities.validate() + model_bytes = policy.model_storage_bytes + host_budget = policy.host_memory_budget_bytes + if host_budget is not None and model_bytes > host_budget: + raise PlacementError("provider model exceeds the host/system-memory budget") + + reserve_bytes = policy.gpu_reserve_mib * MIB + maximum_bytes = policy.max_vram_mib * MIB if policy.max_vram_mib is not None else None + gpu_budget = effective_gpu_budget_bytes( + diagnostics.free_vram_bytes, policy.gpu_reserve_mib, policy.max_vram_mib + ) + cuda_usable = diagnostics.accelerator_available and diagnostics.execution_probe_succeeded + + def decision( + mode: ExecutionMode, + precision: Precision, + reason: str, + ) -> PlacementDecision: + estimated_model = _precision_bytes(model_bytes, precision) + return PlacementDecision( + execution_mode=mode, + execution_device=( + ExecutionDevice.CUDA if mode is not ExecutionMode.CPU else ExecutionDevice.CPU + ), + offload=( + OffloadMode.SEQUENTIAL_CPU + if mode is ExecutionMode.SEQUENTIAL_CPU_OFFLOAD + else OffloadMode.NONE + ), + precision=precision, + reason=reason, + configured_gpu_reserve_bytes=reserve_bytes, + configured_max_vram_bytes=maximum_bytes, + effective_gpu_budget_bytes=gpu_budget, + estimated_model_bytes=estimated_model, + estimated_workspace_bytes=policy.workspace_bytes, + full_cuda_required_bytes=estimated_model + policy.workspace_bytes, + host_memory_required_bytes=model_bytes, + sequential_offload_supported=capabilities.supports_sequential_cpu_offload, + ) + + if policy.execution_device is ExecutionDevice.CPU: + if policy.precision is Precision.AUTO: + precision = Precision.FLOAT32 + else: + precision = policy.precision + if precision not in capabilities.cpu_precisions: + raise PlacementError(f"CPU backend does not declare {precision.value} support") + return decision(ExecutionMode.CPU, precision, "CPU was explicitly requested") + + if not cuda_usable: + reason = diagnostics.probe_error or "accelerator execution probe failed" + if ( + policy.execution_device is ExecutionDevice.CUDA + or policy.offload is OffloadMode.SEQUENTIAL_CPU + ): + raise PlacementError(f"CUDA execution is unusable: {reason}") + return decision(ExecutionMode.CPU, Precision.FLOAT32, f"AUTO selected CPU because {reason}") + + precision = _choose_cuda_precision(policy, diagnostics, capabilities) + required = _precision_bytes(model_bytes, precision) + policy.workspace_bytes + fits = required <= gpu_budget + + if policy.offload is OffloadMode.SEQUENTIAL_CPU: + if not capabilities.supports_sequential_cpu_offload: + raise PlacementError("sequential CPU offload is unsupported by this provider") + return decision( + ExecutionMode.SEQUENTIAL_CPU_OFFLOAD, + precision, + "sequential CPU offload was explicitly requested", + ) + + if policy.offload is OffloadMode.NONE: + if not fits: + if policy.execution_device is ExecutionDevice.AUTO: + return decision( + ExecutionMode.CPU, + Precision.FLOAT32, + "AUTO selected CPU because full CUDA exceeds the effective budget and offload is disabled", + ) + raise PlacementError("full CUDA exceeds the effective GPU budget") + return decision( + ExecutionMode.FULL_CUDA, + precision, + "full CUDA was explicitly requested" if policy.execution_device is ExecutionDevice.CUDA else "full CUDA fits the effective GPU budget", + ) + + if fits: + return decision(ExecutionMode.FULL_CUDA, precision, "full CUDA fits the effective GPU budget") + if capabilities.supports_sequential_cpu_offload: + return decision( + ExecutionMode.SEQUENTIAL_CPU_OFFLOAD, + precision, + "full CUDA exceeds the effective GPU budget; using CPU-backed sequential execution", + ) + if policy.execution_device is ExecutionDevice.AUTO: + return decision( + ExecutionMode.CPU, + Precision.FLOAT32, + "AUTO selected CPU because full CUDA exceeds budget and provider offload is unsupported", + ) + raise PlacementError( + "full CUDA exceeds the effective GPU budget and sequential offload is unsupported" + ) + + +def fallback_policy_after_oom( + policy: PlacementPolicy, + current_mode: ExecutionMode, + capabilities: PlacementCapabilities, +) -> PlacementPolicy | None: + """Return one bounded AUTO retry policy; explicit operator choices never retry.""" + + if policy.execution_device is not ExecutionDevice.AUTO: + return None + if current_mode is ExecutionMode.FULL_CUDA and capabilities.supports_sequential_cpu_offload: + return PlacementPolicy( + execution_device=ExecutionDevice.AUTO, + offload=OffloadMode.SEQUENTIAL_CPU, + precision=policy.precision, + gpu_reserve_mib=policy.gpu_reserve_mib, + max_vram_mib=policy.max_vram_mib, + model_storage_bytes=policy.model_storage_bytes, + workspace_bytes=policy.workspace_bytes, + host_memory_budget_bytes=policy.host_memory_budget_bytes, + ) + if current_mode in {ExecutionMode.FULL_CUDA, ExecutionMode.SEQUENTIAL_CPU_OFFLOAD}: + return PlacementPolicy( + execution_device=ExecutionDevice.CPU, + offload=OffloadMode.NONE, + precision=Precision.FLOAT32, + model_storage_bytes=policy.model_storage_bytes, + workspace_bytes=policy.workspace_bytes, + host_memory_budget_bytes=policy.host_memory_budget_bytes, + ) + return None diff --git a/src/mnel/provider_runtime.py b/src/mnel/provider_runtime.py index 0533d66..dd2e18e 100644 --- a/src/mnel/provider_runtime.py +++ b/src/mnel/provider_runtime.py @@ -13,6 +13,7 @@ from typing import Any from .learned_providers import LearnedProviderDeclaration +from .placement import PlacementCapabilities, PlacementPolicy PROVIDER_ABI_V1 = "mnel-provider-c-abi/1" RUNTIME_MANIFEST_SCHEMA = "mnel-learned-provider-runtime-manifest/0.1" @@ -71,6 +72,8 @@ class ProviderRuntimeManifest: snapshot_transport: str = "identity-bound-compact-binary" weight_residency: str = "resident-on-admission" language_exception: NativeLanguageException | None = None + placement_policy: PlacementPolicy = field(default_factory=PlacementPolicy) + placement_capabilities: PlacementCapabilities = field(default_factory=PlacementCapabilities) authority: str = field(default="diagnostic-only", init=False) verdict_semantics: str = field(default="not-a-verdict", init=False) @@ -90,7 +93,14 @@ def __post_init__(self) -> None: if self.snapshot_transport != "identity-bound-compact-binary": raise ValueError("hot-path snapshots must use identity-bound compact binary transport") if self.weight_residency != "resident-on-admission": - raise ValueError("admitted provider weights must remain resident") + raise ValueError( + "admitted provider weights must remain resident even when physical placement changes" + ) + try: + self.placement_policy.validate() + self.placement_capabilities.validate() + except RuntimeError as error: + raise ValueError(str(error)) from error if self.execution_tier is ExecutionTier.NATIVE_TRUSTED: if ( self.implementation_language is not ImplementationLanguage.RUST @@ -100,7 +110,10 @@ def __post_init__(self) -> None: elif self.execution_tier is ExecutionTier.WASM_QUARANTINED: if self.implementation_language is not ImplementationLanguage.WASM: raise ValueError("wasm-quarantined tier requires a WASM provider") - elif self.language_exception is not None: + if self.language_exception is not None and not ( + self.execution_tier is ExecutionTier.NATIVE_TRUSTED + and self.implementation_language is not ImplementationLanguage.RUST + ): raise ValueError("language exceptions apply only to non-Rust native providers") def validate_declaration(self, declaration: LearnedProviderDeclaration) -> None: @@ -127,6 +140,19 @@ def to_dict(self) -> dict[str, object]: "persistent_host": self.persistent_host, "snapshot_transport": self.snapshot_transport, "weight_residency": self.weight_residency, + "placement": { + "execution_device": self.placement_policy.execution_device.value, + "offload": self.placement_policy.offload.value, + "precision": self.placement_policy.precision.value, + "gpu_reserve_mib": self.placement_policy.gpu_reserve_mib, + "max_vram_mib": self.placement_policy.max_vram_mib, + "model_storage_bytes": self.placement_policy.model_storage_bytes, + "workspace_bytes": self.placement_policy.workspace_bytes, + "host_memory_budget_bytes": self.placement_policy.host_memory_budget_bytes, + "supports_sequential_cpu_offload": self.placement_capabilities.supports_sequential_cpu_offload, + "cpu_precisions": [item.value for item in self.placement_capabilities.cpu_precisions], + "cuda_precisions": [item.value for item in self.placement_capabilities.cuda_precisions], + }, }, "authority": self.authority, "verdict_semantics": self.verdict_semantics, @@ -152,6 +178,51 @@ def from_dict(cls, value: dict[str, Any]) -> "ProviderRuntimeManifest": benchmark_evidence_ids=tuple(exception_value.get("benchmark_evidence_ids", ())), threat_review_id=str(exception_value.get("threat_review_id", "")), ) + placement_value = runtime.get("placement", {}) + if not isinstance(placement_value, dict): + raise ValueError("runtime placement must be an object") + from .placement import ( + ExecutionDevice, + OffloadMode, + Precision, + ) + + placement_policy = PlacementPolicy( + execution_device=ExecutionDevice( + placement_value.get("execution_device", ExecutionDevice.AUTO.value) + ), + offload=OffloadMode(placement_value.get("offload", OffloadMode.AUTO.value)), + precision=Precision(placement_value.get("precision", Precision.AUTO.value)), + gpu_reserve_mib=int(placement_value.get("gpu_reserve_mib", 256)), + max_vram_mib=( + None + if placement_value.get("max_vram_mib") is None + else int(placement_value["max_vram_mib"]) + ), + model_storage_bytes=int(placement_value.get("model_storage_bytes", 0)), + workspace_bytes=int(placement_value.get("workspace_bytes", 256 * 1024 * 1024)), + host_memory_budget_bytes=( + None + if placement_value.get("host_memory_budget_bytes") is None + else int(placement_value["host_memory_budget_bytes"]) + ), + ) + placement_capabilities = PlacementCapabilities( + supports_sequential_cpu_offload=bool( + placement_value.get("supports_sequential_cpu_offload", False) + ), + cpu_precisions=tuple( + Precision(item) + for item in placement_value.get("cpu_precisions", [Precision.FLOAT32.value]) + ), + cuda_precisions=tuple( + Precision(item) + for item in placement_value.get( + "cuda_precisions", + [Precision.FLOAT32.value, Precision.FLOAT16.value, Precision.BFLOAT16.value], + ) + ), + ) manifest = cls( provider_id=str(value.get("provider_id", "")), provider_version=str(value.get("provider_version", "")), @@ -166,6 +237,8 @@ def from_dict(cls, value: dict[str, Any]) -> "ProviderRuntimeManifest": snapshot_transport=str(runtime.get("snapshot_transport", "")), weight_residency=str(runtime.get("weight_residency", "")), language_exception=exception, + placement_policy=placement_policy, + placement_capabilities=placement_capabilities, ) if value.get("authority") != manifest.authority: raise ValueError("runtime manifest authority must be diagnostic-only") diff --git a/src/mnel/torch_runtime.py b/src/mnel/torch_runtime.py new file mode 100644 index 0000000..3d32ff6 --- /dev/null +++ b/src/mnel/torch_runtime.py @@ -0,0 +1,170 @@ +"""Optional Torch adapter for the generic MNEL placement policy. + +Torch is intentionally imported only through a caller-supplied module. Core MNEL +installation and tests remain dependency-free. A sequential-offload result is marked +verified only when inference completed and observed module/parameter residency supports +that claim. +""" + +from __future__ import annotations + +import gc +from collections.abc import Callable +from typing import Any + +from .placement import ( + AcceleratorDiagnostics, + ExecutionMode, + PlacementDecision, + PlacementError, + Precision, +) + + +def _error_text(error: BaseException) -> str: + return f"{type(error).__name__}: {error}"[:512] + + +def _probe_dtype(torch: Any, dtype: Any) -> tuple[bool, str | None]: + try: + left = torch.ones((32, 32), device="cuda", dtype=dtype) + right = torch.ones((32, 32), device="cuda", dtype=dtype) + output = left @ right + torch.cuda.synchronize(0) + success = bool(torch.isfinite(output).all().item()) + return success, None if success else "CUDA dtype probe returned non-finite values" + except Exception as error: # backend exceptions are part of the diagnostic surface + return False, _error_text(error) + + +def collect_torch_diagnostics(torch: Any) -> AcceleratorDiagnostics: + """Require a real CUDA kernel probe instead of trusting discovery alone.""" + + available = bool(torch.cuda.is_available()) + if not available: + return AcceleratorDiagnostics(probe_error="installed Torch build has no CUDA runtime") + try: + free, _total = torch.cuda.mem_get_info(0) + major, minor = torch.cuda.get_device_capability(0) + float32_ok, float32_error = _probe_dtype(torch, torch.float32) + float16_ok = _probe_dtype(torch, torch.float16)[0] if float32_ok else False + bf16_reported = bool(getattr(torch.cuda, "is_bf16_supported", lambda: False)()) + bf16_ok = _probe_dtype(torch, torch.bfloat16)[0] if float32_ok and bf16_reported else False + return AcceleratorDiagnostics( + accelerator_available=True, + execution_probe_succeeded=float32_ok, + free_vram_bytes=int(free), + accelerator_identity=f"{torch.cuda.get_device_name(0)} sm_{major}{minor}", + float16_probe_succeeded=float16_ok, + bfloat16_probe_succeeded=bf16_ok, + probe_error=float32_error, + ) + except Exception as error: + return AcceleratorDiagnostics( + accelerator_available=True, + execution_probe_succeeded=False, + probe_error=_error_text(error), + ) + + +def parameter_storage_bytes(model: Any) -> int: + return int(sum(parameter.numel() * parameter.element_size() for parameter in model.parameters())) + + +def _torch_dtype(torch: Any, precision: Precision) -> Any: + try: + return { + Precision.FLOAT32: torch.float32, + Precision.FLOAT16: torch.float16, + Precision.BFLOAT16: torch.bfloat16, + }[precision] + except KeyError as error: + raise PlacementError(f"unsupported Torch precision: {precision.value}") from error + + +def apply_torch_placement( + model: Any, + torch: Any, + decision: PlacementDecision, + *, + cpu_offload_fn: Callable[..., Any] | None = None, + preload_module_classes: tuple[str, ...] = (), +) -> Any: + """Apply a decision; sequential mode uses Accelerate's CPU-backed hooks.""" + + dtype = _torch_dtype(torch, decision.precision) + model.to(device="cpu", dtype=dtype) + if decision.execution_mode is ExecutionMode.CPU: + return model + if decision.execution_mode is ExecutionMode.FULL_CUDA: + return model.to("cuda") + if decision.execution_mode is not ExecutionMode.SEQUENTIAL_CPU_OFFLOAD: + raise PlacementError(f"unsupported execution mode: {decision.execution_mode.value}") + if cpu_offload_fn is None: + try: + from accelerate import cpu_offload as cpu_offload_fn + except ImportError as error: + raise PlacementError( + "sequential CPU offload requires Accelerate in the optional provider environment" + ) from error + return cpu_offload_fn( + model, + execution_device=torch.device("cuda"), + offload_buffers=True, + preload_module_classes=list(preload_module_classes) or None, + ) + + +def offload_evidence(model: Any, *, inference_completed: bool) -> dict[str, Any]: + """Return observed evidence; requested offload alone never sets verified true.""" + + hooks = sum(int(hasattr(module, "_hf_hook")) for module in model.modules()) + meta_bytes = 0 + cuda_bytes = 0 + devices: dict[str, int] = {} + for parameter in model.parameters(): + device = str(parameter.device) + devices[device] = devices.get(device, 0) + 1 + size = int(parameter.numel() * parameter.element_size()) + if device == "meta": + meta_bytes += size + elif device.startswith("cuda"): + cuda_bytes += size + return { + "sequential_offload_hook_count": hooks, + "offloaded_meta_parameter_bytes": meta_bytes, + "persistent_cuda_parameter_bytes": cuda_bytes, + "parameter_device_counts": dict(sorted(devices.items())), + "inference_completed": inference_completed, + "sequential_offload_verified": bool( + inference_completed and hooks > 0 and meta_bytes > 0 and cuda_bytes == 0 + ), + } + + +def cuda_memory_snapshot(torch: Any) -> dict[str, int | None]: + if not bool(torch.cuda.is_available()): + return {"cuda_allocated_bytes": None, "cuda_reserved_bytes": None} + return { + "cuda_allocated_bytes": int(torch.cuda.max_memory_allocated(0)), + "cuda_reserved_bytes": int(torch.cuda.max_memory_reserved(0)), + } + + +def reset_cuda_peaks(torch: Any) -> None: + if bool(torch.cuda.is_available()): + torch.cuda.synchronize(0) + torch.cuda.reset_peak_memory_stats(0) + + +def restore_model_to_cpu(model: Any, torch: Any, *, remove_hooks_fn: Callable[[Any], Any] | None = None) -> Any: + if any(hasattr(module, "_hf_hook") for module in model.modules()): + if remove_hooks_fn is None: + from accelerate.hooks import remove_hook_from_submodules as remove_hooks_fn + + remove_hooks_fn(model) + model.to("cpu") + gc.collect() + if bool(torch.cuda.is_available()): + torch.cuda.empty_cache() + return model diff --git a/tests/test_investigator_harness.py b/tests/test_investigator_harness.py new file mode 100644 index 0000000..54fa6e9 --- /dev/null +++ b/tests/test_investigator_harness.py @@ -0,0 +1,73 @@ +import unittest + +from mnel.investigator_harness import ( + CandidateTransaction, + ContextPackingError, + MorningReport, + QuarantineEntry, + QuarantineQueue, + RuntimeIdentityEnvelope, + InvestigatorWorkspace, + WorkspaceAccess, + pack_eligible_context, +) + + +class InvestigatorHarnessTests(unittest.TestCase): + def test_context_packing_is_stable_and_bounded(self) -> None: + context = pack_eligible_context( + [ + {"record_id": "b", "visibility": "development-visible", "value": 2}, + {"record_id": "a", "visibility": "selection-observed-not-repairable", "value": 1}, + ], + max_records=2, + max_bytes=1024, + ) + self.assertEqual(context.record_ids, ("a", "b")) + self.assertEqual(context.to_dict()["authority"], "proposal-only") + self.assertEqual( + context.snapshot_identity, + pack_eligible_context( + [{"record_id": "b", "visibility": "development-visible", "value": 2}, + {"record_id": "a", "visibility": "selection-observed-not-repairable", "value": 1}], + max_records=2, + max_bytes=1024, + ).snapshot_identity, + ) + + def test_hidden_context_and_oversized_context_are_rejected(self) -> None: + with self.assertRaises(ContextPackingError): + pack_eligible_context([{"record_id": "hidden", "visibility": "transfer-hidden"}]) + with self.assertRaises(ContextPackingError): + pack_eligible_context([{"record_id": "large", "text": "x" * 100}], max_bytes=8) + + def test_transaction_identity_and_runtime_lineage_are_explicit(self) -> None: + envelope = RuntimeIdentityEnvelope("model", "quant", "runtime", "prompt", "tools") + transaction = CandidateTransaction.create( + parent_candidate_id="candidate-parent", + context_snapshot_identity="sha256:context", + workspace="build/proposal", + ) + self.assertEqual(transaction.access, WorkspaceAccess.PROPOSAL) + self.assertTrue(transaction.proposal_only) + self.assertEqual(envelope.to_dict()["runtime_identity"], "runtime") + self.assertEqual(transaction.transaction_identity, CandidateTransaction.create( + parent_candidate_id="candidate-parent", + context_snapshot_identity="sha256:context", + workspace="build/proposal", + ).transaction_identity) + with self.assertRaises(PermissionError): + InvestigatorWorkspace.read_only("build/read-only").assert_write_allowed() + InvestigatorWorkspace.proposal("build/proposal").assert_write_allowed() + + def test_quarantine_and_report_remain_observable(self) -> None: + queue = QuarantineQueue() + queue.enqueue(QuarantineEntry("provider-a", "runtime failure", "sha256:evidence")) + self.assertEqual(queue.list()[0].provider_id, "provider-a") + report = MorningReport("sha256:context", 2, 3, 1, 1) + self.assertEqual(report.to_dict()["authority"], "proposal-only") + self.assertNotIn("verdict", report.to_dict()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_placement.py b/tests/test_placement.py new file mode 100644 index 0000000..f7b36ad --- /dev/null +++ b/tests/test_placement.py @@ -0,0 +1,144 @@ +import unittest + +from mnel.placement import ( + MIB, + AcceleratorDiagnostics, + ExecutionDevice, + ExecutionMode, + OffloadMode, + PlacementCapabilities, + PlacementError, + PlacementPolicy, + Precision, + decide_placement, + effective_gpu_budget_bytes, + fallback_policy_after_oom, +) + + +def cuda(*, free_mib: int = 4096, float16: bool = True, bfloat16: bool = False) -> AcceleratorDiagnostics: + return AcceleratorDiagnostics( + accelerator_available=True, + execution_probe_succeeded=True, + free_vram_bytes=free_mib * MIB, + accelerator_identity="fake-cuda-0", + float16_probe_succeeded=float16, + bfloat16_probe_succeeded=bfloat16, + ) + + +class PlacementPolicyTests(unittest.TestCase): + def test_reserve_and_cap_math(self) -> None: + self.assertEqual( + effective_gpu_budget_bytes(2048 * MIB, reserve_mib=256, max_vram_mib=1024), + 768 * MIB, + ) + self.assertEqual(effective_gpu_budget_bytes(None, 0, None), 0) + + def test_cpu_is_selected_without_accelerator(self) -> None: + decision = decide_placement( + PlacementPolicy(model_storage_bytes=64 * MIB), + AcceleratorDiagnostics(probe_error="no CUDA runtime"), + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + self.assertEqual(decision.execution_mode, ExecutionMode.CPU) + self.assertEqual(decision.execution_device, ExecutionDevice.CPU) + self.assertEqual(decision.precision, Precision.FLOAT32) + + def test_auto_selects_full_cuda_when_model_and_workspace_fit(self) -> None: + decision = decide_placement( + PlacementPolicy(model_storage_bytes=256 * MIB, workspace_bytes=128 * MIB), + cuda(free_mib=1024), + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + self.assertEqual(decision.execution_mode, ExecutionMode.FULL_CUDA) + self.assertEqual(decision.precision, Precision.FLOAT16) + + def test_auto_selects_sequential_offload_when_full_model_does_not_fit(self) -> None: + decision = decide_placement( + PlacementPolicy(model_storage_bytes=2048 * MIB, workspace_bytes=512 * MIB), + cuda(free_mib=1024), + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + self.assertEqual(decision.execution_mode, ExecutionMode.SEQUENTIAL_CPU_OFFLOAD) + self.assertEqual(decision.offload, OffloadMode.SEQUENTIAL_CPU) + + def test_auto_falls_back_to_cpu_when_offload_is_unavailable(self) -> None: + decision = decide_placement( + PlacementPolicy(model_storage_bytes=2048 * MIB, workspace_bytes=512 * MIB), + cuda(free_mib=1024), + PlacementCapabilities(supports_sequential_cpu_offload=False), + ) + self.assertEqual(decision.execution_mode, ExecutionMode.CPU) + + def test_explicit_cuda_does_not_silently_fallback(self) -> None: + with self.assertRaisesRegex(PlacementError, "exceeds"): + decide_placement( + PlacementPolicy( + execution_device=ExecutionDevice.CUDA, + offload=OffloadMode.NONE, + model_storage_bytes=2048 * MIB, + workspace_bytes=512 * MIB, + ), + cuda(free_mib=1024), + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + + def test_explicit_offload_requires_real_cuda_and_capability(self) -> None: + with self.assertRaisesRegex(PlacementError, "unsupported"): + decide_placement( + PlacementPolicy( + execution_device=ExecutionDevice.CUDA, + offload=OffloadMode.SEQUENTIAL_CPU, + ), + cuda(), + PlacementCapabilities(supports_sequential_cpu_offload=False), + ) + with self.assertRaisesRegex(PlacementError, "unusable"): + decide_placement( + PlacementPolicy(offload=OffloadMode.SEQUENTIAL_CPU), + AcceleratorDiagnostics(probe_error="kernel probe failed"), + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + + def test_dtype_probe_is_required_for_explicit_precision(self) -> None: + with self.assertRaisesRegex(PlacementError, "float16"): + decide_placement( + PlacementPolicy(precision=Precision.FLOAT16), + cuda(float16=False), + PlacementCapabilities(), + ) + + def test_oom_recovery_is_bounded_and_auto_only(self) -> None: + policy = PlacementPolicy(model_storage_bytes=512) + retry = fallback_policy_after_oom( + policy, ExecutionMode.FULL_CUDA, PlacementCapabilities(supports_sequential_cpu_offload=True) + ) + self.assertIsNotNone(retry) + self.assertEqual(retry.offload, OffloadMode.SEQUENTIAL_CPU) + final = fallback_policy_after_oom( + retry, + ExecutionMode.SEQUENTIAL_CPU_OFFLOAD, + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + self.assertIsNotNone(final) + self.assertEqual(final.execution_device, ExecutionDevice.CPU) + self.assertIsNone( + fallback_policy_after_oom( + PlacementPolicy(execution_device=ExecutionDevice.CUDA), + ExecutionMode.FULL_CUDA, + PlacementCapabilities(supports_sequential_cpu_offload=True), + ) + ) + + def test_host_memory_budget_is_enforced(self) -> None: + with self.assertRaisesRegex(PlacementError, "host/system-memory"): + decide_placement( + PlacementPolicy(model_storage_bytes=1024, host_memory_budget_bytes=512), + AcceleratorDiagnostics(), + PlacementCapabilities(), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_provider_runtime.py b/tests/test_provider_runtime.py index c081260..02295b6 100644 --- a/tests/test_provider_runtime.py +++ b/tests/test_provider_runtime.py @@ -10,6 +10,7 @@ ProviderRuntimeManifest, load_runtime_manifest, ) +from mnel.placement import ExecutionDevice, OffloadMode, PlacementPolicy, Precision class ProviderRuntimePolicyTests(unittest.TestCase): @@ -51,6 +52,24 @@ def test_specialized_native_language_requires_exception_evidence(self) -> None: ) self.assertEqual(manifest.execution_tier, ExecutionTier.NATIVE_TRUSTED) + def test_placement_metadata_round_trips_without_changing_abi(self) -> None: + manifest = self.manifest( + placement_policy=PlacementPolicy( + execution_device=ExecutionDevice.AUTO, + offload=OffloadMode.SEQUENTIAL_CPU, + precision=Precision.FLOAT16, + gpu_reserve_mib=512, + max_vram_mib=4096, + model_storage_bytes=1024, + workspace_bytes=2048, + ), + ) + raw = manifest.to_dict() + self.assertEqual(raw["runtime"]["abi"], "mnel-provider-c-abi/1") + restored = ProviderRuntimeManifest.from_dict(raw) + self.assertEqual(restored.placement_policy.offload, OffloadMode.SEQUENTIAL_CPU) + self.assertEqual(restored.placement_policy.precision, Precision.FLOAT16) + def test_example_manifest_round_trips(self) -> None: path = Path("examples/learned-providers/runtime-manifest.json") raw = json.loads(path.read_text(encoding="utf-8")) diff --git a/tests/test_torch_runtime.py b/tests/test_torch_runtime.py new file mode 100644 index 0000000..99afaae --- /dev/null +++ b/tests/test_torch_runtime.py @@ -0,0 +1,49 @@ +import unittest + +from mnel.torch_runtime import offload_evidence + + +class Parameter: + def __init__(self, device: str, size: int = 4, element_size: int = 4) -> None: + self.device = device + self._size = size + self._element_size = element_size + + def numel(self) -> int: + return self._size + + def element_size(self) -> int: + return self._element_size + + +class Module: + def __init__(self, hook: bool) -> None: + if hook: + self._hf_hook = object() + + +class Model: + def __init__(self, *, hook: bool, devices: tuple[str, ...]) -> None: + self._modules = (Module(hook),) + self._parameters = tuple(Parameter(device) for device in devices) + + def modules(self): + return iter(self._modules) + + def parameters(self): + return iter(self._parameters) + + +class TorchAdapterEvidenceTests(unittest.TestCase): + def test_requested_offload_without_residency_evidence_is_not_verified(self) -> None: + evidence = offload_evidence(Model(hook=True, devices=("cuda:0",)), inference_completed=True) + self.assertFalse(evidence["sequential_offload_verified"]) + + def test_completed_hooked_cpu_backed_run_can_be_verified(self) -> None: + evidence = offload_evidence(Model(hook=True, devices=("meta",)), inference_completed=True) + self.assertTrue(evidence["sequential_offload_verified"]) + self.assertEqual(evidence["persistent_cuda_parameter_bytes"], 0) + + +if __name__ == "__main__": + unittest.main()