soma_application/graduation/
comparison.rs

1use std::{collections::HashSet, path::Path};
2
3use serde::{Deserialize, Serialize};
4use serde_json::{Value, json};
5use soma_provider_core::ProviderInvocationContext;
6
7use super::{
8    ConformanceAttestation, GraduationArtifact, GraduationState, MAX_FIXTURE_BYTES, WorkspaceLock,
9    digest_bytes, digest_file, ensure_no_transaction, read_bounded, read_state,
10    validate_state_paths, write_state,
11};
12
13const MAX_FIXTURE_VALUE_BYTES: usize = 128 * 1024;
14const MAX_FIXTURES: usize = 64;
15const MAX_COMPARISON_REPORT_BYTES: usize = 32 * 1024;
16
17/// One recorded Python input/output pair used for component conformance.
18#[derive(Debug, Clone, Serialize, Deserialize)]
19#[serde(deny_unknown_fields)]
20pub struct GraduationFixture {
21    /// Stable fixture name used in comparison reports.
22    pub name: String,
23    /// Fixture selector containing only provider, action, and arguments.
24    pub input: Value,
25    /// Expected JSON result produced by the Python provider.
26    pub expected: Value,
27}
28
29struct PreparedComparison {
30    state: GraduationState,
31    candidate: GraduationArtifact,
32    catalog_digest: String,
33    component: soma_provider_adapters::wasm::PreparedComponentArtifact,
34}
35
36struct ComparisonOutcome {
37    candidate: GraduationArtifact,
38    fixture_digest: String,
39    fixture_count: usize,
40    matches: bool,
41    source_sha256: String,
42    catalog_sha256: String,
43}
44
45pub(crate) struct FixtureSnapshot {
46    pub fixtures: Vec<GraduationFixture>,
47    pub digest: String,
48}
49
50/// Host-owned inputs for one bounded Python/component dual-run.
51pub struct ComparisonRequest<'a> {
52    /// Optional candidate path, which must resolve to the published candidate.
53    pub component: Option<&'a Path>,
54    /// Immutable fixture snapshot used by both runtime executions.
55    pub(crate) fixtures: FixtureSnapshot,
56    /// Canonical host-owned input and live Python output for every fixture.
57    pub live_runs: Vec<(Value, Value)>,
58    /// Authenticated invocation context shared by both runtimes.
59    pub context: &'a ProviderInvocationContext,
60    /// Canonical managed provider root.
61    pub provider_root: &'a Path,
62    /// Absolute deadline covering preparation, execution, and persistence.
63    pub deadline: tokio::time::Instant,
64    /// Negotiated byte limit for the exact surface response envelope.
65    pub max_response_bytes: usize,
66}
67
68/// Replay fixtures and persist digest-bound conformance evidence.
69pub async fn compare(workspace: &Path, request: ComparisonRequest<'_>) -> anyhow::Result<Value> {
70    let ComparisonRequest {
71        component,
72        fixtures,
73        live_runs,
74        context,
75        provider_root,
76        deadline,
77        max_response_bytes,
78    } = request;
79    let workspace_for_prepare = workspace.to_path_buf();
80    let component_for_prepare = component.map(Path::to_path_buf);
81    let provider_root_for_prepare = provider_root.to_path_buf();
82    let compile_deadline = deadline.into_std();
83    let prepare_task = tokio::task::spawn_blocking(move || {
84        prepare_comparison(
85            &workspace_for_prepare,
86            component_for_prepare.as_deref(),
87            &provider_root_for_prepare,
88            compile_deadline,
89        )
90    });
91    let remaining = deadline
92        .checked_duration_since(tokio::time::Instant::now())
93        .ok_or_else(|| anyhow::anyhow!("graduation comparison exceeded its 30 second limit"))?;
94    let prepared = tokio::time::timeout(remaining, prepare_task)
95        .await
96        .map_err(|_| anyhow::anyhow!("graduation comparison exceeded its 30 second limit"))???;
97    if live_runs.len() != fixtures.fixtures.len() {
98        anyhow::bail!("live Python result count does not match the fixture corpus");
99    }
100    let mut results = Vec::with_capacity(fixtures.fixtures.len());
101    for (fixture, (effective_input, live_output)) in fixtures.fixtures.iter().zip(live_runs) {
102        let remaining = deadline
103            .checked_duration_since(tokio::time::Instant::now())
104            .ok_or_else(|| anyhow::anyhow!("graduation comparison exceeded its 30 second limit"))?;
105        let actual = tokio::time::timeout(
106            remaining,
107            soma_provider_adapters::wasm::invoke_prepared_component_artifact_before_async(
108                &prepared.component,
109                &effective_input,
110                &prepared.state.catalog.capabilities,
111                context,
112                deadline.into_std(),
113            ),
114        )
115        .await
116        .map_err(|_| anyhow::anyhow!("graduation comparison exceeded its 30 second limit"))?;
117        let component_matches_live = actual.as_ref().is_ok_and(|actual| actual == &live_output);
118        let recorded_matches_live = fixture.expected == live_output;
119        results.push(json!({
120            "name": fixture.name.chars().take(64).collect::<String>(),
121            "recorded_matches_live": recorded_matches_live,
122            "component_matches_live": component_matches_live,
123            "error": actual.as_ref().err().map(|error| error.chars().take(96).collect::<String>()),
124        }));
125    }
126    let matches = results.iter().all(|result| {
127        result["recorded_matches_live"] == true && result["component_matches_live"] == true
128    });
129    let report = json!({
130        "ok": matches,
131        "artifact_sha256": prepared.candidate.sha256,
132        "source_sha256": prepared.state.source_sha256,
133        "fixtures_sha256": fixtures.digest,
134        "fixtures": results
135    });
136    if serde_json::to_vec(&report)?.len() > MAX_COMPARISON_REPORT_BYTES {
137        anyhow::bail!("graduation comparison report exceeds {MAX_COMPARISON_REPORT_BYTES} bytes");
138    }
139    crate::ExecuteActionResponse {
140        output: report.clone(),
141        request_id: context.request_id.clone(),
142        progress: context.progress.events(),
143    }
144    .enforce_serialized_limit(max_response_bytes)
145    .map_err(|error| anyhow::anyhow!(error.to_string()))?;
146    let workspace_for_finish = workspace.to_path_buf();
147    let outcome = ComparisonOutcome {
148        candidate: prepared.candidate.clone(),
149        fixture_digest: fixtures.digest.clone(),
150        fixture_count: fixtures.fixtures.len(),
151        matches,
152        source_sha256: prepared.state.source_sha256.clone(),
153        catalog_sha256: prepared.catalog_digest.clone(),
154    };
155    let finish_deadline = deadline.into_std();
156    let finish_task = tokio::task::spawn_blocking(move || {
157        finish_comparison(&workspace_for_finish, &outcome, finish_deadline)
158    });
159    let remaining = deadline
160        .checked_duration_since(tokio::time::Instant::now())
161        .ok_or_else(|| anyhow::anyhow!("graduation comparison exceeded its 30 second limit"))?;
162    tokio::time::timeout(remaining, finish_task)
163        .await
164        .map_err(|_| anyhow::anyhow!("graduation comparison exceeded its 30 second limit"))???;
165    Ok(report)
166}
167
168fn prepare_comparison(
169    workspace: &Path,
170    component: Option<&Path>,
171    provider_root: &Path,
172    deadline: std::time::Instant,
173) -> anyhow::Result<PreparedComparison> {
174    let _lock = WorkspaceLock::acquire_before(workspace, deadline)?;
175    ensure_no_transaction(workspace)?;
176    let mut state = read_state(workspace)?;
177    validate_state_paths(workspace, provider_root, &state)?;
178    if digest_file(&state.source)? != state.source_sha256 {
179        anyhow::bail!("live Python source changed since graduation was scaffolded");
180    }
181    let candidate = state
182        .candidate
183        .clone()
184        .ok_or_else(|| anyhow::anyhow!("no component candidate exists"))?;
185    let component = component.unwrap_or(&candidate.path).canonicalize()?;
186    if component != candidate.path.canonicalize()? {
187        anyhow::bail!("comparison component is not the published candidate");
188    }
189    if digest_file(&candidate.path)? != candidate.sha256 {
190        anyhow::bail!("component artifact digest mismatch");
191    }
192    // Starting a new comparison invalidates any prior proof immediately.
193    // Timeouts, response-limit failures, or mismatches must never leave an
194    // older candidate attestation available to activation.
195    if state.attestation.take().is_some() {
196        write_state(workspace, &state)?;
197    }
198    let component =
199        soma_provider_adapters::wasm::prepare_component_artifact_before(&candidate.path, deadline)
200            .map_err(anyhow::Error::msg)?;
201    let catalog_digest = super::catalog_contract_digest(&state.catalog)?;
202    Ok(PreparedComparison {
203        state,
204        candidate,
205        catalog_digest,
206        component,
207    })
208}
209
210fn finish_comparison(
211    workspace: &Path,
212    outcome: &ComparisonOutcome,
213    deadline: std::time::Instant,
214) -> anyhow::Result<()> {
215    let _lock = WorkspaceLock::acquire_before(workspace, deadline)?;
216    ensure_no_transaction(workspace)?;
217    let mut state = read_state(workspace)?;
218    if state.candidate.as_ref() != Some(&outcome.candidate) {
219        anyhow::bail!("graduation candidate changed while comparison was running");
220    }
221    if state.source_sha256 != outcome.source_sha256
222        || digest_file(&state.source)? != outcome.source_sha256
223    {
224        anyhow::bail!("live Python source changed while comparison was running");
225    }
226    if state.catalog_sha256 != outcome.catalog_sha256
227        || super::catalog_contract_digest(&state.catalog)? != outcome.catalog_sha256
228    {
229        anyhow::bail!("provider catalog changed while comparison was running");
230    }
231    if digest_file(&outcome.candidate.path)? != outcome.candidate.sha256 {
232        anyhow::bail!("component candidate changed while comparison was running");
233    }
234    state.attestation = outcome.matches.then(|| ConformanceAttestation {
235        artifact_sha256: outcome.candidate.sha256.clone(),
236        fixtures_sha256: outcome.fixture_digest.clone(),
237        fixture_count: outcome.fixture_count,
238        source_sha256: outcome.source_sha256.clone(),
239        catalog_sha256: outcome.catalog_sha256.clone(),
240        verified_unix_ms: super::unix_ms(),
241    });
242    write_state(workspace, &state)
243}
244
245pub(crate) fn read_fixture_snapshot(path: &Path) -> anyhow::Result<FixtureSnapshot> {
246    let bytes = read_bounded(path, MAX_FIXTURE_BYTES, "graduation fixtures")?;
247    Ok(FixtureSnapshot {
248        fixtures: parse_fixtures(&bytes)?,
249        digest: digest_bytes(&bytes),
250    })
251}
252
253pub(crate) fn read_fixtures(path: &Path) -> anyhow::Result<Vec<GraduationFixture>> {
254    Ok(read_fixture_snapshot(path)?.fixtures)
255}
256
257fn parse_fixtures(bytes: &[u8]) -> anyhow::Result<Vec<GraduationFixture>> {
258    let corpus: Vec<GraduationFixture> = serde_json::from_slice(bytes)?;
259    if corpus.is_empty() {
260        anyhow::bail!("graduation fixture corpus must not be empty");
261    }
262    if corpus.len() > MAX_FIXTURES {
263        anyhow::bail!("graduation fixture corpus exceeds {MAX_FIXTURES} entries");
264    }
265    let mut names = HashSet::with_capacity(corpus.len());
266    for fixture in &corpus {
267        if fixture.name.is_empty() || fixture.name.len() > 256 {
268            anyhow::bail!("graduation fixture names must contain 1 to 256 bytes");
269        }
270        if !names.insert(fixture.name.as_str()) {
271            anyhow::bail!("graduation fixture names must be unique");
272        }
273        let input = fixture
274            .input
275            .as_object()
276            .ok_or_else(|| anyhow::anyhow!("graduation fixture input must be an object"))?;
277        if !["provider", "action", "arguments"]
278            .iter()
279            .all(|key| input.contains_key(*key))
280            || input
281                .keys()
282                .any(|key| !matches!(key.as_str(), "provider" | "action" | "arguments"))
283        {
284            anyhow::bail!(
285                "graduation fixture input must contain only provider, action, and arguments"
286            );
287        }
288        if serde_json::to_vec(&(&fixture.input, &fixture.expected))?.len() > MAX_FIXTURE_VALUE_BYTES
289        {
290            anyhow::bail!(
291                "graduation fixture `{}` exceeds {MAX_FIXTURE_VALUE_BYTES} input/output bytes",
292                fixture.name
293            );
294        }
295    }
296    Ok(corpus)
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302
303    #[test]
304    fn requires_a_nonempty_recorded_fixture_set() {
305        assert!(
306            parse_fixtures(b"[]")
307                .expect_err("empty corpus")
308                .to_string()
309                .contains("must not be empty")
310        );
311    }
312
313    #[test]
314    fn corpus_and_entries_are_bounded() {
315        assert!(parse_fixtures(&vec![b' '; MAX_FIXTURE_BYTES + 1]).is_err());
316
317        let duplicate = serde_json::to_vec(&[
318            GraduationFixture {
319                name: "same".to_owned(),
320                input: json!({"provider": "example", "action": "echo", "arguments": {}}),
321                expected: json!({}),
322            },
323            GraduationFixture {
324                name: "same".to_owned(),
325                input: json!({"provider": "example", "action": "echo", "arguments": {}}),
326                expected: json!({}),
327            },
328        ])
329        .expect("fixture JSON");
330        assert!(
331            parse_fixtures(&duplicate)
332                .expect_err("duplicates rejected")
333                .to_string()
334                .contains("unique")
335        );
336
337        let oversized = serde_json::to_vec(&[GraduationFixture {
338            name: "large".to_owned(),
339            input: json!({
340                "provider": "example",
341                "action": "echo",
342                "arguments": {"value": "x".repeat(MAX_FIXTURE_VALUE_BYTES)}
343            }),
344            expected: json!({}),
345        }])
346        .expect("fixture JSON");
347        assert!(
348            parse_fixtures(&oversized)
349                .expect_err("oversized value rejected")
350                .to_string()
351                .contains("input/output bytes")
352        );
353    }
354
355    #[test]
356    fn fixture_snapshot_is_immutable_after_the_source_file_changes() {
357        let temp = tempfile::NamedTempFile::new().expect("fixture file");
358        let first = serde_json::to_vec(&[GraduationFixture {
359            name: "first".to_owned(),
360            input: json!({"provider": "example", "action": "echo", "arguments": {}}),
361            expected: json!({"value": 1}),
362        }])
363        .expect("fixture JSON");
364        std::fs::write(temp.path(), &first).expect("first corpus");
365        let snapshot = read_fixture_snapshot(temp.path()).expect("snapshot");
366        std::fs::write(
367            temp.path(),
368            serde_json::to_vec(&[GraduationFixture {
369                name: "second".to_owned(),
370                input: json!({"provider": "example", "action": "echo", "arguments": {}}),
371                expected: json!({"value": 2}),
372            }])
373            .expect("second fixture JSON"),
374        )
375        .expect("replace corpus");
376
377        assert_eq!(snapshot.fixtures[0].name, "first");
378        assert_eq!(snapshot.digest, digest_bytes(&first));
379    }
380}
soma_application/graduation/comparison.rs

soma_application/graduation/
comparison.rs
