Skip to main content

supercov_engine/
rust_probe_transport.rs

1//! Bounded, lock-free, file-backed transport for owned Rust probe events.
2//!
3//! The supervisor creates and authenticates the fixed-layout file before a
4//! test starts. Target code maps that file and publishes variable-length
5//! records through fixed descriptors. A release-store commit byte makes every
6//! complete descriptor independently recoverable even if another writer or
7//! the whole process dies midway through a later record.
8
9use std::{
10    collections::{BTreeMap, BTreeSet},
11    fs::{self, File, OpenOptions},
12    path::Path,
13    sync::atomic::{AtomicU8, AtomicU64, Ordering},
14};
15
16use memmap2::{Mmap, MmapMut, MmapOptions};
17use serde::{Deserialize, Serialize};
18
19use crate::rust_runtime::{RustProbeObservation, valid_probe_id};
20
21pub const RUST_TRANSPORT_ENV: &str = "SUPERCOV_RUST_TRANSPORT_FILE";
22pub const RUST_TRANSPORT_TOKEN_ENV: &str = "SUPERCOV_RUST_TRANSPORT_TOKEN";
23pub const RUST_CONTEXT_ENV: &str = "SUPERCOV_RUST_CONTEXT_ID";
24pub const DEFAULT_DESCRIPTOR_CAPACITY: u32 = 32_768;
25pub const DEFAULT_PAYLOAD_CAPACITY: u32 = 4 * 1024 * 1024;
26
27const MAGIC: &[u8; 8] = b"SCVRUST3";
28const VERSION: u32 = 3;
29const HEADER_SIZE: usize = 128;
30const DESCRIPTOR_SIZE: usize = 40;
31const ENDIAN_MARKER: u32 = 0x0102_0304;
32const NEXT_DESCRIPTOR_OFFSET: usize = 32;
33const NEXT_PAYLOAD_OFFSET: usize = 40;
34const DROPPED_OFFSET: usize = 48;
35const TOKEN_OFFSET: usize = 56;
36const TOKEN_SIZE: usize = 16;
37const ATTACHMENTS_OFFSET: usize = 72;
38const NEXT_PHASE_OFFSET: usize = 80;
39
40const COMMIT_OFFSET: usize = 0;
41const KIND_OFFSET: usize = 1;
42const OUTCOME_OFFSET: usize = 2;
43const PID_OFFSET: usize = 4;
44const CONTEXT_OFFSET: usize = 8;
45const PAYLOAD_OFFSET_OFFSET: usize = 16;
46const PAYLOAD_LENGTH_OFFSET: usize = 20;
47const ID_LENGTH_OFFSET: usize = 24;
48const VALUE_LENGTH_OFFSET: usize = 28;
49const CHECKSUM_OFFSET: usize = 32;
50
51const KIND_HIT: u8 = 1;
52const KIND_DECISION: u8 = 2;
53const KIND_ORDINAL_HIT: u8 = 3;
54const KIND_PHASE: u8 = 4;
55const KIND_THREAD_PHASE: u8 = 5;
56const KIND_THREAD_END: u8 = 6;
57const KIND_TEST_BOUNDARY: u8 = 7;
58
59const RUNTIME_TEMPLATE: &str = include_str!("../runtime-assets/rust-mmap-runtime.rs");
60
61#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
62#[serde(rename_all = "camelCase", deny_unknown_fields)]
63pub struct RustTransportRead {
64    pub observations: Vec<RustTransportObservation>,
65    pub ordinal_hits: Vec<RustOrdinalHit>,
66    pub phases: Vec<RustPhaseContext>,
67    pub thread_phases: Vec<RustThreadPhase>,
68    pub thread_ends: Vec<RustThreadEnd>,
69    pub test_boundaries: Vec<RustTestBoundary>,
70    pub committed: u64,
71    pub incomplete: u64,
72    pub dropped: u64,
73    pub attachments: u64,
74}
75
76impl RustTransportRead {
77    pub fn empty() -> Self {
78        Self {
79            observations: Vec::new(),
80            ordinal_hits: Vec::new(),
81            phases: Vec::new(),
82            thread_phases: Vec::new(),
83            thread_ends: Vec::new(),
84            test_boundaries: Vec::new(),
85            committed: 0,
86            incomplete: 0,
87            dropped: 0,
88            attachments: 0,
89        }
90    }
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
94#[serde(rename_all = "camelCase", deny_unknown_fields)]
95pub struct RustTransportObservation {
96    pub process_id: u32,
97    pub context_id: u64,
98    pub observation: RustProbeObservation,
99}
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
102#[serde(rename_all = "camelCase", deny_unknown_fields)]
103pub struct RustOrdinalHit {
104    pub process_id: u32,
105    pub context_id: u64,
106    pub ordinal: u64,
107}
108
109#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
110#[serde(rename_all = "camelCase", deny_unknown_fields)]
111pub struct RustPhaseContext {
112    pub process_id: u32,
113    pub child_context_id: u64,
114    pub parent_context_id: u64,
115    pub invocation_nonce: u64,
116    pub decision_id: String,
117}
118
119/// An inherited native thread's derived phase context definition. The
120/// `commit_index` is the record's global transport descriptor index, which is
121/// the total order used by the join-bounded acceptance rule.
122#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
123#[serde(rename_all = "camelCase", deny_unknown_fields)]
124pub struct RustThreadPhase {
125    pub process_id: u32,
126    pub child_context_id: u64,
127    pub parent_context_id: u64,
128    pub invocation_nonce: u64,
129    pub commit_index: u64,
130}
131
132/// The end-of-thread record bounding one thread phase in commit order.
133#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
134#[serde(rename_all = "camelCase", deny_unknown_fields)]
135pub struct RustThreadEnd {
136    pub process_id: u32,
137    pub context_id: u64,
138    pub commit_index: u64,
139}
140
141/// The exact-test boundary record committed when a test context is exited.
142#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
143#[serde(rename_all = "camelCase", deny_unknown_fields)]
144pub struct RustTestBoundary {
145    pub process_id: u32,
146    pub context_id: u64,
147    pub commit_index: u64,
148}
149
150#[derive(Debug, Clone, PartialEq, Eq)]
151pub struct RustTransportPartition {
152    pub attributed: BTreeMap<u64, RustTransportRead>,
153    pub background: RustTransportRead,
154    /// One entry per thread phase whose lifetime escaped its root test; every
155    /// record under such a chain is deterministic background evidence.
156    pub thread_scope_limitations: BTreeSet<String>,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
160pub enum RustTransportError {
161    Io(String),
162    UnsafeFile(String),
163    InvalidHeader,
164    InvalidLength,
165    InvalidDescriptor(u64),
166    InvalidRecord(u64),
167    InvalidAssertionContext(String),
168    InvalidAttribution(String),
169}
170
171impl std::fmt::Display for RustTransportError {
172    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
173        match self {
174            Self::Io(error) => write!(formatter, "Rust transport I/O failed: {error}"),
175            Self::UnsafeFile(path) => write!(formatter, "unsafe Rust transport file: {path}"),
176            Self::InvalidHeader => write!(formatter, "invalid Rust transport header"),
177            Self::InvalidLength => write!(formatter, "invalid Rust transport length"),
178            Self::InvalidDescriptor(index) => {
179                write!(formatter, "invalid Rust transport descriptor {index}")
180            }
181            Self::InvalidRecord(index) => {
182                write!(formatter, "invalid Rust transport record {index}")
183            }
184            Self::InvalidAssertionContext(reason) => {
185                write!(formatter, "invalid Rust assertion context: {reason}")
186            }
187            Self::InvalidAttribution(reason) => {
188                write!(formatter, "invalid Rust transport attribution: {reason}")
189            }
190        }
191    }
192}
193
194impl std::error::Error for RustTransportError {}
195
196pub fn rust_assertion_context_id(
197    parent: u64,
198    decision_id: &str,
199    invocation_nonce: u64,
200) -> Result<u64, RustTransportError> {
201    if parent == 0 {
202        return Ok(0);
203    }
204    if parent == u64::MAX {
205        return Err(RustTransportError::InvalidAssertionContext(
206            "the reserved nesting sentinel cannot be a parent".into(),
207        ));
208    }
209    let digest = decision_id
210        .strip_prefix("rs:decision:")
211        .filter(|digest| digest.len() == 24 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()))
212        .ok_or_else(|| {
213            RustTransportError::InvalidAssertionContext(format!(
214                "invalid decision ID {decision_id}"
215            ))
216        })?;
217    let id_high = u64::from_str_radix(&digest[..16], 16).map_err(|error| {
218        RustTransportError::InvalidAssertionContext(format!(
219            "invalid decision ID {decision_id}: {error}"
220        ))
221    })?;
222    let id_low = u32::from_str_radix(&digest[16..], 16).map_err(|error| {
223        RustTransportError::InvalidAssertionContext(format!(
224            "invalid decision ID {decision_id}: {error}"
225        ))
226    })?;
227    let mut value = 0xcbf2_9ce4_8422_2325_u64;
228    for byte in b"supercov-rust-assertion-phase-v2"
229        .iter()
230        .copied()
231        .chain(parent.to_le_bytes())
232        .chain(id_high.to_le_bytes())
233        .chain(id_low.to_le_bytes())
234        .chain(invocation_nonce.to_le_bytes())
235    {
236        value ^= u64::from(byte);
237        value = value.wrapping_mul(0x0000_0100_0000_01b3);
238    }
239    Ok(if matches!(value, 0 | u64::MAX) {
240        value ^ 0xa5a5_5a5a_d3c3_b4b4
241    } else {
242        value
243    })
244}
245
246/// Derive the exact thread-phase context for an inherited native thread. This
247/// mirrors the in-runtime derivation byte for byte so tampered thread-phase
248/// records fail authentication offline.
249pub fn rust_thread_context_id(parent: u64, invocation_nonce: u64) -> u64 {
250    let mut value = 0xcbf2_9ce4_8422_2325_u64;
251    for byte in b"supercov-rust-thread-phase-v1\0"
252        .iter()
253        .copied()
254        .chain(parent.to_le_bytes())
255        .chain(invocation_nonce.to_le_bytes())
256    {
257        value ^= u64::from(byte);
258        value = value.wrapping_mul(0x0000_0100_0000_01b3);
259    }
260    if matches!(value, 0 | u64::MAX) {
261        value ^ 0xa5a5_5a5a_d3c3_b4b4
262    } else {
263        value
264    }
265}
266
267fn put_u32(target: &mut [u8], offset: usize, value: u32) {
268    target[offset..offset + 4].copy_from_slice(&value.to_le_bytes());
269}
270
271fn get_u32(source: &[u8], offset: usize) -> Option<u32> {
272    Some(u32::from_le_bytes(
273        source.get(offset..offset + 4)?.try_into().ok()?,
274    ))
275}
276
277fn get_u64(source: &[u8], offset: usize) -> Option<u64> {
278    Some(u64::from_le_bytes(
279        source.get(offset..offset + 8)?.try_into().ok()?,
280    ))
281}
282
283fn total_size(descriptors: u32, payload: u32) -> Option<usize> {
284    HEADER_SIZE
285        .checked_add(
286            usize::try_from(descriptors)
287                .ok()?
288                .checked_mul(DESCRIPTOR_SIZE)?,
289        )?
290        .checked_add(usize::try_from(payload).ok()?)
291}
292
293fn map_mut(file: &File) -> Result<MmapMut, RustTransportError> {
294    // SAFETY: this function owns the newly created file mapping and no slice
295    // alias is produced outside the returned MmapMut.
296    unsafe { MmapOptions::new().map_mut(file) }
297        .map_err(|error| RustTransportError::Io(error.to_string()))
298}
299
300fn map(file: &File) -> Result<Mmap, RustTransportError> {
301    // SAFETY: the immutable mapping is retained for the duration of all reads.
302    unsafe { MmapOptions::new().map(file) }
303        .map_err(|error| RustTransportError::Io(error.to_string()))
304}
305
306pub fn create_rust_transport(
307    path: &Path,
308    token: [u8; TOKEN_SIZE],
309    descriptor_capacity: u32,
310    payload_capacity: u32,
311) -> Result<(), RustTransportError> {
312    if descriptor_capacity == 0 || payload_capacity == 0 {
313        return Err(RustTransportError::InvalidLength);
314    }
315    let total = total_size(descriptor_capacity, payload_capacity)
316        .ok_or(RustTransportError::InvalidLength)?;
317    let mut options = OpenOptions::new();
318    options.read(true).write(true).create_new(true);
319    #[cfg(unix)]
320    {
321        use std::os::unix::fs::OpenOptionsExt as _;
322        options.mode(0o600);
323    }
324    let file = options
325        .open(path)
326        .map_err(|error| RustTransportError::Io(error.to_string()))?;
327    file.set_len(u64::try_from(total).map_err(|_| RustTransportError::InvalidLength)?)
328        .map_err(|error| RustTransportError::Io(error.to_string()))?;
329    let mut mapping = map_mut(&file)?;
330    mapping[..MAGIC.len()].copy_from_slice(MAGIC);
331    put_u32(&mut mapping, 8, VERSION);
332    put_u32(&mut mapping, 12, HEADER_SIZE as u32);
333    put_u32(&mut mapping, 16, DESCRIPTOR_SIZE as u32);
334    put_u32(&mut mapping, 20, descriptor_capacity);
335    put_u32(&mut mapping, 24, payload_capacity);
336    put_u32(&mut mapping, 28, ENDIAN_MARKER);
337    mapping[TOKEN_OFFSET..TOKEN_OFFSET + TOKEN_SIZE].copy_from_slice(&token);
338    mapping
339        .flush_range(0, HEADER_SIZE)
340        .map_err(|error| RustTransportError::Io(error.to_string()))
341}
342
343fn regular_file(path: &Path) -> Result<File, RustTransportError> {
344    let metadata =
345        fs::symlink_metadata(path).map_err(|error| RustTransportError::Io(error.to_string()))?;
346    if !metadata.file_type().is_file() {
347        return Err(RustTransportError::UnsafeFile(path.display().to_string()));
348    }
349    let mut options = OpenOptions::new();
350    options.read(true);
351    #[cfg(any(target_os = "linux", target_os = "macos"))]
352    {
353        use std::os::unix::fs::OpenOptionsExt as _;
354        #[cfg(target_os = "linux")]
355        const O_NOFOLLOW: i32 = 0x2_0000;
356        #[cfg(target_os = "macos")]
357        const O_NOFOLLOW: i32 = 0x100;
358        options.custom_flags(O_NOFOLLOW);
359    }
360    let file = options
361        .open(path)
362        .map_err(|error| RustTransportError::Io(error.to_string()))?;
363    if !file
364        .metadata()
365        .map_err(|error| RustTransportError::Io(error.to_string()))?
366        .file_type()
367        .is_file()
368    {
369        return Err(RustTransportError::UnsafeFile(path.display().to_string()));
370    }
371    Ok(file)
372}
373
374fn atomic_u64(mapping: &[u8], offset: usize) -> Result<&AtomicU64, RustTransportError> {
375    let pointer = mapping
376        .get(offset..offset + 8)
377        .ok_or(RustTransportError::InvalidHeader)?
378        .as_ptr();
379    if !(pointer as usize).is_multiple_of(std::mem::align_of::<AtomicU64>()) {
380        return Err(RustTransportError::InvalidHeader);
381    }
382    // SAFETY: the fixed layout guarantees alignment and the mapped bytes live
383    // at least as long as the returned reference.
384    Ok(unsafe { &*pointer.cast::<AtomicU64>() })
385}
386
387fn atomic_u8(mapping: &[u8], offset: usize) -> Result<&AtomicU8, RustTransportError> {
388    let pointer = mapping
389        .get(offset)
390        .ok_or(RustTransportError::InvalidLength)? as *const u8;
391    // SAFETY: AtomicU8 has byte alignment and the mapping outlives the reference.
392    Ok(unsafe { &*pointer.cast::<AtomicU8>() })
393}
394
395#[allow(clippy::too_many_arguments)]
396fn checksum(
397    kind: u8,
398    outcome: u8,
399    pid: u32,
400    context: u64,
401    payload_offset: u32,
402    payload_length: u32,
403    id_length: u32,
404    value_length: u32,
405    id: &[u8],
406    values: &[u8],
407) -> u64 {
408    let mut value = 0xcbf2_9ce4_8422_2325_u64;
409    for byte in [kind, outcome]
410        .into_iter()
411        .chain(pid.to_le_bytes())
412        .chain(context.to_le_bytes())
413        .chain(payload_offset.to_le_bytes())
414        .chain(payload_length.to_le_bytes())
415        .chain(id_length.to_le_bytes())
416        .chain(value_length.to_le_bytes())
417        .chain(id.iter().copied())
418        .chain(values.iter().copied())
419    {
420        value ^= u64::from(byte);
421        value = value.wrapping_mul(0x0000_0100_0000_01b3);
422    }
423    value
424}
425
426pub fn read_rust_transport(
427    path: &Path,
428    expected_token: &[u8; TOKEN_SIZE],
429) -> Result<RustTransportRead, RustTransportError> {
430    let file = regular_file(path)?;
431    let mapping = map(&file)?;
432    if mapping.get(..8) != Some(MAGIC.as_slice())
433        || get_u32(&mapping, 8) != Some(VERSION)
434        || get_u32(&mapping, 12) != Some(HEADER_SIZE as u32)
435        || get_u32(&mapping, 16) != Some(DESCRIPTOR_SIZE as u32)
436        || get_u32(&mapping, 28) != Some(ENDIAN_MARKER)
437        || mapping.get(TOKEN_OFFSET..TOKEN_OFFSET + TOKEN_SIZE) != Some(expected_token.as_slice())
438        || mapping.get(52..56).is_none_or(|bytes| bytes != [0; 4])
439        || mapping
440            .get(NEXT_PHASE_OFFSET + 8..HEADER_SIZE)
441            .is_none_or(|bytes| bytes != [0; 40])
442    {
443        return Err(RustTransportError::InvalidHeader);
444    }
445    let descriptors = get_u32(&mapping, 20).ok_or(RustTransportError::InvalidHeader)?;
446    let payload_capacity = get_u32(&mapping, 24).ok_or(RustTransportError::InvalidHeader)?;
447    if descriptors == 0 || payload_capacity == 0 {
448        return Err(RustTransportError::InvalidHeader);
449    }
450    if mapping.len()
451        != total_size(descriptors, payload_capacity).ok_or(RustTransportError::InvalidLength)?
452    {
453        return Err(RustTransportError::InvalidLength);
454    }
455    let next = atomic_u64(&mapping, NEXT_DESCRIPTOR_OFFSET)?.load(Ordering::Acquire);
456    let next_payload = atomic_u64(&mapping, NEXT_PAYLOAD_OFFSET)?.load(Ordering::Acquire);
457    let recorded_dropped = atomic_u64(&mapping, DROPPED_OFFSET)?.load(Ordering::Acquire);
458    let attachments = atomic_u64(&mapping, ATTACHMENTS_OFFSET)?.load(Ordering::Acquire);
459    let inspect = next.min(u64::from(descriptors));
460    let payload_base = HEADER_SIZE + descriptors as usize * DESCRIPTOR_SIZE;
461    let mut observations = Vec::new();
462    let mut ordinal_hits = Vec::new();
463    let mut phases = Vec::new();
464    let mut thread_phases = Vec::new();
465    let mut thread_ends = Vec::new();
466    let mut test_boundaries = Vec::new();
467    let mut phase_definitions = BTreeMap::<u64, (u8, u64, u64, String)>::new();
468    let mut thread_end_contexts = BTreeSet::<u64>::new();
469    let mut boundary_contexts = BTreeSet::<u64>::new();
470    let mut committed = 0_u64;
471    for index in 0..inspect {
472        let descriptor = HEADER_SIZE + index as usize * DESCRIPTOR_SIZE;
473        match atomic_u8(&mapping, descriptor + COMMIT_OFFSET)?.load(Ordering::Acquire) {
474            0 => continue,
475            1 => {}
476            _ => return Err(RustTransportError::InvalidDescriptor(index)),
477        }
478        committed += 1;
479        let kind = mapping[descriptor + KIND_OFFSET];
480        let outcome = mapping[descriptor + OUTCOME_OFFSET];
481        if mapping[descriptor + 3] != 0 {
482            return Err(RustTransportError::InvalidDescriptor(index));
483        }
484        let pid = get_u32(&mapping, descriptor + PID_OFFSET)
485            .ok_or(RustTransportError::InvalidDescriptor(index))?;
486        let context_id = get_u64(&mapping, descriptor + CONTEXT_OFFSET)
487            .ok_or(RustTransportError::InvalidDescriptor(index))?;
488        let payload_offset = get_u32(&mapping, descriptor + PAYLOAD_OFFSET_OFFSET)
489            .ok_or(RustTransportError::InvalidDescriptor(index))?
490            as usize;
491        let payload_length = get_u32(&mapping, descriptor + PAYLOAD_LENGTH_OFFSET)
492            .ok_or(RustTransportError::InvalidDescriptor(index))?
493            as usize;
494        let id_length = get_u32(&mapping, descriptor + ID_LENGTH_OFFSET)
495            .ok_or(RustTransportError::InvalidDescriptor(index))? as usize;
496        let value_length = get_u32(&mapping, descriptor + VALUE_LENGTH_OFFSET)
497            .ok_or(RustTransportError::InvalidDescriptor(index))?
498            as usize;
499        let expected_checksum = get_u64(&mapping, descriptor + CHECKSUM_OFFSET)
500            .ok_or(RustTransportError::InvalidDescriptor(index))?;
501        let end = payload_offset
502            .checked_add(payload_length)
503            .filter(|end| *end <= payload_capacity as usize)
504            .ok_or(RustTransportError::InvalidDescriptor(index))?;
505        if payload_length != id_length.saturating_add(value_length)
506            || u64::from(end as u32) > next_payload
507        {
508            return Err(RustTransportError::InvalidDescriptor(index));
509        }
510        let payload = mapping
511            .get(payload_base + payload_offset..payload_base + end)
512            .ok_or(RustTransportError::InvalidDescriptor(index))?;
513        let (id, values) = payload.split_at(id_length);
514        if checksum(
515            kind,
516            outcome,
517            pid,
518            context_id,
519            payload_offset as u32,
520            payload_length as u32,
521            id_length as u32,
522            value_length as u32,
523            id,
524            values,
525        ) != expected_checksum
526        {
527            return Err(RustTransportError::InvalidRecord(index));
528        }
529        let id = std::str::from_utf8(id).map_err(|_| RustTransportError::InvalidRecord(index))?;
530        if matches!(kind, KIND_HIT | KIND_DECISION | KIND_PHASE) && !valid_probe_id(id) {
531            return Err(RustTransportError::InvalidRecord(index));
532        }
533        match kind {
534            KIND_HIT if outcome == 0 && values.is_empty() => {
535                observations.push(RustTransportObservation {
536                    process_id: pid,
537                    context_id,
538                    observation: RustProbeObservation::Hit { id: id.into() },
539                });
540            }
541            KIND_DECISION
542                if matches!(outcome, 0 | 1)
543                    && id.starts_with("rs:decision:")
544                    && !values.is_empty()
545                    && values.iter().all(|value| matches!(*value, 0..=2)) =>
546            {
547                observations.push(RustTransportObservation {
548                    process_id: pid,
549                    context_id,
550                    observation: RustProbeObservation::Decision {
551                        id: id.into(),
552                        values: values
553                            .iter()
554                            .map(|value| match value {
555                                0 => None,
556                                1 => Some(false),
557                                2 => Some(true),
558                                _ => unreachable!(),
559                            })
560                            .collect(),
561                        outcome: outcome == 1,
562                    },
563                });
564            }
565            KIND_ORDINAL_HIT if outcome == 0 && id.is_empty() && values.len() == 8 => {
566                ordinal_hits.push(RustOrdinalHit {
567                    process_id: pid,
568                    context_id,
569                    ordinal: u64::from_le_bytes(
570                        values
571                            .try_into()
572                            .map_err(|_| RustTransportError::InvalidRecord(index))?,
573                    ),
574                });
575            }
576            KIND_PHASE
577                if outcome == 0
578                    && id.starts_with("rs:decision:")
579                    && values.len() == 16
580                    && !matches!(context_id, 0 | u64::MAX) =>
581            {
582                let parent_context_id = u64::from_le_bytes(
583                    values[..8]
584                        .try_into()
585                        .map_err(|_| RustTransportError::InvalidRecord(index))?,
586                );
587                let invocation_nonce = u64::from_le_bytes(
588                    values[8..]
589                        .try_into()
590                        .map_err(|_| RustTransportError::InvalidRecord(index))?,
591                );
592                if matches!(parent_context_id, 0 | u64::MAX)
593                    || rust_assertion_context_id(parent_context_id, id, invocation_nonce)?
594                        != context_id
595                {
596                    return Err(RustTransportError::InvalidAssertionContext(format!(
597                        "phase record {index} does not derive child {context_id:016x} from parent {parent_context_id:016x} and {id}"
598                    )));
599                }
600                let definition = (
601                    KIND_PHASE,
602                    parent_context_id,
603                    invocation_nonce,
604                    id.to_owned(),
605                );
606                if phase_definitions
607                    .insert(context_id, definition.clone())
608                    .is_some_and(|existing| existing != definition)
609                {
610                    return Err(RustTransportError::InvalidAssertionContext(format!(
611                        "child {context_id:016x} has conflicting phase definitions"
612                    )));
613                }
614                phases.push(RustPhaseContext {
615                    process_id: pid,
616                    child_context_id: context_id,
617                    parent_context_id,
618                    invocation_nonce,
619                    decision_id: id.into(),
620                });
621            }
622            KIND_THREAD_PHASE
623                if outcome == 0
624                    && id.is_empty()
625                    && values.len() == 16
626                    && !matches!(context_id, 0 | u64::MAX) =>
627            {
628                let parent_context_id = u64::from_le_bytes(
629                    values[..8]
630                        .try_into()
631                        .map_err(|_| RustTransportError::InvalidRecord(index))?,
632                );
633                let invocation_nonce = u64::from_le_bytes(
634                    values[8..]
635                        .try_into()
636                        .map_err(|_| RustTransportError::InvalidRecord(index))?,
637                );
638                if matches!(parent_context_id, 0 | u64::MAX)
639                    || rust_thread_context_id(parent_context_id, invocation_nonce) != context_id
640                {
641                    return Err(RustTransportError::InvalidAssertionContext(format!(
642                        "thread phase record {index} does not derive child {context_id:016x} from parent {parent_context_id:016x} and nonce {invocation_nonce}"
643                    )));
644                }
645                let definition = (
646                    KIND_THREAD_PHASE,
647                    parent_context_id,
648                    invocation_nonce,
649                    String::new(),
650                );
651                if phase_definitions
652                    .insert(context_id, definition.clone())
653                    .is_some_and(|existing| existing != definition)
654                {
655                    return Err(RustTransportError::InvalidAssertionContext(format!(
656                        "child {context_id:016x} has conflicting phase definitions"
657                    )));
658                }
659                thread_phases.push(RustThreadPhase {
660                    process_id: pid,
661                    child_context_id: context_id,
662                    parent_context_id,
663                    invocation_nonce,
664                    commit_index: index,
665                });
666            }
667            KIND_THREAD_END
668                if outcome == 0
669                    && id.is_empty()
670                    && values.is_empty()
671                    && !matches!(context_id, 0 | u64::MAX) =>
672            {
673                if !thread_end_contexts.insert(context_id) {
674                    return Err(RustTransportError::InvalidRecord(index));
675                }
676                thread_ends.push(RustThreadEnd {
677                    process_id: pid,
678                    context_id,
679                    commit_index: index,
680                });
681            }
682            KIND_TEST_BOUNDARY
683                if outcome == 0
684                    && id.is_empty()
685                    && values.is_empty()
686                    && !matches!(context_id, 0 | u64::MAX) =>
687            {
688                if !boundary_contexts.insert(context_id) {
689                    return Err(RustTransportError::InvalidRecord(index));
690                }
691                test_boundaries.push(RustTestBoundary {
692                    process_id: pid,
693                    context_id,
694                    commit_index: index,
695                });
696            }
697            _ => return Err(RustTransportError::InvalidRecord(index)),
698        }
699    }
700    let overflow = next.saturating_sub(u64::from(descriptors));
701    Ok(RustTransportRead {
702        observations,
703        ordinal_hits,
704        phases,
705        thread_phases,
706        thread_ends,
707        test_boundaries,
708        committed,
709        incomplete: inspect.saturating_sub(committed),
710        dropped: recorded_dropped.max(overflow),
711        attachments,
712    })
713}
714
715pub fn validate_rust_phase_contexts(
716    base_context_id: u64,
717    read: &RustTransportRead,
718) -> Result<(), RustTransportError> {
719    if matches!(base_context_id, 0 | u64::MAX) {
720        return Err(RustTransportError::InvalidAssertionContext(
721            "the supervisor base context must be nonzero and not reserved".into(),
722        ));
723    }
724    let mut definitions = BTreeMap::<u64, (u8, u64, u64, &str)>::new();
725    for phase in &read.phases {
726        let definition = (
727            KIND_PHASE,
728            phase.parent_context_id,
729            phase.invocation_nonce,
730            phase.decision_id.as_str(),
731        );
732        if definitions
733            .insert(phase.child_context_id, definition)
734            .is_some_and(|existing| existing != definition)
735        {
736            return Err(RustTransportError::InvalidAssertionContext(format!(
737                "child {:016x} has conflicting phase definitions",
738                phase.child_context_id
739            )));
740        }
741    }
742    for phase in &read.thread_phases {
743        if rust_thread_context_id(phase.parent_context_id, phase.invocation_nonce)
744            != phase.child_context_id
745        {
746            return Err(RustTransportError::InvalidAssertionContext(format!(
747                "thread phase child {:016x} does not derive from parent {:016x} and nonce {}",
748                phase.child_context_id, phase.parent_context_id, phase.invocation_nonce
749            )));
750        }
751        let definition = (
752            KIND_THREAD_PHASE,
753            phase.parent_context_id,
754            phase.invocation_nonce,
755            "",
756        );
757        if definitions
758            .insert(phase.child_context_id, definition)
759            .is_some_and(|existing| existing != definition)
760        {
761            return Err(RustTransportError::InvalidAssertionContext(format!(
762                "child {:016x} has conflicting phase definitions",
763                phase.child_context_id
764            )));
765        }
766    }
767
768    let used_contexts = read
769        .observations
770        .iter()
771        .map(|observation| observation.context_id)
772        .chain(read.ordinal_hits.iter().map(|hit| hit.context_id))
773        .chain(read.thread_ends.iter().map(|end| end.context_id))
774        .chain(
775            read.test_boundaries
776                .iter()
777                .map(|boundary| boundary.context_id),
778        )
779        .filter(|context| *context != 0 && *context != base_context_id)
780        .collect::<BTreeSet<_>>();
781    for start in used_contexts {
782        let mut context = start;
783        let mut path = BTreeSet::new();
784        while context != base_context_id {
785            if !path.insert(context) {
786                return Err(RustTransportError::InvalidAssertionContext(format!(
787                    "phase context cycle at {context:016x}"
788                )));
789            }
790            let (_, parent, _, _) = definitions.get(&context).ok_or_else(|| {
791                RustTransportError::InvalidAssertionContext(format!(
792                    "context {context:016x} does not resolve to base {base_context_id:016x}"
793                ))
794            })?;
795            context = *parent;
796            if matches!(context, 0 | u64::MAX) {
797                return Err(RustTransportError::InvalidAssertionContext(format!(
798                    "context {start:016x} crosses the supervisor attempt boundary"
799                )));
800            }
801        }
802    }
803    Ok(())
804}
805
806/// Partition one concurrently written artifact transport without copying any
807/// record between tests. Every non-background context must resolve through the
808/// authenticated assertion-phase graph to exactly one known libtest root.
809pub fn partition_rust_transport_by_test_contexts(
810    read: &RustTransportRead,
811    base_contexts: &BTreeSet<u64>,
812) -> Result<RustTransportPartition, RustTransportError> {
813    if base_contexts
814        .iter()
815        .any(|context| matches!(*context, 0 | u64::MAX))
816    {
817        return Err(RustTransportError::InvalidAttribution(
818            "known test roots must be nonzero and non-reserved".into(),
819        ));
820    }
821    if read.incomplete != 0 || read.dropped != 0 {
822        return Err(RustTransportError::InvalidAttribution(format!(
823            "shared transport is incomplete (incomplete={}, dropped={})",
824            read.incomplete, read.dropped
825        )));
826    }
827
828    let mut parents = BTreeMap::<u64, u64>::new();
829    let mut thread_children = BTreeSet::<u64>::new();
830    for (child, parent) in read
831        .phases
832        .iter()
833        .map(|phase| (phase.child_context_id, phase.parent_context_id))
834        .chain(
835            read.thread_phases
836                .iter()
837                .map(|phase| (phase.child_context_id, phase.parent_context_id)),
838        )
839    {
840        if matches!(child, 0 | u64::MAX)
841            || matches!(parent, 0 | u64::MAX)
842            || base_contexts.contains(&child)
843            || parents.insert(child, parent).is_some()
844        {
845            return Err(RustTransportError::InvalidAttribution(format!(
846                "phase {child:016x} has an invalid or repeated ownership definition"
847            )));
848        }
849    }
850    for phase in &read.thread_phases {
851        thread_children.insert(phase.child_context_id);
852    }
853    let mut thread_end_index = BTreeMap::<u64, u64>::new();
854    for end in &read.thread_ends {
855        if !thread_children.contains(&end.context_id)
856            || thread_end_index
857                .insert(end.context_id, end.commit_index)
858                .is_some()
859        {
860            return Err(RustTransportError::InvalidAttribution(format!(
861                "thread end {:016x} has no unique thread-phase definition",
862                end.context_id
863            )));
864        }
865    }
866    let mut boundary_index = BTreeMap::<u64, u64>::new();
867    for boundary in &read.test_boundaries {
868        if !base_contexts.contains(&boundary.context_id)
869            || boundary_index
870                .insert(boundary.context_id, boundary.commit_index)
871                .is_some()
872        {
873            return Err(RustTransportError::InvalidAttribution(format!(
874                "test boundary {:016x} does not identify one known test root",
875                boundary.context_id
876            )));
877        }
878    }
879
880    // One record's exact destination: its root test, safe background, or the
881    // root plus the thread phases whose lifetimes escaped it.
882    enum Resolution {
883        Attributed(u64),
884        Background,
885        Quarantined { root: u64, escaped: Vec<u64> },
886    }
887    let resolve = |start: u64| -> Result<Resolution, RustTransportError> {
888        if start == 0 {
889            return Ok(Resolution::Background);
890        }
891        if start == u64::MAX {
892            return Err(RustTransportError::InvalidAttribution(
893                "evidence used the reserved context sentinel".into(),
894            ));
895        }
896        let mut context = start;
897        let mut seen = BTreeSet::new();
898        let mut chain_threads = Vec::new();
899        loop {
900            if thread_children.contains(&context) {
901                chain_threads.push(context);
902            }
903            let Some(parent) = parents.get(&context) else {
904                break;
905            };
906            if !seen.insert(context) {
907                return Err(RustTransportError::InvalidAttribution(format!(
908                    "phase context cycle at {context:016x}"
909                )));
910            }
911            context = *parent;
912        }
913        if !base_contexts.contains(&context) {
914            return Err(RustTransportError::InvalidAttribution(format!(
915                "context {start:016x} resolves to unknown test root {context:016x}"
916            )));
917        }
918        let boundary = boundary_index.get(&context).copied();
919        let escaped = chain_threads
920            .into_iter()
921            .filter(|thread| {
922                !matches!(
923                    (thread_end_index.get(thread), boundary),
924                    (Some(end), Some(boundary)) if *end < boundary
925                )
926            })
927            .collect::<Vec<_>>();
928        if escaped.is_empty() {
929            Ok(Resolution::Attributed(context))
930        } else {
931            Ok(Resolution::Quarantined {
932                root: context,
933                escaped,
934            })
935        }
936    };
937
938    let mut attributed = base_contexts
939        .iter()
940        .map(|context| (*context, RustTransportRead::empty()))
941        .collect::<BTreeMap<_, _>>();
942    let mut background = RustTransportRead::empty();
943    background.attachments = read.attachments;
944    let mut thread_scope_limitations = BTreeSet::new();
945    let mut note_escape = |root: u64, escaped: Vec<u64>| {
946        for thread in escaped {
947            thread_scope_limitations.insert(format!(
948                "RUST_THREAD_OUTLIVED_TEST: thread phase {thread:016x} escaped test {root:016x}"
949            ));
950        }
951    };
952
953    for observation in &read.observations {
954        match resolve(observation.context_id)? {
955            Resolution::Attributed(root) => attributed
956                .get_mut(&root)
957                .expect("resolved root was preallocated")
958                .observations
959                .push(observation.clone()),
960            Resolution::Background => background.observations.push(observation.clone()),
961            Resolution::Quarantined { root, escaped } => {
962                note_escape(root, escaped);
963                background.observations.push(observation.clone());
964            }
965        }
966    }
967    for hit in &read.ordinal_hits {
968        match resolve(hit.context_id)? {
969            Resolution::Attributed(root) => attributed
970                .get_mut(&root)
971                .expect("resolved root was preallocated")
972                .ordinal_hits
973                .push(*hit),
974            Resolution::Background => background.ordinal_hits.push(*hit),
975            Resolution::Quarantined { root, escaped } => {
976                note_escape(root, escaped);
977                background.ordinal_hits.push(*hit);
978            }
979        }
980    }
981    for phase in &read.phases {
982        match resolve(phase.child_context_id)? {
983            Resolution::Attributed(root) => attributed
984                .get_mut(&root)
985                .expect("resolved root was preallocated")
986                .phases
987                .push(phase.clone()),
988            Resolution::Background => {
989                return Err(RustTransportError::InvalidAttribution(format!(
990                    "phase {:016x} resolved to background",
991                    phase.child_context_id
992                )));
993            }
994            Resolution::Quarantined { root, escaped } => {
995                note_escape(root, escaped);
996                background.phases.push(phase.clone());
997            }
998        }
999    }
1000    for phase in &read.thread_phases {
1001        match resolve(phase.child_context_id)? {
1002            Resolution::Attributed(root) => attributed
1003                .get_mut(&root)
1004                .expect("resolved root was preallocated")
1005                .thread_phases
1006                .push(*phase),
1007            Resolution::Background => {
1008                return Err(RustTransportError::InvalidAttribution(format!(
1009                    "thread phase {:016x} resolved to background",
1010                    phase.child_context_id
1011                )));
1012            }
1013            Resolution::Quarantined { root, escaped } => {
1014                note_escape(root, escaped);
1015                background.thread_phases.push(*phase);
1016            }
1017        }
1018    }
1019    for end in &read.thread_ends {
1020        match resolve(end.context_id)? {
1021            Resolution::Attributed(root) => attributed
1022                .get_mut(&root)
1023                .expect("resolved root was preallocated")
1024                .thread_ends
1025                .push(*end),
1026            Resolution::Background => {
1027                return Err(RustTransportError::InvalidAttribution(format!(
1028                    "thread end {:016x} resolved to background",
1029                    end.context_id
1030                )));
1031            }
1032            Resolution::Quarantined { root, escaped } => {
1033                note_escape(root, escaped);
1034                background.thread_ends.push(*end);
1035            }
1036        }
1037    }
1038    for boundary in &read.test_boundaries {
1039        attributed
1040            .get_mut(&boundary.context_id)
1041            .expect("boundary contexts were checked against the known roots")
1042            .test_boundaries
1043            .push(*boundary);
1044    }
1045
1046    let set_committed = |transport: &mut RustTransportRead| -> Result<u64, RustTransportError> {
1047        transport.committed = u64::try_from(
1048            transport.observations.len()
1049                + transport.ordinal_hits.len()
1050                + transport.phases.len()
1051                + transport.thread_phases.len()
1052                + transport.thread_ends.len()
1053                + transport.test_boundaries.len(),
1054        )
1055        .map_err(|_| {
1056            RustTransportError::InvalidAttribution("partition record count exceeds u64".into())
1057        })?;
1058        Ok(transport.committed)
1059    };
1060    let mut assigned = set_committed(&mut background)?;
1061    for transport in attributed.values_mut() {
1062        assigned = assigned
1063            .checked_add(set_committed(transport)?)
1064            .ok_or_else(|| {
1065                RustTransportError::InvalidAttribution(
1066                    "partition record count overflowed u64".into(),
1067                )
1068            })?;
1069    }
1070    if assigned != read.committed {
1071        return Err(RustTransportError::InvalidAttribution(format!(
1072            "partition assigned {assigned} of {} committed records",
1073            read.committed
1074        )));
1075    }
1076    for (base, transport) in &attributed {
1077        validate_rust_phase_contexts(*base, transport)?;
1078    }
1079    Ok(RustTransportPartition {
1080        attributed,
1081        background,
1082        thread_scope_limitations,
1083    })
1084}
1085
1086pub fn render_rust_mmap_runtime(module_name: &str) -> Result<String, String> {
1087    let valid_identifier = !module_name.is_empty()
1088        && module_name.bytes().enumerate().all(|(index, byte)| {
1089            byte == b'_' || byte.is_ascii_alphabetic() || (index > 0 && byte.is_ascii_digit())
1090        });
1091    if !valid_identifier {
1092        return Err("invalid Rust runtime module name".into());
1093    }
1094    Ok(RUNTIME_TEMPLATE.replace("__SUPERCOV_MODULE__", module_name))
1095}
1096
1097#[cfg(test)]
1098mod tests {
1099    use std::{
1100        fs,
1101        io::{BufRead as _, BufReader},
1102        process::{Command, Stdio},
1103        sync::atomic::{AtomicUsize, Ordering},
1104        time::{SystemTime, UNIX_EPOCH},
1105    };
1106
1107    use super::*;
1108
1109    const TOKEN: [u8; TOKEN_SIZE] = [0x42; TOKEN_SIZE];
1110    const CONTEXT: u64 = 42;
1111
1112    fn token_hex() -> String {
1113        TOKEN.iter().map(|byte| format!("{byte:02x}")).collect()
1114    }
1115
1116    fn temporary_directory(name: &str) -> std::path::PathBuf {
1117        let nonce = SystemTime::now()
1118            .duration_since(UNIX_EPOCH)
1119            .unwrap()
1120            .as_nanos();
1121        let path = std::env::temp_dir().join(format!(
1122            "supercov-rust-transport-{}-{nonce}-{name}",
1123            std::process::id()
1124        ));
1125        fs::create_dir(&path).unwrap();
1126        path
1127    }
1128
1129    fn compile_fixture(directory: &Path) -> std::path::PathBuf {
1130        let source = directory.join("main.rs");
1131        let binary = directory.join("program");
1132        let runtime = render_rust_mmap_runtime("__supercov_runtime_v1").unwrap();
1133        fs::write(
1134            &source,
1135            format!(
1136                r#"{runtime}
1137fn main() {{
1138    let mode = std::env::args().nth(1).unwrap_or_default();
1139    if mode == "threads" {{
1140        let mut threads = Vec::new();
1141        for _ in 0..8 {{
1142            threads.push(std::thread::spawn(|| {{
1143                for _ in 0..100 {{ __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567"); }}
1144            }}));
1145        }}
1146        for thread in threads {{ thread.join().unwrap(); }}
1147        let mut frame = __supercov_runtime_v1::DecisionFrame::new("rs:decision:0123456789abcdef01234567", 2);
1148        let first = __supercov_runtime_v1::condition(true, &mut frame, 0);
1149        let second = __supercov_runtime_v1::condition(false, &mut frame, 1);
1150        __supercov_runtime_v1::decision(first && second, &mut frame);
1151        __supercov_runtime_v1::ordinal_hit(7);
1152    }} else if mode == "contexts" {{
1153        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1154        let outer = __supercov_runtime_v1::enter_context(100);
1155        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1156        let inner = __supercov_runtime_v1::enter_context(200);
1157        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1158        __supercov_runtime_v1::exit_context(inner);
1159        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1160        __supercov_runtime_v1::exit_context(outer);
1161        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1162        let assertion = __supercov_runtime_v1::enter_assertion_context(
1163            0x0123_4567_89ab_cdef,
1164            0x0123_4567,
1165        );
1166        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1167        __supercov_runtime_v1::exit_context(assertion);
1168        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1169        let repeated_assertion = __supercov_runtime_v1::enter_assertion_context(
1170            0x0123_4567_89ab_cdef,
1171            0x0123_4567,
1172        );
1173        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1174        __supercov_runtime_v1::exit_context(repeated_assertion);
1175        __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1176    }} else if mode == "mir-decisions" {{
1177        let before_outer = __supercov_runtime_v1::enter_context(901);
1178        let outer = __supercov_runtime_v1::mir_decision_start(
1179            0x0123_4567_89ab_cdef,
1180            0x0123_4567,
1181            2,
1182        );
1183        __supercov_runtime_v1::mir_decision_condition(outer, 0, true);
1184        let before_inner = __supercov_runtime_v1::enter_context(902);
1185        let inner = __supercov_runtime_v1::mir_decision_start(
1186            0xfedc_ba98_7654_3210,
1187            0xfedc_ba98,
1188            1,
1189        );
1190        let migrated = std::thread::spawn(move || {{
1191            __supercov_runtime_v1::mir_decision_condition(inner, 0, false);
1192            __supercov_runtime_v1::mir_decision_finish(inner, false);
1193        }});
1194        __supercov_runtime_v1::exit_context(before_inner);
1195        __supercov_runtime_v1::mir_decision_condition(outer, 1, true);
1196        __supercov_runtime_v1::exit_context(before_outer);
1197        __supercov_runtime_v1::mir_decision_finish(outer, true);
1198        migrated.join().unwrap();
1199        let before_branch = __supercov_runtime_v1::enter_context(903);
1200        let branch = __supercov_runtime_v1::mir_branch_start();
1201        __supercov_runtime_v1::exit_context(before_branch);
1202        let migrated_branch = std::thread::spawn(move || {{
1203            __supercov_runtime_v1::mir_branch_hit(branch, 777);
1204            __supercov_runtime_v1::mir_branch_hit(branch, 888);
1205        }});
1206        migrated_branch.join().unwrap();
1207    }} else if mode == "kill" {{
1208        __supercov_runtime_v1::hit("rs:function:fedcba9876543210fedcba98");
1209        let interrupted = __supercov_runtime_v1::mir_decision_start(
1210            0x1111_1111_1111_1111,
1211            0x2222_2222,
1212            2,
1213        );
1214        __supercov_runtime_v1::mir_decision_condition(interrupted, 0, true);
1215        let _interrupted_branch = __supercov_runtime_v1::mir_branch_start();
1216        println!("ready");
1217        use std::io::Write as _;
1218        std::io::stdout().flush().unwrap();
1219        std::thread::sleep(std::time::Duration::from_secs(30));
1220    }} else {{
1221        for _ in 0..3 {{ __supercov_runtime_v1::hit("rs:branch:aaaaaaaaaaaaaaaaaaaaaaaa"); }}
1222    }}
1223}}
1224"#
1225            ),
1226        )
1227        .unwrap();
1228        let output = Command::new("rustc")
1229            .args(["--edition=2024"])
1230            .arg(&source)
1231            .arg("-o")
1232            .arg(&binary)
1233            .output()
1234            .unwrap();
1235        assert!(
1236            output.status.success(),
1237            "{}",
1238            String::from_utf8_lossy(&output.stderr)
1239        );
1240        binary
1241    }
1242
1243    #[test]
1244    fn implementation_matches_frozen_transport_contract() {
1245        let contract = supercov_contracts::rust_probe_transport_v3_contract().unwrap();
1246        assert_eq!(MAGIC.as_slice(), contract.magic.as_bytes());
1247        assert_eq!(VERSION, contract.protocol_version);
1248        assert_eq!(HEADER_SIZE, contract.header_size);
1249        assert_eq!(DESCRIPTOR_SIZE, contract.descriptor_size);
1250        assert_eq!(TOKEN_SIZE, contract.token_size);
1251        assert_eq!(ENDIAN_MARKER, contract.endian_marker);
1252        assert_eq!(
1253            NEXT_DESCRIPTOR_OFFSET,
1254            contract.header_offsets.next_descriptor
1255        );
1256        assert_eq!(NEXT_PAYLOAD_OFFSET, contract.header_offsets.next_payload);
1257        assert_eq!(DROPPED_OFFSET, contract.header_offsets.dropped);
1258        assert_eq!(TOKEN_OFFSET, contract.header_offsets.token);
1259        assert_eq!(ATTACHMENTS_OFFSET, contract.header_offsets.attachments);
1260        assert_eq!(Some(NEXT_PHASE_OFFSET), contract.header_offsets.next_phase);
1261        assert_eq!(COMMIT_OFFSET, contract.descriptor_offsets.commit);
1262        assert_eq!(PID_OFFSET, contract.descriptor_offsets.process_id);
1263        assert_eq!(CONTEXT_OFFSET, contract.descriptor_offsets.context_id);
1264        assert_eq!(
1265            PAYLOAD_OFFSET_OFFSET,
1266            contract.descriptor_offsets.payload_offset
1267        );
1268        assert_eq!(CHECKSUM_OFFSET, contract.descriptor_offsets.checksum);
1269        assert_eq!(KIND_HIT, contract.record_kinds.hit);
1270        assert_eq!(KIND_DECISION, contract.record_kinds.decision);
1271        assert_eq!(KIND_ORDINAL_HIT, contract.record_kinds.ordinal_hit);
1272        assert_eq!(Some(KIND_PHASE), contract.record_kinds.phase);
1273        assert_eq!(Some(KIND_THREAD_PHASE), contract.record_kinds.thread_phase);
1274        assert_eq!(Some(KIND_THREAD_END), contract.record_kinds.thread_end);
1275        assert_eq!(
1276            Some(KIND_TEST_BOUNDARY),
1277            contract.record_kinds.test_boundary
1278        );
1279        assert!(RUNTIME_TEMPLATE.contains("b\"SCVRUST3\""));
1280        assert!(RUNTIME_TEMPLATE.contains("const DESCRIPTOR_SIZE: usize = 40;"));
1281    }
1282
1283    #[test]
1284    fn assertion_context_derivation_is_exact_nested_and_never_promotes_background() {
1285        let first =
1286            rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 0).unwrap();
1287        let nested =
1288            rust_assertion_context_id(first, "rs:decision:fedcba9876543210fedcba98", 1).unwrap();
1289        assert_ne!(first, CONTEXT);
1290        assert_ne!(nested, first);
1291        assert_eq!(
1292            rust_assertion_context_id(0, "rs:decision:0123456789abcdef01234567", 2).unwrap(),
1293            0
1294        );
1295        assert!(matches!(
1296            rust_assertion_context_id(CONTEXT, "not-a-decision", 2),
1297            Err(RustTransportError::InvalidAssertionContext(_))
1298        ));
1299        assert!(matches!(
1300            rust_assertion_context_id(u64::MAX, "rs:decision:0123456789abcdef01234567", 2),
1301            Err(RustTransportError::InvalidAssertionContext(_))
1302        ));
1303    }
1304
1305    #[test]
1306    fn phase_context_chains_resolve_exactly_to_the_supervisor_attempt() {
1307        let outer_id = "rs:decision:0123456789abcdef01234567";
1308        let inner_id = "rs:decision:fedcba9876543210fedcba98";
1309        let outer = rust_assertion_context_id(CONTEXT, outer_id, 10).unwrap();
1310        let inner = rust_assertion_context_id(outer, inner_id, 11).unwrap();
1311        let read = RustTransportRead {
1312            observations: vec![RustTransportObservation {
1313                process_id: 7,
1314                context_id: inner,
1315                observation: RustProbeObservation::Hit {
1316                    id: "rs:statement:0123456789abcdef01234567".into(),
1317                },
1318            }],
1319            ordinal_hits: Vec::new(),
1320            phases: vec![
1321                RustPhaseContext {
1322                    process_id: 7,
1323                    child_context_id: outer,
1324                    parent_context_id: CONTEXT,
1325                    invocation_nonce: 10,
1326                    decision_id: outer_id.into(),
1327                },
1328                RustPhaseContext {
1329                    process_id: 7,
1330                    child_context_id: inner,
1331                    parent_context_id: outer,
1332                    invocation_nonce: 11,
1333                    decision_id: inner_id.into(),
1334                },
1335            ],
1336            thread_phases: Vec::new(),
1337            thread_ends: Vec::new(),
1338            test_boundaries: Vec::new(),
1339            committed: 3,
1340            incomplete: 0,
1341            dropped: 0,
1342            attachments: 1,
1343        };
1344        validate_rust_phase_contexts(CONTEXT, &read).unwrap();
1345
1346        let mut threaded = read.clone();
1347        let thread_child = rust_thread_context_id(inner, 7);
1348        threaded.thread_phases.push(RustThreadPhase {
1349            process_id: 7,
1350            child_context_id: thread_child,
1351            parent_context_id: inner,
1352            invocation_nonce: 7,
1353            commit_index: 3,
1354        });
1355        threaded.observations[0].context_id = thread_child;
1356        threaded.committed = 4;
1357        validate_rust_phase_contexts(CONTEXT, &threaded).unwrap();
1358
1359        let mut tampered_thread = threaded.clone();
1360        tampered_thread.thread_phases[0].invocation_nonce = 8;
1361        assert!(matches!(
1362            validate_rust_phase_contexts(CONTEXT, &tampered_thread),
1363            Err(RustTransportError::InvalidAssertionContext(_))
1364        ));
1365
1366        let mut invalid = read.clone();
1367        invalid.phases[0].parent_context_id = 99;
1368        assert!(matches!(
1369            validate_rust_phase_contexts(CONTEXT, &invalid),
1370            Err(RustTransportError::InvalidAssertionContext(_))
1371        ));
1372
1373        let mut unused = read.clone();
1374        let unused_child = rust_assertion_context_id(CONTEXT, inner_id, 12).unwrap();
1375        unused.phases.push(RustPhaseContext {
1376            process_id: 7,
1377            child_context_id: unused_child,
1378            parent_context_id: CONTEXT,
1379            invocation_nonce: 12,
1380            decision_id: inner_id.into(),
1381        });
1382        validate_rust_phase_contexts(CONTEXT, &unused).unwrap();
1383
1384        let mut cycle = read;
1385        cycle.phases[0].parent_context_id = inner;
1386        assert!(matches!(
1387            validate_rust_phase_contexts(CONTEXT, &cycle),
1388            Err(RustTransportError::InvalidAssertionContext(_))
1389        ));
1390    }
1391
1392    #[test]
1393    fn shared_transport_partitions_each_record_once_by_exact_test_root() {
1394        const SECOND: u64 = 84;
1395        let outer_id = "rs:decision:0123456789abcdef01234567";
1396        let inner_id = "rs:decision:fedcba9876543210fedcba98";
1397        let outer = rust_assertion_context_id(CONTEXT, outer_id, 10).unwrap();
1398        let inner = rust_assertion_context_id(outer, inner_id, 11).unwrap();
1399        let read = RustTransportRead {
1400            observations: vec![
1401                RustTransportObservation {
1402                    process_id: 7,
1403                    context_id: inner,
1404                    observation: RustProbeObservation::Hit {
1405                        id: "rs:statement:0123456789abcdef01234567".into(),
1406                    },
1407                },
1408                RustTransportObservation {
1409                    process_id: 7,
1410                    context_id: SECOND,
1411                    observation: RustProbeObservation::Hit {
1412                        id: "rs:function:fedcba9876543210fedcba98".into(),
1413                    },
1414                },
1415                RustTransportObservation {
1416                    process_id: 7,
1417                    context_id: 0,
1418                    observation: RustProbeObservation::Hit {
1419                        id: "rs:statement:aaaaaaaaaaaaaaaaaaaaaaaa".into(),
1420                    },
1421                },
1422            ],
1423            ordinal_hits: vec![RustOrdinalHit {
1424                process_id: 7,
1425                context_id: CONTEXT,
1426                ordinal: 9,
1427            }],
1428            phases: vec![
1429                RustPhaseContext {
1430                    process_id: 7,
1431                    child_context_id: outer,
1432                    parent_context_id: CONTEXT,
1433                    invocation_nonce: 10,
1434                    decision_id: outer_id.into(),
1435                },
1436                RustPhaseContext {
1437                    process_id: 7,
1438                    child_context_id: inner,
1439                    parent_context_id: outer,
1440                    invocation_nonce: 11,
1441                    decision_id: inner_id.into(),
1442                },
1443            ],
1444            thread_phases: Vec::new(),
1445            thread_ends: Vec::new(),
1446            test_boundaries: Vec::new(),
1447            committed: 6,
1448            incomplete: 0,
1449            dropped: 0,
1450            attachments: 3,
1451        };
1452        let bases = BTreeSet::from([CONTEXT, SECOND]);
1453        let partition = partition_rust_transport_by_test_contexts(&read, &bases).unwrap();
1454        assert_eq!(partition.attributed[&CONTEXT].committed, 4);
1455        assert_eq!(partition.attributed[&SECOND].committed, 1);
1456        assert_eq!(partition.background.committed, 1);
1457        assert_eq!(partition.background.attachments, 3);
1458        assert!(partition.thread_scope_limitations.is_empty());
1459        assert_eq!(
1460            partition
1461                .attributed
1462                .values()
1463                .map(|transport| transport.committed)
1464                .sum::<u64>()
1465                + partition.background.committed,
1466            read.committed
1467        );
1468
1469        let mut foreign = read.clone();
1470        foreign.observations[1].context_id = 99;
1471        assert!(matches!(
1472            partition_rust_transport_by_test_contexts(&foreign, &bases),
1473            Err(RustTransportError::InvalidAttribution(_))
1474        ));
1475
1476        let mut incomplete = read.clone();
1477        incomplete.dropped = 1;
1478        assert!(matches!(
1479            partition_rust_transport_by_test_contexts(&incomplete, &bases),
1480            Err(RustTransportError::InvalidAttribution(_))
1481        ));
1482
1483        let mut cycle = read;
1484        cycle.phases[0].parent_context_id = inner;
1485        assert!(matches!(
1486            partition_rust_transport_by_test_contexts(&cycle, &bases),
1487            Err(RustTransportError::InvalidAttribution(_))
1488        ));
1489    }
1490
1491    #[test]
1492    fn thread_phases_are_join_bounded_by_the_test_boundary() {
1493        const SECOND: u64 = 84;
1494        let bases = BTreeSet::from([CONTEXT, SECOND]);
1495        let joined = rust_thread_context_id(CONTEXT, 0);
1496        let escaped = rust_thread_context_id(CONTEXT, 1);
1497        let hit = |context_id: u64| RustTransportObservation {
1498            process_id: 7,
1499            context_id,
1500            observation: RustProbeObservation::Hit {
1501                id: "rs:statement:0123456789abcdef01234567".into(),
1502            },
1503        };
1504        let read = RustTransportRead {
1505            observations: vec![hit(joined), hit(escaped), hit(CONTEXT)],
1506            thread_phases: vec![
1507                RustThreadPhase {
1508                    process_id: 7,
1509                    child_context_id: joined,
1510                    parent_context_id: CONTEXT,
1511                    invocation_nonce: 0,
1512                    commit_index: 0,
1513                },
1514                RustThreadPhase {
1515                    process_id: 7,
1516                    child_context_id: escaped,
1517                    parent_context_id: CONTEXT,
1518                    invocation_nonce: 1,
1519                    commit_index: 1,
1520                },
1521            ],
1522            thread_ends: vec![RustThreadEnd {
1523                process_id: 7,
1524                context_id: joined,
1525                commit_index: 5,
1526            }],
1527            test_boundaries: vec![
1528                RustTestBoundary {
1529                    process_id: 7,
1530                    context_id: CONTEXT,
1531                    commit_index: 6,
1532                },
1533                RustTestBoundary {
1534                    process_id: 7,
1535                    context_id: SECOND,
1536                    commit_index: 7,
1537                },
1538            ],
1539            committed: 8,
1540            attachments: 1,
1541            ..RustTransportRead::empty()
1542        };
1543        let partition = partition_rust_transport_by_test_contexts(&read, &bases).unwrap();
1544        // The joined thread's phase, end and hit are exactly attributed; the
1545        // escaped thread's definition and hit fail closed to background.
1546        assert_eq!(partition.attributed[&CONTEXT].committed, 5);
1547        assert_eq!(partition.attributed[&CONTEXT].thread_phases.len(), 1);
1548        assert_eq!(partition.attributed[&CONTEXT].thread_ends.len(), 1);
1549        assert_eq!(partition.attributed[&CONTEXT].test_boundaries.len(), 1);
1550        assert_eq!(partition.attributed[&SECOND].committed, 1);
1551        assert_eq!(partition.background.committed, 2);
1552        assert_eq!(partition.background.thread_phases.len(), 1);
1553        assert_eq!(
1554            partition.background.observations,
1555            vec![hit(escaped)],
1556            "the escaped thread's record must be background"
1557        );
1558        assert_eq!(
1559            partition.thread_scope_limitations,
1560            BTreeSet::from([format!(
1561                "RUST_THREAD_OUTLIVED_TEST: thread phase {escaped:016x} escaped test {CONTEXT:016x}"
1562            )])
1563        );
1564
1565        // An end that commits after the boundary is escaped, not joined.
1566        let mut late_end = read.clone();
1567        late_end.thread_ends.push(RustThreadEnd {
1568            process_id: 7,
1569            context_id: escaped,
1570            commit_index: 8,
1571        });
1572        late_end.committed = 9;
1573        let partition = partition_rust_transport_by_test_contexts(&late_end, &bases).unwrap();
1574        assert_eq!(partition.thread_scope_limitations.len(), 1);
1575        assert_eq!(partition.background.committed, 3);
1576
1577        // A missing boundary for the root quarantines even ended threads.
1578        let mut unbounded = read.clone();
1579        unbounded.test_boundaries.remove(0);
1580        unbounded.committed = 7;
1581        let partition = partition_rust_transport_by_test_contexts(&unbounded, &bases).unwrap();
1582        assert_eq!(partition.attributed[&CONTEXT].committed, 1);
1583        assert_eq!(partition.thread_scope_limitations.len(), 2);
1584
1585        let mut duplicate_end = read.clone();
1586        duplicate_end.thread_ends.push(RustThreadEnd {
1587            process_id: 7,
1588            context_id: joined,
1589            commit_index: 9,
1590        });
1591        duplicate_end.committed = 9;
1592        assert!(matches!(
1593            partition_rust_transport_by_test_contexts(&duplicate_end, &bases),
1594            Err(RustTransportError::InvalidAttribution(_))
1595        ));
1596
1597        let mut orphan_end = read.clone();
1598        orphan_end.thread_ends.push(RustThreadEnd {
1599            process_id: 7,
1600            context_id: 99,
1601            commit_index: 9,
1602        });
1603        orphan_end.committed = 9;
1604        assert!(matches!(
1605            partition_rust_transport_by_test_contexts(&orphan_end, &bases),
1606            Err(RustTransportError::InvalidAttribution(_))
1607        ));
1608
1609        let mut duplicate_boundary = read.clone();
1610        duplicate_boundary.test_boundaries.push(RustTestBoundary {
1611            process_id: 7,
1612            context_id: CONTEXT,
1613            commit_index: 9,
1614        });
1615        duplicate_boundary.committed = 9;
1616        assert!(matches!(
1617            partition_rust_transport_by_test_contexts(&duplicate_boundary, &bases),
1618            Err(RustTransportError::InvalidAttribution(_))
1619        ));
1620
1621        let mut unknown_boundary = read.clone();
1622        unknown_boundary.test_boundaries.push(RustTestBoundary {
1623            process_id: 7,
1624            context_id: 99,
1625            commit_index: 9,
1626        });
1627        unknown_boundary.committed = 9;
1628        assert!(matches!(
1629            partition_rust_transport_by_test_contexts(&unknown_boundary, &bases),
1630            Err(RustTransportError::InvalidAttribution(_))
1631        ));
1632
1633        let mut duplicate_definition = read;
1634        duplicate_definition.phases.push(RustPhaseContext {
1635            process_id: 7,
1636            child_context_id: joined,
1637            parent_context_id: CONTEXT,
1638            invocation_nonce: 0,
1639            decision_id: "rs:decision:0123456789abcdef01234567".into(),
1640        });
1641        duplicate_definition.committed = 9;
1642        assert!(matches!(
1643            partition_rust_transport_by_test_contexts(&duplicate_definition, &bases),
1644            Err(RustTransportError::InvalidAttribution(_))
1645        ));
1646    }
1647
1648    fn append_record(
1649        bytes: &mut [u8],
1650        kind: u8,
1651        outcome: u8,
1652        context_id: u64,
1653        id: &[u8],
1654        values: &[u8],
1655    ) {
1656        let descriptor_capacity = get_u32(bytes, 20).unwrap() as usize;
1657        let payload_base = HEADER_SIZE + descriptor_capacity * DESCRIPTOR_SIZE;
1658        let next = get_u64(bytes, NEXT_DESCRIPTOR_OFFSET).unwrap();
1659        let next_payload = get_u64(bytes, NEXT_PAYLOAD_OFFSET).unwrap();
1660        let descriptor = HEADER_SIZE + usize::try_from(next).unwrap() * DESCRIPTOR_SIZE;
1661        let payload_length = (id.len() + values.len()) as u32;
1662        bytes[descriptor + KIND_OFFSET] = kind;
1663        bytes[descriptor + OUTCOME_OFFSET] = outcome;
1664        bytes[descriptor + 3] = 0;
1665        put_u32(bytes, descriptor + PID_OFFSET, 7);
1666        bytes[descriptor + CONTEXT_OFFSET..descriptor + CONTEXT_OFFSET + 8]
1667            .copy_from_slice(&context_id.to_le_bytes());
1668        put_u32(
1669            bytes,
1670            descriptor + PAYLOAD_OFFSET_OFFSET,
1671            next_payload as u32,
1672        );
1673        put_u32(bytes, descriptor + PAYLOAD_LENGTH_OFFSET, payload_length);
1674        put_u32(bytes, descriptor + ID_LENGTH_OFFSET, id.len() as u32);
1675        put_u32(bytes, descriptor + VALUE_LENGTH_OFFSET, values.len() as u32);
1676        let record_checksum = checksum(
1677            kind,
1678            outcome,
1679            7,
1680            context_id,
1681            next_payload as u32,
1682            payload_length,
1683            id.len() as u32,
1684            values.len() as u32,
1685            id,
1686            values,
1687        );
1688        bytes[descriptor + CHECKSUM_OFFSET..descriptor + CHECKSUM_OFFSET + 8]
1689            .copy_from_slice(&record_checksum.to_le_bytes());
1690        let payload = payload_base + usize::try_from(next_payload).unwrap();
1691        bytes[payload..payload + id.len()].copy_from_slice(id);
1692        bytes[payload + id.len()..payload + id.len() + values.len()].copy_from_slice(values);
1693        bytes[descriptor + COMMIT_OFFSET] = 1;
1694        bytes[NEXT_DESCRIPTOR_OFFSET..NEXT_DESCRIPTOR_OFFSET + 8]
1695            .copy_from_slice(&(next + 1).to_le_bytes());
1696        bytes[NEXT_PAYLOAD_OFFSET..NEXT_PAYLOAD_OFFSET + 8]
1697            .copy_from_slice(&(next_payload + u64::from(payload_length)).to_le_bytes());
1698    }
1699
1700    #[test]
1701    fn thread_kind_records_are_authenticated_unique_and_strict() {
1702        let directory = temporary_directory("thread-kinds");
1703        let child = rust_thread_context_id(CONTEXT, 3);
1704        let mut definition = [0_u8; 16];
1705        definition[..8].copy_from_slice(&CONTEXT.to_le_bytes());
1706        definition[8..].copy_from_slice(&3_u64.to_le_bytes());
1707
1708        type SyntheticRecord<'bytes> = (u8, u8, u64, &'bytes [u8], &'bytes [u8]);
1709        let case = AtomicUsize::new(0);
1710        let build = |records: &[SyntheticRecord<'_>]| {
1711            let path = directory.join(format!(
1712                "case-{}.transport",
1713                case.fetch_add(1, Ordering::Relaxed)
1714            ));
1715            create_rust_transport(&path, TOKEN, 16, 4_096).unwrap();
1716            let mut bytes = fs::read(&path).unwrap();
1717            for (kind, outcome, context, id, values) in records {
1718                append_record(&mut bytes, *kind, *outcome, *context, id, values);
1719            }
1720            fs::write(&path, bytes).unwrap();
1721            path
1722        };
1723
1724        let valid = build(&[
1725            (KIND_THREAD_PHASE, 0, child, b"", &definition),
1726            (KIND_THREAD_END, 0, child, b"", b""),
1727            (KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b""),
1728        ]);
1729        let read = read_rust_transport(&valid, &TOKEN).unwrap();
1730        assert_eq!(read.committed, 3);
1731        assert_eq!(
1732            read.thread_phases,
1733            vec![RustThreadPhase {
1734                process_id: 7,
1735                child_context_id: child,
1736                parent_context_id: CONTEXT,
1737                invocation_nonce: 3,
1738                commit_index: 0,
1739            }]
1740        );
1741        assert_eq!(
1742            read.thread_ends,
1743            vec![RustThreadEnd {
1744                process_id: 7,
1745                context_id: child,
1746                commit_index: 1,
1747            }]
1748        );
1749        assert_eq!(
1750            read.test_boundaries,
1751            vec![RustTestBoundary {
1752                process_id: 7,
1753                context_id: CONTEXT,
1754                commit_index: 2,
1755            }]
1756        );
1757
1758        // A thread phase whose context does not derive from its payload fails
1759        // authentication like a tampered assertion phase.
1760        let tampered = build(&[(KIND_THREAD_PHASE, 0, child ^ 1, b"", &definition)]);
1761        assert!(matches!(
1762            read_rust_transport(&tampered, &TOKEN),
1763            Err(RustTransportError::InvalidAssertionContext(_))
1764        ));
1765
1766        let conflicting_definition = {
1767            let mut other = [0_u8; 16];
1768            other[..8].copy_from_slice(&CONTEXT.to_le_bytes());
1769            other[8..].copy_from_slice(&4_u64.to_le_bytes());
1770            build(&[
1771                (KIND_THREAD_PHASE, 0, child, b"", &definition),
1772                (KIND_THREAD_PHASE, 0, child, b"", &other),
1773            ])
1774        };
1775        assert!(matches!(
1776            read_rust_transport(&conflicting_definition, &TOKEN),
1777            Err(RustTransportError::InvalidAssertionContext(_))
1778        ));
1779
1780        let duplicate_end = build(&[
1781            (KIND_THREAD_PHASE, 0, child, b"", &definition),
1782            (KIND_THREAD_END, 0, child, b"", b""),
1783            (KIND_THREAD_END, 0, child, b"", b""),
1784        ]);
1785        assert_eq!(
1786            read_rust_transport(&duplicate_end, &TOKEN),
1787            Err(RustTransportError::InvalidRecord(2))
1788        );
1789
1790        let duplicate_boundary = build(&[
1791            (KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b""),
1792            (KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b""),
1793        ]);
1794        assert_eq!(
1795            read_rust_transport(&duplicate_boundary, &TOKEN),
1796            Err(RustTransportError::InvalidRecord(1))
1797        );
1798
1799        for invalid in [
1800            // Zero and reserved contexts are never valid thread-kind contexts.
1801            build(&[(KIND_THREAD_END, 0, 0, b"", b"")]),
1802            build(&[(KIND_TEST_BOUNDARY, 0, u64::MAX, b"", b"")]),
1803            // Thread-kind records carry no probe identity or extra payload.
1804            build(&[(KIND_THREAD_END, 0, CONTEXT, b"x", b"")]),
1805            build(&[(KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b"y")]),
1806            build(&[(KIND_THREAD_PHASE, 0, child, b"", &definition[..8])]),
1807            // Outcomes are meaningless for thread-kind records.
1808            build(&[(KIND_THREAD_END, 1, CONTEXT, b"", b"")]),
1809        ] {
1810            assert_eq!(
1811                read_rust_transport(&invalid, &TOKEN),
1812                Err(RustTransportError::InvalidRecord(0))
1813            );
1814        }
1815
1816        fs::remove_dir_all(directory).unwrap();
1817    }
1818
1819    /// Off macOS, Linux and Windows the runtime template compiles to its
1820    /// stub, whose probes record nothing and attach to nothing. That is the
1821    /// premise of `run_direct_rust` refusing there: the stub must build, and
1822    /// it must not pretend to measure.
1823    #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))]
1824    #[test]
1825    fn runtime_stub_compiles_and_records_nothing_elsewhere() {
1826        let directory = temporary_directory("stub");
1827        let binary = compile_fixture(&directory);
1828        let transport = directory.join("stub.transport");
1829        create_rust_transport(&transport, TOKEN, 1_024, 128 * 1024).unwrap();
1830        let output = Command::new(&binary)
1831            .arg("threads")
1832            .env(RUST_TRANSPORT_ENV, &transport)
1833            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1834            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1835            .output()
1836            .unwrap();
1837        assert!(output.status.success());
1838        let read = read_rust_transport(&transport, &TOKEN).unwrap();
1839        assert_eq!(read.committed, 0);
1840        assert_eq!(read.attachments, 0);
1841        fs::remove_dir_all(directory).unwrap();
1842    }
1843
1844    #[cfg(any(target_os = "macos", target_os = "linux", target_os = "windows"))]
1845    #[test]
1846    fn mmap_transport_is_concurrent_bounded_strict_and_kill_resilient() {
1847        let directory = temporary_directory("all");
1848        let binary = compile_fixture(&directory);
1849
1850        let concurrent = directory.join("concurrent.transport");
1851        create_rust_transport(&concurrent, TOKEN, 1_024, 128 * 1024).unwrap();
1852        let output = Command::new(&binary)
1853            .arg("threads")
1854            .env(RUST_TRANSPORT_ENV, &concurrent)
1855            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1856            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1857            .output()
1858            .unwrap();
1859        assert!(output.status.success());
1860        let read = read_rust_transport(&concurrent, &TOKEN).unwrap();
1861        assert_eq!(read.committed, 818);
1862        assert_eq!(read.incomplete, 0);
1863        assert_eq!(read.dropped, 0);
1864        assert_eq!(read.attachments, 1);
1865        assert_eq!(read.observations.len(), 801);
1866        assert!(
1867            matches!(read.observations.last(), Some(RustTransportObservation { context_id: CONTEXT, observation: RustProbeObservation::Decision { values, outcome: false, .. }, .. }) if values == &[Some(true), Some(false)])
1868        );
1869        // Every spawned thread runs under its own derived thread-phase context
1870        // parented to the creating context; the main thread stays exact.
1871        assert_eq!(read.thread_phases.len(), 8);
1872        assert!(
1873            read.thread_phases
1874                .iter()
1875                .all(|phase| phase.parent_context_id == CONTEXT)
1876        );
1877        assert_eq!(read.thread_ends.len(), 8);
1878        let thread_contexts = read
1879            .thread_phases
1880            .iter()
1881            .map(|phase| phase.child_context_id)
1882            .collect::<BTreeSet<_>>();
1883        assert_eq!(thread_contexts.len(), 8);
1884        assert_eq!(
1885            read.thread_ends
1886                .iter()
1887                .map(|end| end.context_id)
1888                .collect::<BTreeSet<_>>(),
1889            thread_contexts
1890        );
1891        assert!(read.observations.iter().all(|item| {
1892            item.context_id == CONTEXT || thread_contexts.contains(&item.context_id)
1893        }));
1894        assert_eq!(
1895            read.observations
1896                .iter()
1897                .filter(|item| thread_contexts.contains(&item.context_id))
1898                .count(),
1899            800
1900        );
1901        validate_rust_phase_contexts(CONTEXT, &read).unwrap();
1902        assert_eq!(read.ordinal_hits.len(), 1);
1903        assert_eq!(read.ordinal_hits[0].ordinal, 7);
1904        assert_eq!(read.ordinal_hits[0].context_id, CONTEXT);
1905        assert_eq!(
1906            read_rust_transport(&concurrent, &[0x43; TOKEN_SIZE]),
1907            Err(RustTransportError::InvalidHeader),
1908            "a supervisor must never accept evidence from another task token"
1909        );
1910
1911        let mir_decisions = directory.join("mir-decisions.transport");
1912        create_rust_transport(&mir_decisions, TOKEN, 16, 4_096).unwrap();
1913        let output = Command::new(&binary)
1914            .arg("mir-decisions")
1915            .env(RUST_TRANSPORT_ENV, &mir_decisions)
1916            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1917            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1918            .output()
1919            .unwrap();
1920        assert!(output.status.success());
1921        let read = read_rust_transport(&mir_decisions, &TOKEN).unwrap();
1922        assert_eq!(
1923            read.committed, 7,
1924            "two migrated threads add one thread phase and one thread end each"
1925        );
1926        assert_eq!(read.thread_phases.len(), 2);
1927        assert_eq!(read.thread_ends.len(), 2);
1928        assert_eq!(read.dropped, 0);
1929        assert_eq!(read.incomplete, 0);
1930        assert!(read.observations.iter().any(|observation| matches!(
1931            observation,
1932            RustTransportObservation {
1933                context_id: 901,
1934                observation: RustProbeObservation::Decision { id, values, outcome: true },
1935                ..
1936            }
1937                if id == "rs:decision:0123456789abcdef01234567"
1938                    && values == &[Some(true), Some(true)]
1939        )));
1940        assert!(
1941            read.ordinal_hits
1942                .iter()
1943                .any(|hit| hit.context_id == 903 && hit.ordinal == 777)
1944        );
1945        assert!(!read.ordinal_hits.iter().any(|hit| hit.ordinal == 888));
1946        assert!(read.observations.iter().any(|observation| matches!(
1947            observation,
1948            RustTransportObservation {
1949                context_id: 902,
1950                observation: RustProbeObservation::Decision { id, values, outcome: false },
1951                ..
1952            }
1953                if id == "rs:decision:fedcba9876543210fedcba98"
1954                    && values == &[Some(false)]
1955        )));
1956
1957        let processes = directory.join("processes.transport");
1958        create_rust_transport(&processes, TOKEN, 64, 8 * 1024).unwrap();
1959        let mut children = (0..8)
1960            .map(|_| {
1961                Command::new(&binary)
1962                    .env(RUST_TRANSPORT_ENV, &processes)
1963                    .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1964                    .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1965                    .spawn()
1966                    .unwrap()
1967            })
1968            .collect::<Vec<_>>();
1969        for child in &mut children {
1970            assert!(child.wait().unwrap().success());
1971        }
1972        let read = read_rust_transport(&processes, &TOKEN).unwrap();
1973        assert_eq!(read.attachments, 8);
1974        assert_eq!(read.committed, 24);
1975        assert_eq!(read.incomplete, 0);
1976        assert_eq!(read.dropped, 0);
1977        assert_eq!(
1978            read.observations
1979                .iter()
1980                .map(|item| item.process_id)
1981                .collect::<std::collections::BTreeSet<_>>()
1982                .len(),
1983            8
1984        );
1985
1986        let nested = directory.join("nested-context.transport");
1987        create_rust_transport(&nested, TOKEN, 16, 4_096).unwrap();
1988        let output = Command::new(&binary)
1989            .arg("contexts")
1990            .env(RUST_TRANSPORT_ENV, &nested)
1991            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1992            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1993            .output()
1994            .unwrap();
1995        assert!(output.status.success());
1996        let read = read_rust_transport(&nested, &TOKEN).unwrap();
1997        assert_eq!(
1998            read.observations
1999                .iter()
2000                .map(|item| item.context_id)
2001                .collect::<Vec<_>>(),
2002            [
2003                CONTEXT,
2004                100,
2005                200,
2006                100,
2007                CONTEXT,
2008                rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 0)
2009                    .unwrap(),
2010                CONTEXT,
2011                rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 1)
2012                    .unwrap(),
2013                CONTEXT,
2014            ]
2015        );
2016        let assertion =
2017            rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 0).unwrap();
2018        assert_eq!(read.phases.len(), 2);
2019        assert_eq!(read.phases[0].child_context_id, assertion);
2020        assert_eq!(read.phases[0].parent_context_id, CONTEXT);
2021        assert_eq!(read.phases[0].invocation_nonce, 0);
2022        assert_eq!(
2023            read.phases[0].decision_id,
2024            "rs:decision:0123456789abcdef01234567"
2025        );
2026        assert_ne!(
2027            read.phases[0].child_context_id,
2028            read.phases[1].child_context_id
2029        );
2030        assert_eq!(read.phases[1].parent_context_id, CONTEXT);
2031        assert_eq!(read.phases[1].invocation_nonce, 1);
2032        assert_eq!(read.phases[1].decision_id, read.phases[0].decision_id);
2033        assert_eq!(read.committed, 11);
2034
2035        let invalid_phase = directory.join("invalid-phase.transport");
2036        fs::copy(&nested, &invalid_phase).unwrap();
2037        let mut bytes = fs::read(&invalid_phase).unwrap();
2038        let descriptor_capacity = get_u32(&bytes, 20).unwrap();
2039        let payload_base = HEADER_SIZE + descriptor_capacity as usize * DESCRIPTOR_SIZE;
2040        let descriptor = (0..read.committed as usize)
2041            .map(|index| HEADER_SIZE + index * DESCRIPTOR_SIZE)
2042            .find(|offset| bytes[*offset + KIND_OFFSET] == KIND_PHASE)
2043            .unwrap();
2044        let payload_offset = get_u32(&bytes, descriptor + PAYLOAD_OFFSET_OFFSET).unwrap();
2045        let payload_length = get_u32(&bytes, descriptor + PAYLOAD_LENGTH_OFFSET).unwrap();
2046        let id_length = get_u32(&bytes, descriptor + ID_LENGTH_OFFSET).unwrap();
2047        let value_length = get_u32(&bytes, descriptor + VALUE_LENGTH_OFFSET).unwrap();
2048        let payload = payload_base + payload_offset as usize;
2049        let id = bytes[payload..payload + id_length as usize].to_vec();
2050        let values =
2051            bytes[payload + id_length as usize..payload + payload_length as usize].to_vec();
2052        let invalid_child = 7_u64;
2053        bytes[descriptor + CONTEXT_OFFSET..descriptor + CONTEXT_OFFSET + 8]
2054            .copy_from_slice(&invalid_child.to_le_bytes());
2055        let replacement_checksum = checksum(
2056            KIND_PHASE,
2057            0,
2058            get_u32(&bytes, descriptor + PID_OFFSET).unwrap(),
2059            invalid_child,
2060            payload_offset,
2061            payload_length,
2062            id_length,
2063            value_length,
2064            &id,
2065            &values,
2066        );
2067        bytes[descriptor + CHECKSUM_OFFSET..descriptor + CHECKSUM_OFFSET + 8]
2068            .copy_from_slice(&replacement_checksum.to_le_bytes());
2069        fs::write(&invalid_phase, bytes).unwrap();
2070        assert!(matches!(
2071            read_rust_transport(&invalid_phase, &TOKEN),
2072            Err(RustTransportError::InvalidAssertionContext(_))
2073        ));
2074
2075        let rejected = directory.join("rejected-token.transport");
2076        create_rust_transport(&rejected, TOKEN, 16, 4_096).unwrap();
2077        let output = Command::new(&binary)
2078            .env(RUST_TRANSPORT_ENV, &rejected)
2079            .env(RUST_TRANSPORT_TOKEN_ENV, "00".repeat(TOKEN_SIZE))
2080            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2081            .output()
2082            .unwrap();
2083        assert!(output.status.success());
2084        let read = read_rust_transport(&rejected, &TOKEN).unwrap();
2085        assert_eq!(read.attachments, 0);
2086        assert_eq!(read.committed, 0);
2087        assert_eq!(read.dropped, 0);
2088
2089        let malformed_context = directory.join("malformed-context.transport");
2090        create_rust_transport(&malformed_context, TOKEN, 16, 4_096).unwrap();
2091        let output = Command::new(&binary)
2092            .env(RUST_TRANSPORT_ENV, &malformed_context)
2093            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2094            .env(RUST_CONTEXT_ENV, "not-a-context")
2095            .output()
2096            .unwrap();
2097        assert!(output.status.success());
2098        let read = read_rust_transport(&malformed_context, &TOKEN).unwrap();
2099        assert_eq!(read.attachments, 0);
2100        assert_eq!(read.committed, 0);
2101
2102        let bounded = directory.join("bounded.transport");
2103        create_rust_transport(&bounded, TOKEN, 2, 256).unwrap();
2104        let output = Command::new(&binary)
2105            .env(RUST_TRANSPORT_ENV, &bounded)
2106            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2107            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2108            .output()
2109            .unwrap();
2110        assert!(output.status.success());
2111        let read = read_rust_transport(&bounded, &TOKEN).unwrap();
2112        assert_eq!(read.committed, 2);
2113        assert_eq!(read.dropped, 1);
2114
2115        let payload_bounded = directory.join("payload-bounded.transport");
2116        create_rust_transport(&payload_bounded, TOKEN, 8, 16).unwrap();
2117        let output = Command::new(&binary)
2118            .env(RUST_TRANSPORT_ENV, &payload_bounded)
2119            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2120            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2121            .output()
2122            .unwrap();
2123        assert!(output.status.success());
2124        let read = read_rust_transport(&payload_bounded, &TOKEN).unwrap();
2125        assert_eq!(read.committed, 0);
2126        assert_eq!(read.incomplete, 3);
2127        assert_eq!(read.dropped, 3);
2128
2129        let killed = directory.join("killed.transport");
2130        create_rust_transport(&killed, TOKEN, 16, 4_096).unwrap();
2131        let mut child = Command::new(&binary)
2132            .arg("kill")
2133            .env(RUST_TRANSPORT_ENV, &killed)
2134            .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2135            .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2136            .stdout(Stdio::piped())
2137            .spawn()
2138            .unwrap();
2139        let mut ready = String::new();
2140        BufReader::new(child.stdout.take().unwrap())
2141            .read_line(&mut ready)
2142            .unwrap();
2143        assert_eq!(ready, "ready\n");
2144        child.kill().unwrap();
2145        child.wait().unwrap();
2146        let read = read_rust_transport(&killed, &TOKEN).unwrap();
2147        assert_eq!(read.committed, 1);
2148        assert_eq!(read.incomplete, 2);
2149        assert_eq!(read.dropped, 0);
2150        assert_eq!(
2151            read.observations,
2152            [RustTransportObservation {
2153                process_id: read.observations[0].process_id,
2154                context_id: CONTEXT,
2155                observation: RustProbeObservation::Hit {
2156                    id: "rs:function:fedcba9876543210fedcba98".into()
2157                }
2158            }]
2159        );
2160
2161        let corrupt = directory.join("corrupt.transport");
2162        create_rust_transport(&corrupt, TOKEN, 2, 256).unwrap();
2163        let mut bytes = fs::read(&corrupt).unwrap();
2164        bytes[0] ^= 1;
2165        fs::write(&corrupt, bytes).unwrap();
2166        assert_eq!(
2167            read_rust_transport(&corrupt, &TOKEN),
2168            Err(RustTransportError::InvalidHeader)
2169        );
2170
2171        let invalid_commit = directory.join("invalid-commit.transport");
2172        fs::copy(&bounded, &invalid_commit).unwrap();
2173        let mut bytes = fs::read(&invalid_commit).unwrap();
2174        bytes[HEADER_SIZE + COMMIT_OFFSET] = 2;
2175        fs::write(&invalid_commit, bytes).unwrap();
2176        assert_eq!(
2177            read_rust_transport(&invalid_commit, &TOKEN),
2178            Err(RustTransportError::InvalidDescriptor(0))
2179        );
2180
2181        let invalid_flags = directory.join("invalid-flags.transport");
2182        fs::copy(&bounded, &invalid_flags).unwrap();
2183        let mut bytes = fs::read(&invalid_flags).unwrap();
2184        bytes[HEADER_SIZE + 3] = 1;
2185        fs::write(&invalid_flags, bytes).unwrap();
2186        assert_eq!(
2187            read_rust_transport(&invalid_flags, &TOKEN),
2188            Err(RustTransportError::InvalidDescriptor(0))
2189        );
2190
2191        let invalid_checksum = directory.join("invalid-checksum.transport");
2192        fs::copy(&bounded, &invalid_checksum).unwrap();
2193        let mut bytes = fs::read(&invalid_checksum).unwrap();
2194        bytes[HEADER_SIZE + CHECKSUM_OFFSET] ^= 1;
2195        fs::write(&invalid_checksum, bytes).unwrap();
2196        assert_eq!(
2197            read_rust_transport(&invalid_checksum, &TOKEN),
2198            Err(RustTransportError::InvalidRecord(0))
2199        );
2200
2201        #[cfg(unix)]
2202        {
2203            use std::os::unix::fs::symlink;
2204
2205            let linked = directory.join("linked.transport");
2206            symlink(&bounded, &linked).unwrap();
2207            assert_eq!(
2208                read_rust_transport(&linked, &TOKEN),
2209                Err(RustTransportError::UnsafeFile(linked.display().to_string()))
2210            );
2211        }
2212
2213        let truncated = directory.join("truncated.transport");
2214        create_rust_transport(&truncated, TOKEN, 2, 256).unwrap();
2215        OpenOptions::new()
2216            .write(true)
2217            .open(&truncated)
2218            .unwrap()
2219            .set_len(63)
2220            .unwrap();
2221        assert_eq!(
2222            read_rust_transport(&truncated, &TOKEN),
2223            Err(RustTransportError::InvalidHeader)
2224        );
2225
2226        let incomplete = directory.join("incomplete.transport");
2227        create_rust_transport(&incomplete, TOKEN, 2, 256).unwrap();
2228        let file = OpenOptions::new()
2229            .read(true)
2230            .write(true)
2231            .open(&incomplete)
2232            .unwrap();
2233        let mapping = map_mut(&file).unwrap();
2234        atomic_u64(&mapping, NEXT_DESCRIPTOR_OFFSET)
2235            .unwrap()
2236            .store(1, Ordering::Release);
2237        mapping.flush().unwrap();
2238        drop(mapping);
2239        let read = read_rust_transport(&incomplete, &TOKEN).unwrap();
2240        assert_eq!(read.committed, 0);
2241        assert_eq!(read.incomplete, 1);
2242        assert!(read.observations.is_empty());
2243        assert!(read.ordinal_hits.is_empty());
2244
2245        fs::remove_dir_all(directory).unwrap();
2246    }
2247}