sim_incremental_core/projection/
engine.rs

1use std::{
2    collections::{BTreeMap, BTreeSet},
3    sync::Arc,
4};
5
6use sim_kernel::{ContentId, Datum, Symbol};
7
8use super::{
9    ConfinementEvidence, FederatedClosure, MediatedAccessWitness, ProjectionDigest,
10    ProjectionError, ProjectionKindRef, ProjectionProvider, ProjectionResult, ProjectionSpec,
11    ProjectorPolicy, ProjectorQualification, admission::content_id_datum, admission::policy_id,
12};
13
14/// Port used to validate a provider configuration through its declared Shape.
15pub trait ConfigShapeVerifier {
16    /// Rejects a config that does not match `shape`.
17    fn verify(&self, shape: &ContentId, config: &Datum) -> Result<(), String>;
18}
19
20/// Open loaded registry keyed by projection kind.
21#[derive(Default)]
22pub struct ProjectionRegistry {
23    providers: BTreeMap<ProjectionKindRef, RegisteredProvider>,
24}
25
26struct RegisteredProvider {
27    identity: super::PackageIdentity,
28    provider: Arc<dyn ProjectionProvider>,
29}
30
31impl ProjectionRegistry {
32    /// Constructs an empty registry.
33    #[must_use]
34    pub fn new() -> Self {
35        Self::default()
36    }
37
38    /// Loads one provider without changing a central kind enum.
39    pub fn register(
40        &mut self,
41        identity: super::PackageIdentity,
42        provider: Arc<dyn ProjectionProvider>,
43    ) -> Result<(), ProjectionError> {
44        let kind = provider.kind().clone();
45        if self
46            .providers
47            .insert(kind.clone(), RegisteredProvider { identity, provider })
48            .is_some()
49        {
50            return Err(ProjectionError::DuplicateProvider(kind));
51        }
52        Ok(())
53    }
54
55    /// Resolves a loaded provider.
56    #[must_use]
57    pub fn get(
58        &self,
59        kind: &ProjectionKindRef,
60    ) -> Option<(&super::PackageIdentity, &Arc<dyn ProjectionProvider>)> {
61        self.providers
62            .get(kind)
63            .map(|registered| (&registered.identity, &registered.provider))
64    }
65
66    /// Lists loaded kinds in canonical order.
67    pub fn kinds(&self) -> impl ExactSizeIterator<Item = &ProjectionKindRef> {
68        self.providers.keys()
69    }
70}
71
72/// Pure projection coordinator over a sealed world and federated closure.
73pub struct ProjectionEngine<'a> {
74    registry: &'a ProjectionRegistry,
75    shapes: &'a dyn ConfigShapeVerifier,
76    closure: &'a FederatedClosure,
77}
78
79impl<'a> ProjectionEngine<'a> {
80    /// Binds the loaded registry, Shape verifier, and sealed closure.
81    #[must_use]
82    pub fn new(
83        registry: &'a ProjectionRegistry,
84        shapes: &'a dyn ConfigShapeVerifier,
85        closure: &'a FederatedClosure,
86    ) -> Self {
87        Self {
88            registry,
89            shapes,
90            closure,
91        }
92    }
93
94    /// Runs one qualified projection without acquiring observation or effects.
95    pub fn project(
96        &self,
97        world: &super::ObservedWorld,
98        spec: &ProjectionSpec,
99        policy: &ProjectorPolicy,
100        qualification: Option<&ProjectorQualification>,
101        confinement: Option<ConfinementEvidence>,
102    ) -> Result<ProjectionResult, ProjectionError> {
103        let qualification = qualification.ok_or_else(|| {
104            ProjectionError::UnqualifiedProjector("projector qualification is missing".to_owned())
105        })?;
106        let (loaded_identity, provider) = self
107            .registry
108            .get(&spec.kind)
109            .ok_or_else(|| ProjectionError::UnknownProvider(spec.kind.clone()))?;
110        if loaded_identity != &spec.provider {
111            return Err(ProjectionError::CodeIdentityMismatch);
112        }
113        if provider.config_shape() != &spec.config_shape {
114            return Err(ProjectionError::ConfigShapeMismatch);
115        }
116        self.shapes
117            .verify(&spec.config_shape, &spec.config)
118            .map_err(ProjectionError::InvalidConfig)?;
119        let expected_policy = policy_id(policy)
120            .map_err(|error| ProjectionError::UnqualifiedProjector(error.to_string()))?;
121        if qualification.policy() != &expected_policy {
122            return Err(ProjectionError::UnqualifiedProjector(
123                "qualification policy identity differs".to_owned(),
124            ));
125        }
126        if qualification.implementation() != &spec.provider.code {
127            return Err(ProjectionError::CodeIdentityMismatch);
128        }
129        if policy.requires_confinement {
130            let evidence = confinement.as_ref().ok_or_else(|| {
131                ProjectionError::UnavailableConfinement("required membrane absent".to_owned())
132            })?;
133            if !evidence.live {
134                return Err(ProjectionError::UnavailableConfinement(
135                    "required membrane unavailable on selected host".to_owned(),
136                ));
137            }
138        }
139        if policy.reads.facts().len() > policy.budgets.max_inputs {
140            return Err(ProjectionError::BudgetExceeded("selected inputs"));
141        }
142        let inputs = world.select(&policy.reads)?;
143        let output = provider.project(&inputs, &spec.config)?;
144        let accessed = inputs.accessed();
145        for fact in &accessed {
146            if !output.dependencies.contains(fact) {
147                return Err(ProjectionError::UndeclaredAccess {
148                    accessed: fact.clone(),
149                });
150            }
151        }
152        for fact in &output.dependencies {
153            if !accessed.contains(fact) {
154                return Err(ProjectionError::UnreadDependency(fact.clone()));
155            }
156        }
157        let output_bytes = output
158            .value
159            .canonical_bytes()
160            .map_err(|error| ProjectionError::Canonical(error.to_string()))?;
161        if output_bytes.len() > policy.budgets.max_output_bytes {
162            return Err(ProjectionError::BudgetExceeded("projection output"));
163        }
164        let digest = projection_digest(
165            spec,
166            qualification,
167            &inputs,
168            &output.dependencies,
169            &output.value,
170        )?;
171        let affected = self.closure.affected(output.dependencies.iter().cloned());
172        let mut explanations = Vec::new();
173        for conclusion in &affected {
174            for fact in &output.dependencies {
175                if let Ok(explanation) = self.closure.explain(conclusion, fact) {
176                    explanations.push(explanation);
177                }
178            }
179        }
180        Ok(ProjectionResult {
181            projection: output.value,
182            mediated_access: MediatedAccessWitness {
183                selected: policy.reads.facts().cloned().collect(),
184                accessed,
185            },
186            projector_qualification: qualification.clone(),
187            confinement,
188            digest,
189            affected,
190            explanations,
191        })
192    }
193}
194
195fn projection_digest(
196    spec: &ProjectionSpec,
197    qualification: &ProjectorQualification,
198    inputs: &super::ProjectionInputs,
199    dependencies: &BTreeSet<super::FactId>,
200    output: &Datum,
201) -> Result<ProjectionDigest, ProjectionError> {
202    let facts = dependencies
203        .iter()
204        .map(|id| {
205            let value = inputs
206                .get(id)
207                .expect("provider dependency was validated as an accessed selected fact");
208            Datum::Node {
209                tag: Symbol::qualified("projection", "semantic-input-v1"),
210                fields: vec![
211                    (Symbol::new("id"), Datum::String(id.as_str().to_owned())),
212                    (Symbol::new("value"), value.clone()),
213                ],
214            }
215        })
216        .collect();
217    let preimage = Datum::Node {
218        tag: Symbol::qualified("projection", "semantic-projection-digest-v1"),
219        fields: vec![
220            (
221                Symbol::new("kind"),
222                Datum::String(spec.kind.as_str().to_owned()),
223            ),
224            (
225                Symbol::new("package"),
226                Datum::String(spec.provider.name.clone()),
227            ),
228            (
229                Symbol::new("version"),
230                Datum::String(spec.provider.version.clone()),
231            ),
232            (
233                Symbol::new("implementation"),
234                content_id_datum(qualification.implementation()),
235            ),
236            (
237                Symbol::new("policy"),
238                content_id_datum(qualification.policy()),
239            ),
240            (Symbol::new("config"), spec.config.clone()),
241            (Symbol::new("inputs"), Datum::Vector(facts)),
242            (Symbol::new("projection"), output.clone()),
243        ],
244    };
245    preimage
246        .content_id()
247        .map(ProjectionDigest)
248        .map_err(|error| ProjectionError::Canonical(error.to_string()))
249}
sim_incremental_core/projection/engine.rs

sim_incremental_core/projection/
engine.rs
