feat(session-types): the fly-session-types crate: domain scalars, closed enums, method payloads, canonical JSON and digests

The CONTRACT-01 type definitions, committed early so the session slice can build on
them while the fixtures and cross-language tests are still being written.

- scalar: Scope, RationalNs (reduced, positive denominator, 0/1 zero, checked
  arithmetic and the step-v1 accumulator), SchemaRef, TypedValue with its 32-KiB
  canonical-JSON cap, and the four identities that must not be confused (bus callId,
  domain requestId, artifact identity, delivery/hold owner token) as distinct types.
  Id, U64 and Digest call into flybus::wire instead of forking the encodings.
- canonical: RFC 8785 canonical JSON, SHA-256 digests, the operation key and the
  canonical body rules of ipc-v1 section 5, with bus identities refused in a body.
- rpc, workers, media, publishing: the payloads of ipc-v1, workers-v1, state-media-v1
  and publishing-v1, each with a validate step for the documented ranges, uniqueness,
  ordering and bounds.
- schema: the canonical schema set, whose digest is contractDigest, declared as data
  so source formatting cannot change it.
- trace: the step-v1 section 8 record, behaviour separated from operational metadata.
- seed, checkpoint: the seed derivation and FLYSESS1 envelope layout.

services/flysim/Cargo.lock

@@ -416,6 +416,16 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "fly-session-types"
+version = "0.1.1"
+dependencies = [
+ "flybus",
+ "ryu-js",
+ "serde_json",
+ "sha2",
+]
+
 [[package]]
 name = "flybrain-core"
 version = "0.1.1"

services/flysim/Cargo.toml

@@ -1,6 +1,12 @@
 [workspace]
 resolver = "3"
-members = ["crates/flybrain-core", "crates/flybrain-gb", "crates/flybus", "crates/flysim"]
+members = [
+    "crates/fly-session-types",
+    "crates/flybrain-core",
+    "crates/flybrain-gb",
+    "crates/flybus",
+    "crates/flysim",
+]
 
 [workspace.package]
 version = "0.1.1"

services/flysim/crates/fly-session-types/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "fly-session-types"
+version.workspace = true
+edition = "2024"
+rust-version.workspace = true
+license.workspace = true
+publish = false
+description = "Session domain scalars, closed enums, method payloads, canonical JSON digests and the step trace format (session-framework CONTRACT-01)."
+
+[dependencies]
+# The bus owns the Id/U64/Digest encodings, strict JSON and ArtifactRef; this crate reuses
+# them rather than forking their semantics.
+flybus = { path = "../flybus" }
+ryu-js.workspace = true
+serde_json = { workspace = true }
+sha2 = { workspace = true }

services/flysim/crates/fly-session-types/src/canonical.rs

@@ -0,0 +1,273 @@
+//! Canonical JSON (RFC 8785) and the digest rules of ipc-v1 section 5.
+//!
+//! One serialization, two languages: keys sorted by UTF-16 code unit, numbers printed by the
+//! ECMAScript `Number::toString` algorithm (so a JavaScript `JSON.stringify` over the same
+//! sorted tree produces the same bytes), strings escaped the way `JSON.stringify` escapes
+//! them, no insignificant whitespace. A digest is the SHA-256 of those bytes, lowercase hex.
+//!
+//! Numbers outside the exactly representable double range are refused rather than rounded:
+//! every counter and clock in these contracts is a `U64` decimal string, so a JSON number
+//! larger than 2^53-1 is a schema error, not something to canonicalize approximately.
+
+use serde_json::{Number, Value};
+use sha2::{Digest as _, Sha256};
+
+use crate::scalar::{Result, Scope, err, wire_err};
+
+/// The largest integer a double represents exactly.
+pub const MAX_EXACT_INTEGER: i64 = 9_007_199_254_740_991;
+
+/// The bus envelope ceiling every domain message must also fit (bus-v1 section 4).
+pub const MAX_ENVELOPE_BYTES: usize = flybus::wire::MAX_ENVELOPE_BYTES;
+
+/// The `f64` a JSON number denotes, or `None` if it is not a finite exactly representable one.
+pub fn finite_double(n: &Number) -> Option<f64> {
+    if let Some(u) = n.as_u64() {
+        return (u <= MAX_EXACT_INTEGER as u64).then_some(u as f64);
+    }
+    if let Some(i) = n.as_i64() {
+        return (i >= -MAX_EXACT_INTEGER).then_some(i as f64);
+    }
+    n.as_f64().filter(|v| v.is_finite())
+}
+
+/// `String(number)` for a finite double, the ECMAScript algorithm RFC 8785 requires.
+fn number_to_string(value: f64) -> String {
+    if value == 0.0 {
+        // Covers -0.0, which `JSON.stringify` prints as "0".
+        return "0".to_owned();
+    }
+    let mut buffer = ryu_js::Buffer::new();
+    buffer.format(value).to_owned()
+}
+
+/// Escapes one string the way `JSON.stringify` does.
+fn write_string(out: &mut String, s: &str) {
+    out.push('"');
+    for c in s.chars() {
+        match c {
+            '"' => out.push_str("\\\""),
+            '\\' => out.push_str("\\\\"),
+            '\u{08}' => out.push_str("\\b"),
+            '\u{09}' => out.push_str("\\t"),
+            '\u{0a}' => out.push_str("\\n"),
+            '\u{0c}' => out.push_str("\\f"),
+            '\u{0d}' => out.push_str("\\r"),
+            c if (c as u32) < 0x20 => out.push_str(&format!("\\u{:04x}", c as u32)),
+            c => out.push(c),
+        }
+    }
+    out.push('"');
+}
+
+/// Sorts object keys by UTF-16 code unit, as RFC 8785 section 3.2.3 specifies.
+fn utf16_key(key: &str) -> Vec<u16> {
+    key.encode_utf16().collect()
+}
+
+/// The canonical JSON text of `value`.
+pub fn canonicalize(value: &Value) -> Result<String> {
+    let mut out = String::new();
+    write_value(&mut out, value)?;
+    Ok(out)
+}
+
+/// The canonical JSON bytes of `value`.
+pub fn canonical_bytes(value: &Value) -> Result<Vec<u8>> {
+    canonicalize(value).map(String::into_bytes)
+}
+
+fn write_value(out: &mut String, value: &Value) -> Result<()> {
+    match value {
+        Value::Null => out.push_str("null"),
+        Value::Bool(true) => out.push_str("true"),
+        Value::Bool(false) => out.push_str("false"),
+        Value::Number(n) => {
+            let d = finite_double(n).ok_or_else(|| {
+                wire_err(format!(
+                    "canonical JSON: {n} is not a finite number in the exact double range"
+                ))
+            })?;
+            out.push_str(&number_to_string(d));
+        }
+        Value::String(s) => write_string(out, s),
+        Value::Array(items) => {
+            out.push('[');
+            for (i, item) in items.iter().enumerate() {
+                if i > 0 {
+                    out.push(',');
+                }
+                write_value(out, item)?;
+            }
+            out.push(']');
+        }
+        Value::Object(map) => {
+            let mut keys: Vec<&String> = map.keys().collect();
+            keys.sort_by_cached_key(|k| utf16_key(k));
+            out.push('{');
+            for (i, key) in keys.iter().enumerate() {
+                if i > 0 {
+                    out.push(',');
+                }
+                write_string(out, key);
+                out.push(':');
+                write_value(out, &map[key.as_str()])?;
+            }
+            out.push('}');
+        }
+    }
+    Ok(())
+}
+
+/// Lowercase hex SHA-256.
+pub fn sha256_hex(bytes: &[u8]) -> String {
+    let digest = Sha256::digest(bytes);
+    let mut out = String::with_capacity(64);
+    for byte in digest {
+        out.push_str(&format!("{byte:02x}"));
+    }
+    out
+}
+
+/// The canonical digest of a JSON value: SHA-256 over its canonical JSON bytes.
+pub fn digest_of(value: &Value) -> Result<String> {
+    canonical_bytes(value).map(|bytes| sha256_hex(&bytes))
+}
+
+/// Parses JSON strictly: duplicate keys at any depth, invalid UTF-8, non-finite numbers and
+/// trailing bytes are refused. The bus reader, reused so both layers agree byte for byte.
+pub fn parse_strict(bytes: &[u8]) -> Result<Value> {
+    flybus::wire::parse_json_strict(bytes).map_err(|e| wire_err(e.0))
+}
+
+/// Refuses a domain payload that does not fit the bus envelope ceiling.
+///
+/// The check is on canonical bytes, and the caller passes the overhead the surrounding
+/// envelope adds, so a payload that only fits without its envelope still fails.
+pub fn require_envelope_fit(value: &Value, envelope_overhead: usize) -> Result<usize> {
+    let len = canonicalize(value)?.len();
+    let total = len + envelope_overhead;
+    if total > MAX_ENVELOPE_BYTES {
+        return err(format!(
+            "envelope: {total} bytes exceeds the {MAX_ENVELOPE_BYTES}-byte maximum"
+        ));
+    }
+    Ok(total)
+}
+
+// ---------------------------------------------------------------------------------------------
+// Operation keys and canonical bodies
+
+/// The keys that belong to the bus, never to a domain body (ipc-v1 section 5: the canonical
+/// body "excludes changing bus callIds, deliveryIds and owner tokens").
+pub const BUS_ONLY_KEYS: &[&str] = &[
+    "callId",
+    "deliveryId",
+    "ownerId",
+    "ownerIds",
+    "deliveryIds",
+    "requestDeliveryId",
+    "expectedIncarnation",
+    "serviceIncarnation",
+    "connectionId",
+    "topicSequence",
+    "subscriptionId",
+];
+
+/// Fails if any bus-only key appears anywhere in `value`.
+pub fn reject_bus_identities(value: &Value) -> Result<()> {
+    match value {
+        Value::Object(map) => {
+            for (key, inner) in map {
+                if BUS_ONLY_KEYS.contains(&key.as_str()) {
+                    return err(format!(
+                        "canonical body: {key:?} is a bus identity and never part of a domain body"
+                    ));
+                }
+                reject_bus_identities(inner)?;
+            }
+            Ok(())
+        }
+        Value::Array(items) => {
+            for item in items {
+                reject_bus_identities(item)?;
+            }
+            Ok(())
+        }
+        _ => Ok(()),
+    }
+}
+
+/// `(sessionId, epoch, step, method, workerId)`: the operation key of a step mutation.
+///
+/// There is at most one Prepare, Commit or Advance for one key (ipc-v1 section 5). The key
+/// deliberately does not contain the requestId: a changed id for an existing key is CONFLICT,
+/// which can only be detected if the key is the same.
+#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct OperationKey {
+    pub scope: Scope,
+    pub method: String,
+    pub worker_id: String,
+}
+
+impl OperationKey {
+    pub fn new(scope: Scope, method: &str, worker_id: &str) -> Result<OperationKey> {
+        let key = OperationKey {
+            scope,
+            method: method.to_owned(),
+            worker_id: worker_id.to_owned(),
+        };
+        key.validate()?;
+        Ok(key)
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        use crate::scalar::DomainType;
+        self.scope.validate()?;
+        if !flybus::wire::is_method(&self.method) {
+            return err("OperationKey: method must be 1..=128 printable ASCII characters");
+        }
+        if !crate::scalar::is_id(&self.worker_id) {
+            return err("OperationKey: workerId is not a valid id");
+        }
+        Ok(())
+    }
+
+    pub fn to_json(&self) -> Value {
+        use crate::scalar::DomainType;
+        crate::scalar::obj(vec![
+            ("scope", self.scope.to_json()),
+            ("method", self.method.clone().into()),
+            ("workerId", self.worker_id.clone().into()),
+        ])
+    }
+
+    /// The canonical digest of the key, for a deduplication table that stores digests.
+    pub fn digest(&self) -> Result<String> {
+        digest_of(&self.to_json())
+    }
+}
+
+/// The canonical body of a domain operation: method, scope and validated params.
+///
+/// Two calls of the same operation key whose body digests differ are CONFLICT; two calls with
+/// the same digest are the same operation, whatever bus callId carried them.
+pub fn canonical_body(method: &str, scope: Option<&Scope>, params: &Value) -> Result<Value> {
+    if !flybus::wire::is_method(method) {
+        return err("canonical body: method must be 1..=128 printable ASCII characters");
+    }
+    if !params.is_object() {
+        return err("canonical body: params must be an object");
+    }
+    reject_bus_identities(params)?;
+    Ok(crate::scalar::obj(vec![
+        ("method", method.into()),
+        ("scope", Scope::nullable_to_json(scope)),
+        ("params", params.clone()),
+    ]))
+}
+
+/// The canonical body digest of a domain operation.
+pub fn body_digest(method: &str, scope: Option<&Scope>, params: &Value) -> Result<String> {
+    digest_of(&canonical_body(method, scope, params)?)
+}

services/flysim/crates/fly-session-types/src/checkpoint.rs

@@ -0,0 +1,368 @@
+//! `FLYSESS1`: the envelope layout of `docs/design/session-framework/checkpoint-envelope-v1.md`.
+//!
+//! This is the layout half of the specification, not the store: it lays out a header, a
+//! canonical-JSON manifest, a payload table and the payload bytes, and it reads one back.
+//! Writing generations, fsyncing and committing a manifest belong to the STATE-01 store slice.
+//! `FLYSIM01` is a different format with a different magic and is not touched by any of this.
+
+use serde_json::Value;
+use sha2::{Digest as _, Sha256};
+
+use crate::canonical;
+use crate::scalar::{Result, err, is_id};
+
+/// Envelope magic. Eight ASCII bytes, distinct from `FLYSIM01`.
+pub const MAGIC: &[u8; 8] = b"FLYSESS1";
+/// Footer magic, so a truncated file cannot look complete.
+pub const FOOTER_MAGIC: &[u8; 8] = b"FLYSESSF";
+/// Envelope version, in the header and in the manifest.
+pub const VERSION: u32 = 1;
+/// Fixed header size in bytes.
+pub const HEADER_BYTES: usize = 32;
+/// One payload table entry: a 64-byte name field, offset, length and a 32-byte digest.
+pub const TABLE_ENTRY_BYTES: usize = 112;
+/// Payload name field width.
+pub const NAME_BYTES: usize = 64;
+/// Footer size in bytes: total length, whole-prefix digest and the footer magic.
+pub const FOOTER_BYTES: usize = 48;
+/// Payloads start on an eight-byte boundary.
+pub const ALIGNMENT: u64 = 8;
+/// Payloads per envelope.
+pub const MAX_PAYLOADS: usize = 64;
+
+/// One payload's table entry.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PayloadEntry {
+    /// An `Id`: the new envelope widens the historical letters-only chunk name deliberately,
+    /// which is why it is a new version and not an extension of `FLYSIM01`.
+    pub name: String,
+    pub offset: u64,
+    pub byte_length: u64,
+    /// SHA-256 of exactly `byte_length` bytes at `offset`.
+    pub digest: [u8; 32],
+}
+
+/// A laid-out envelope: where everything is, before any bytes are written.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Layout {
+    pub manifest_offset: u64,
+    pub manifest_bytes: u32,
+    pub table_offset: u64,
+    pub entries: Vec<PayloadEntry>,
+    pub footer_offset: u64,
+    pub total_bytes: u64,
+}
+
+fn align_up(value: u64) -> u64 {
+    value.div_ceil(ALIGNMENT) * ALIGNMENT
+}
+
+/// Lays out the envelope for one manifest and a list of `(name, bytes)` payloads.
+pub fn layout(manifest: &Value, payloads: &[(String, Vec<u8>)]) -> Result<Layout> {
+    if payloads.len() > MAX_PAYLOADS {
+        return err("checkpoint envelope: at most 64 payloads");
+    }
+    crate::scalar::require_unique(
+        payloads.iter().map(|(name, _)| name.as_str()),
+        "checkpoint envelope: payload names",
+    )?;
+    for (name, _) in payloads {
+        if !is_id(name) {
+            return err(format!(
+                "checkpoint envelope: payload name {name:?} is not an Id"
+            ));
+        }
+    }
+    let manifest_text = canonical::canonicalize(manifest)?;
+    let manifest_bytes = u32::try_from(manifest_text.len())
+        .map_err(|_| crate::scalar::wire_err("checkpoint envelope: manifest is too large"))?;
+    let manifest_offset = HEADER_BYTES as u64;
+    let table_offset = align_up(manifest_offset + u64::from(manifest_bytes));
+    let mut offset = align_up(table_offset + (payloads.len() * TABLE_ENTRY_BYTES) as u64);
+    let mut entries = Vec::with_capacity(payloads.len());
+    for (name, bytes) in payloads {
+        entries.push(PayloadEntry {
+            name: name.clone(),
+            offset,
+            byte_length: bytes.len() as u64,
+            digest: Sha256::digest(bytes).into(),
+        });
+        offset = align_up(offset + bytes.len() as u64);
+    }
+    Ok(Layout {
+        manifest_offset,
+        manifest_bytes,
+        table_offset,
+        entries,
+        footer_offset: offset,
+        total_bytes: offset + FOOTER_BYTES as u64,
+    })
+}
+
+/// Writes one envelope: header, manifest, payload table, payloads, footer.
+pub fn encode(manifest: &Value, payloads: &[(String, Vec<u8>)]) -> Result<Vec<u8>> {
+    let layout = layout(manifest, payloads)?;
+    let manifest_text = canonical::canonicalize(manifest)?;
+    let mut out = vec![0u8; layout.footer_offset as usize];
+    out[0..8].copy_from_slice(MAGIC);
+    out[8..12].copy_from_slice(&VERSION.to_le_bytes());
+    out[12..16].copy_from_slice(&(HEADER_BYTES as u32).to_le_bytes());
+    out[16..20].copy_from_slice(&layout.manifest_bytes.to_le_bytes());
+    out[20..24].copy_from_slice(&(payloads.len() as u32).to_le_bytes());
+    out[24..28].copy_from_slice(&(layout.table_offset as u32).to_le_bytes());
+    out[28..32].copy_from_slice(&0u32.to_le_bytes());
+    let manifest_start = layout.manifest_offset as usize;
+    out[manifest_start..manifest_start + manifest_text.len()]
+        .copy_from_slice(manifest_text.as_bytes());
+    for (index, entry) in layout.entries.iter().enumerate() {
+        let base = layout.table_offset as usize + index * TABLE_ENTRY_BYTES;
+        out[base..base + entry.name.len()].copy_from_slice(entry.name.as_bytes());
+        let numbers = base + NAME_BYTES;
+        out[numbers..numbers + 8].copy_from_slice(&entry.offset.to_le_bytes());
+        out[numbers + 8..numbers + 16].copy_from_slice(&entry.byte_length.to_le_bytes());
+        out[numbers + 16..numbers + 48].copy_from_slice(&entry.digest);
+    }
+    for (entry, (_, bytes)) in layout.entries.iter().zip(payloads) {
+        let start = entry.offset as usize;
+        out[start..start + bytes.len()].copy_from_slice(bytes);
+    }
+    let digest: [u8; 32] = Sha256::digest(&out).into();
+    out.extend_from_slice(&layout.total_bytes.to_le_bytes());
+    out.extend_from_slice(&digest);
+    out.extend_from_slice(FOOTER_MAGIC);
+    Ok(out)
+}
+
+/// A decoded envelope.
+#[derive(Clone, Debug, PartialEq)]
+pub struct Envelope {
+    pub manifest: Value,
+    pub payloads: Vec<(String, Vec<u8>)>,
+    pub layout: Layout,
+}
+
+impl Envelope {
+    pub fn payload(&self, name: &str) -> Option<&[u8]> {
+        self.payloads
+            .iter()
+            .find(|(key, _)| key == name)
+            .map(|(_, bytes)| bytes.as_slice())
+    }
+}
+
+fn u32_at(bytes: &[u8], offset: usize) -> u32 {
+    u32::from_le_bytes([
+        bytes[offset],
+        bytes[offset + 1],
+        bytes[offset + 2],
+        bytes[offset + 3],
+    ])
+}
+
+fn u64_at(bytes: &[u8], offset: usize) -> u64 {
+    let mut buf = [0u8; 8];
+    buf.copy_from_slice(&bytes[offset..offset + 8]);
+    u64::from_le_bytes(buf)
+}
+
+/// Reads and fully validates one envelope: magic, version, footer digest, table ordering,
+/// alignment, bounds and every payload digest.
+pub fn decode(bytes: &[u8]) -> Result<Envelope> {
+    if bytes.len() < HEADER_BYTES + FOOTER_BYTES {
+        return err("checkpoint envelope: shorter than a header plus a footer");
+    }
+    if &bytes[0..8] != MAGIC {
+        return err("checkpoint envelope: wrong magic (FLYSIM01 is a different format)");
+    }
+    if u32_at(bytes, 8) != VERSION {
+        return err("checkpoint envelope: unsupported version");
+    }
+    if u32_at(bytes, 12) as usize != HEADER_BYTES {
+        return err("checkpoint envelope: headerBytes must be 32");
+    }
+    if u32_at(bytes, 28) != 0 {
+        return err("checkpoint envelope: reserved header word must be zero");
+    }
+    let manifest_bytes = u32_at(bytes, 16) as usize;
+    let payload_count = u32_at(bytes, 20) as usize;
+    let table_offset = u32_at(bytes, 24) as u64;
+    if payload_count > MAX_PAYLOADS {
+        return err("checkpoint envelope: at most 64 payloads");
+    }
+    let footer_offset = bytes.len() - FOOTER_BYTES;
+    if &bytes[footer_offset + 40..] != FOOTER_MAGIC {
+        return err("checkpoint envelope: missing footer magic");
+    }
+    if u64_at(bytes, footer_offset) != bytes.len() as u64 {
+        return err("checkpoint envelope: footer length does not match the file");
+    }
+    let recorded = &bytes[footer_offset + 8..footer_offset + 40];
+    let computed: [u8; 32] = Sha256::digest(&bytes[..footer_offset]).into();
+    if recorded != computed {
+        return err("checkpoint envelope: footer digest does not match the contents");
+    }
+    let manifest_start = HEADER_BYTES;
+    let manifest_end = manifest_start + manifest_bytes;
+    if manifest_end > footer_offset {
+        return err("checkpoint envelope: manifest runs past the payload area");
+    }
+    let manifest = canonical::parse_strict(&bytes[manifest_start..manifest_end])?;
+    let canonical_manifest = canonical::canonicalize(&manifest)?;
+    if canonical_manifest.as_bytes() != &bytes[manifest_start..manifest_end] {
+        return err("checkpoint envelope: the manifest is not canonical JSON");
+    }
+    if table_offset != align_up(manifest_end as u64) {
+        return err("checkpoint envelope: the payload table is not at its laid-out offset");
+    }
+    let table_end = table_offset as usize + payload_count * TABLE_ENTRY_BYTES;
+    if table_end > footer_offset {
+        return err("checkpoint envelope: the payload table runs past the payload area");
+    }
+    let mut entries = Vec::with_capacity(payload_count);
+    let mut payloads = Vec::with_capacity(payload_count);
+    let mut previous_end = align_up(table_end as u64);
+    for index in 0..payload_count {
+        let base = table_offset as usize + index * TABLE_ENTRY_BYTES;
+        let name_field = &bytes[base..base + NAME_BYTES];
+        let length = name_field
+            .iter()
+            .position(|b| *b == 0)
+            .unwrap_or(NAME_BYTES);
+        if name_field[length..].iter().any(|b| *b != 0) {
+            return err("checkpoint envelope: a payload name has bytes after its terminator");
+        }
+        let name = std::str::from_utf8(&name_field[..length])
+            .map_err(|_| crate::scalar::wire_err("checkpoint envelope: payload name is not UTF-8"))?
+            .to_owned();
+        if !is_id(&name) {
+            return err(format!(
+                "checkpoint envelope: payload name {name:?} is not an Id"
+            ));
+        }
+        let numbers = base + NAME_BYTES;
+        let offset = u64_at(bytes, numbers);
+        let byte_length = u64_at(bytes, numbers + 8);
+        let mut digest = [0u8; 32];
+        digest.copy_from_slice(&bytes[numbers + 16..numbers + 48]);
+        if offset != previous_end {
+            return err(format!(
+                "checkpoint envelope: payload {name:?} starts at {offset}, not at its aligned {previous_end}"
+            ));
+        }
+        let end = offset
+            .checked_add(byte_length)
+            .ok_or_else(|| crate::scalar::wire_err("checkpoint envelope: payload overflows"))?;
+        if end > footer_offset as u64 {
+            return err(format!(
+                "checkpoint envelope: payload {name:?} runs past the payload area"
+            ));
+        }
+        let payload = bytes[offset as usize..end as usize].to_vec();
+        let computed: [u8; 32] = Sha256::digest(&payload).into();
+        if computed != digest {
+            return err(format!(
+                "checkpoint envelope: payload {name:?} fails its digest"
+            ));
+        }
+        previous_end = align_up(end);
+        entries.push(PayloadEntry {
+            name: name.clone(),
+            offset,
+            byte_length,
+            digest,
+        });
+        payloads.push((name, payload));
+    }
+    crate::scalar::require_unique(
+        entries.iter().map(|e| e.name.as_str()),
+        "checkpoint envelope: payload names",
+    )?;
+    if previous_end != footer_offset as u64 {
+        return err("checkpoint envelope: padding between the last payload and the footer");
+    }
+    Ok(Envelope {
+        manifest,
+        layout: Layout {
+            manifest_offset: manifest_start as u64,
+            manifest_bytes: manifest_bytes as u32,
+            table_offset,
+            entries,
+            footer_offset: footer_offset as u64,
+            total_bytes: bytes.len() as u64,
+        },
+        payloads,
+    })
+}
+
+/// The manifest fields state-media-v1 section 4 requires, checked as a set: a manifest that
+/// omits one of them is not a complete checkpoint.
+pub const REQUIRED_MANIFEST_FIELDS: &[&str] = &[
+    "envelopeVersion",
+    "checkpointId",
+    "sourceScope",
+    "episodeId",
+    "worldTime",
+    "schedulerId",
+    "compositionDigest",
+    "portMap",
+    "compatibility",
+    "agents",
+    "coordinator",
+    "payloads",
+];
+
+/// Checks the manifest's required field set and that its payload table mirrors the envelope's.
+pub fn validate_manifest(envelope: &Envelope) -> Result<()> {
+    let map = envelope
+        .manifest
+        .as_object()
+        .ok_or_else(|| crate::scalar::wire_err("checkpoint manifest: must be an object"))?;
+    for field in REQUIRED_MANIFEST_FIELDS {
+        if !map.contains_key(*field) {
+            return err(format!("checkpoint manifest: missing {field:?}"));
+        }
+    }
+    if map.get("envelopeVersion").and_then(Value::as_u64) != Some(u64::from(VERSION)) {
+        return err("checkpoint manifest: envelopeVersion must be 1");
+    }
+    let listed = map
+        .get("payloads")
+        .and_then(Value::as_array)
+        .ok_or_else(|| crate::scalar::wire_err("checkpoint manifest: payloads must be an array"))?;
+    if listed.len() != envelope.layout.entries.len() {
+        return err("checkpoint manifest: payloads does not match the payload table");
+    }
+    for (declared, entry) in listed.iter().zip(&envelope.layout.entries) {
+        let name = declared.get("name").and_then(Value::as_str);
+        let length = declared
+            .get("byteLength")
+            .and_then(Value::as_str)
+            .and_then(crate::scalar::parse_u64);
+        let digest = declared.get("digest").and_then(Value::as_str);
+        if name != Some(entry.name.as_str()) {
+            return err("checkpoint manifest: payload name does not match the table");
+        }
+        if length != Some(entry.byte_length) {
+            return err(format!(
+                "checkpoint manifest: payload {:?} byteLength does not match the table",
+                entry.name
+            ));
+        }
+        if digest != Some(hex(&entry.digest).as_str()) {
+            return err(format!(
+                "checkpoint manifest: payload {:?} digest does not match the table",
+                entry.name
+            ));
+        }
+    }
+    Ok(())
+}
+
+/// Lowercase hex of a raw digest, the form the manifest records.
+pub fn hex(bytes: &[u8]) -> String {
+    let mut out = String::with_capacity(bytes.len() * 2);
+    for byte in bytes {
+        out.push_str(&format!("{byte:02x}"));
+    }
+    out
+}

services/flysim/crates/fly-session-types/src/fixtures.rs

@@ -0,0 +1,108 @@
+//! Loading the crate's `fixtures/` directory.
+//!
+//! The same files are read by the Rust tests and by `packages/session-types`, so a case only
+//! has to be written once to hold both languages to it.
+
+use std::path::{Path, PathBuf};
+
+use serde_json::Value;
+
+use crate::canonical;
+use crate::scalar::{Result, err, wire_err};
+
+/// The crate's `fixtures/` directory.
+pub fn dir() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR")).join("fixtures")
+}
+
+/// Reads one fixture file, parsed strictly.
+pub fn load(name: &str) -> Result<Value> {
+    let path = dir().join(name);
+    let bytes =
+        std::fs::read(&path).map_err(|e| wire_err(format!("fixture {}: {e}", path.display())))?;
+    canonical::parse_strict(&bytes)
+}
+
+/// Reads one fixture file as raw bytes, for the cases that are deliberately not valid JSON.
+pub fn load_bytes(name: &str) -> Result<Vec<u8>> {
+    let path = dir().join(name);
+    std::fs::read(&path).map_err(|e| wire_err(format!("fixture {}: {e}", path.display())))
+}
+
+/// The `cases` array of a fixture file.
+pub fn cases(file: &Value) -> Result<&Vec<Value>> {
+    match file.get("cases").and_then(Value::as_array) {
+        Some(cases) if !cases.is_empty() => Ok(cases),
+        _ => err("fixture: cases must be a nonempty array"),
+    }
+}
+
+/// A string field of one case.
+pub fn field<'a>(case: &'a Value, key: &str) -> Result<&'a str> {
+    case.get(key)
+        .and_then(Value::as_str)
+        .ok_or_else(|| wire_err(format!("fixture case: missing string field {key:?}")))
+}
+
+/// Decodes the `base64` field of a case that carries raw bytes.
+pub fn base64(case: &Value, key: &str) -> Result<Vec<u8>> {
+    decode_base64(field(case, key)?)
+}
+
+/// Standard base64 with padding. Small and local: the crate has no base64 dependency and the
+/// fixtures only carry a few hundred bytes.
+pub fn decode_base64(text: &str) -> Result<Vec<u8>> {
+    const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+    let bytes = text.as_bytes();
+    if !bytes.len().is_multiple_of(4) {
+        return err("base64: length must be a multiple of 4");
+    }
+    let mut out = Vec::with_capacity(bytes.len() / 4 * 3);
+    for quad in bytes.chunks_exact(4) {
+        let mut buffer = 0u32;
+        let mut keep = 3;
+        for (index, byte) in quad.iter().enumerate() {
+            let value = if *byte == b'=' {
+                if index < 2 {
+                    return err("base64: misplaced padding");
+                }
+                keep -= 1;
+                0
+            } else {
+                ALPHABET
+                    .iter()
+                    .position(|c| c == byte)
+                    .ok_or_else(|| wire_err("base64: invalid character"))? as u32
+            };
+            buffer = (buffer << 6) | value;
+        }
+        let triple = buffer.to_be_bytes();
+        out.extend_from_slice(&triple[1..1 + keep]);
+    }
+    Ok(out)
+}
+
+/// Standard base64 with padding, for generating fixtures.
+pub fn encode_base64(bytes: &[u8]) -> String {
+    const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
+    for chunk in bytes.chunks(3) {
+        let mut buffer = [0u8; 3];
+        buffer[..chunk.len()].copy_from_slice(chunk);
+        let value = u32::from_be_bytes([0, buffer[0], buffer[1], buffer[2]]);
+        let indexes = [
+            (value >> 18) & 0x3f,
+            (value >> 12) & 0x3f,
+            (value >> 6) & 0x3f,
+            value & 0x3f,
+        ];
+        for (position, index) in indexes.iter().enumerate() {
+            if position <= chunk.len() {
+                out.push(ALPHABET[*index as usize] as char);
+            } else {
+                out.push('=');
+            }
+        }
+    }
+    out
+}

services/flysim/crates/fly-session-types/src/lib.rs

@@ -0,0 +1,56 @@
+//! `fly-session-types`: the executable schemas of the session framework (CONTRACT-01).
+//!
+//! What this crate is:
+//!
+//! - the domain scalars of [ipc-v1] section 2 ([`scalar`]), reusing the bus's `Id`, `U64` and
+//!   `Digest` encodings rather than restating them;
+//!   the domain request/reply envelope and error codes of sections 3 and 7 ([`rpc`]);
+//! - the closed enums and method payloads of [workers-v1] ([`workers`]), the native media
+//!   and State.* payloads of [state-media-v1] ([`media`]), and the publication types of
+//!   [publishing-v1] ([`publishing`]);
+//! - canonical JSON (RFC 8785), canonical digests, the operation key and the canonical body
+//!   rules of ipc-v1 section 5 ([`canonical`]);
+//! - the documented canonical schema set and `contractDigest` ([`schema`]);
+//! - the trace format of [step-v1] section 8, with behaviour separated from operational
+//!   metadata and a comparator over behaviour alone ([`trace`]);
+//! - `seed-derivation-v1` ([`seed`]) and the `FLYSESS1` checkpoint envelope layout
+//!   ([`checkpoint`]), the two specifications CONTRACT-01 has to settle before the real-agent
+//!   and store slices.
+//!
+//! What it is not: a transport, a worker, a coordinator or a store. It holds no Game Boy FFI,
+//! no Melee parser and no console-specific state, and it never reaches the network.
+//!
+//! Every type implements [`scalar::DomainType`]: `from_json` reads and validates, `to_json`
+//! writes the canonical shape, and `validate` re-checks the rules that span fields. Reading
+//! refuses unknown fields, so a payload with a misspelled required field fails instead of
+//! silently defaulting.
+//!
+//! [ipc-v1]: ../../../../docs/design/session-framework/ipc-v1.md
+//! [workers-v1]: ../../../../docs/design/session-framework/workers-v1.md
+//! [state-media-v1]: ../../../../docs/design/session-framework/state-media-v1.md
+//! [publishing-v1]: ../../../../docs/design/session-framework/publishing-v1.md
+//! [step-v1]: ../../../../docs/design/session-framework/step-v1.md
+
+pub mod canonical;
+pub mod checkpoint;
+pub mod fixtures;
+pub mod media;
+pub mod publishing;
+pub mod rpc;
+pub mod scalar;
+pub mod schema;
+pub mod seed;
+pub mod trace;
+pub mod workers;
+
+pub use canonical::{OperationKey, body_digest, canonicalize, digest_of};
+pub use scalar::{
+    ArtifactIdentity, BusCallId, DomainRequestId, DomainType, OwnerKind, OwnerToken, RationalNs,
+    SchemaRef, Scope, TypedValue,
+};
+pub use schema::contract_digest;
+pub use trace::{TraceBehaviour, TraceOperational, TransitionTrace};
+
+/// The bus `ArtifactRef` these contracts reference. Re-exported so a consumer does not have
+/// to decide whether the domain has its own copy: it does not.
+pub use flybus::wire::ArtifactRef;

services/flysim/crates/fly-session-types/src/media.rs

@@ -0,0 +1,716 @@
+//! Native observation media (state-media-v1 section 2) and the State.* payloads (section 5).
+//!
+//! Descriptors carry the shape; refs carry one produced object. Both are validated against
+//! the descriptor, because a ref on its own cannot know its own row stride: use
+//! [`ViewRef::validate_against`] and [`AudioRef::validate_against`] wherever the descriptor
+//! is in hand.
+
+use flybus::wire::{ArtifactRef, Fields};
+
+use serde_json::Value;
+
+use crate::scalar::{
+    DomainType, RationalNs, Result, Scope, constant, err, finite_in, is_digest, is_id, list, obj,
+    require_unique, u64_json,
+};
+
+/// Max views per sensory input (workers-v1 section 1). The same bound applies to a
+/// descriptor's view list and to an observation's view lists: a descriptor that declared more
+/// views than one sensory input can carry could not be satisfied.
+pub const MAX_VIEWS: usize = 8;
+/// View dimensions are integers 1..=4096 (state-media-v1 section 2).
+pub const MAX_VIEW_DIMENSION: u64 = 4096;
+/// Pixel aspect numerator/denominator are positive integers <=65535.
+pub const MAX_PIXEL_ASPECT: u64 = 65_535;
+/// observationDelaySteps is an integer 0..=8.
+pub const MAX_OBSERVATION_DELAY_STEPS: u64 = 8;
+/// sampleFrames is 0..=192000 per chunk; sampleRate is 8000..=192000.
+pub const MAX_SAMPLE_FRAMES: u64 = 192_000;
+/// Audio streams per descriptor. Not a stated bound: chosen so an envelope cannot be filled
+/// with descriptors, and recorded in the schema set so it cannot drift silently.
+pub const MAX_AUDIO_STREAMS: usize = 8;
+
+/// `ViewDescriptor`: the fixed shape of one native view.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ViewDescriptor {
+    pub view_id: String,
+    pub width: u64,
+    pub height: u64,
+    pub row_stride: u64,
+    pub pixel_aspect_numerator: u64,
+    pub pixel_aspect_denominator: u64,
+    pub observation_delay_steps: u64,
+}
+
+impl ViewDescriptor {
+    /// The exact byte length of one frame of this view.
+    pub fn frame_bytes(&self) -> u64 {
+        self.row_stride * self.height
+    }
+
+    /// The producing boundary a required sensory view must have at `boundary`
+    /// (state-media-v1 section 2): `max(0, boundary - observationDelaySteps)`.
+    pub fn required_produced_step(&self, boundary: u64) -> u64 {
+        boundary.saturating_sub(self.observation_delay_steps)
+    }
+}
+
+impl DomainType for ViewDescriptor {
+    const TYPE_NAME: &'static str = "ViewDescriptor";
+
+    fn from_json(value: &Value) -> Result<ViewDescriptor> {
+        let mut f = Fields::new(value, "ViewDescriptor")?;
+        let view_id = f.id("viewId")?;
+        let width = f.int("width", 1, MAX_VIEW_DIMENSION)?;
+        let height = f.int("height", 1, MAX_VIEW_DIMENSION)?;
+        constant(&mut f, "format", "rgba8")?;
+        let row_stride = f.int("rowStride", 1, MAX_VIEW_DIMENSION * 4)?;
+        let aspect = f.value("pixelAspect")?;
+        let (pixel_aspect_numerator, pixel_aspect_denominator) = {
+            let mut a = Fields::new(aspect, "ViewDescriptor.pixelAspect")?;
+            let n = a.int("numerator", 1, MAX_PIXEL_ASPECT)?;
+            let d = a.int("denominator", 1, MAX_PIXEL_ASPECT)?;
+            a.finish()?;
+            (n, d)
+        };
+        let observation_delay_steps =
+            f.int("observationDelaySteps", 0, MAX_OBSERVATION_DELAY_STEPS)?;
+        f.finish()?;
+        let d = ViewDescriptor {
+            view_id,
+            width,
+            height,
+            row_stride,
+            pixel_aspect_numerator,
+            pixel_aspect_denominator,
+            observation_delay_steps,
+        };
+        d.validate()?;
+        Ok(d)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("viewId", self.view_id.clone().into()),
+            ("width", Value::from(self.width)),
+            ("height", Value::from(self.height)),
+            ("format", "rgba8".into()),
+            ("rowStride", Value::from(self.row_stride)),
+            (
+                "pixelAspect",
+                obj(vec![
+                    ("numerator", Value::from(self.pixel_aspect_numerator)),
+                    ("denominator", Value::from(self.pixel_aspect_denominator)),
+                ]),
+            ),
+            (
+                "observationDelaySteps",
+                Value::from(self.observation_delay_steps),
+            ),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.view_id) {
+            return err("ViewDescriptor: viewId is not a valid id");
+        }
+        if !(1..=MAX_VIEW_DIMENSION).contains(&self.width)
+            || !(1..=MAX_VIEW_DIMENSION).contains(&self.height)
+        {
+            return err("ViewDescriptor: width and height must be integers 1..=4096");
+        }
+        if self.row_stride != self.width * 4 {
+            return err(
+                "ViewDescriptor: rowStride must be exactly 4 x width (no padded rows in v1)",
+            );
+        }
+        if !(1..=MAX_PIXEL_ASPECT).contains(&self.pixel_aspect_numerator)
+            || !(1..=MAX_PIXEL_ASPECT).contains(&self.pixel_aspect_denominator)
+        {
+            return err("ViewDescriptor: pixelAspect parts must be positive integers <=65535");
+        }
+        if self.observation_delay_steps > MAX_OBSERVATION_DELAY_STEPS {
+            return err("ViewDescriptor: observationDelaySteps must be 0..=8");
+        }
+        Ok(())
+    }
+}
+
+/// `ViewRef`: one produced frame of one view.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ViewRef {
+    pub view_id: String,
+    pub produced_step: u64,
+    pub pixels: ArtifactRef,
+}
+
+impl ViewRef {
+    /// Byte shape and producing boundary against the descriptor that declared this view.
+    ///
+    /// `boundary` is the observation's boundary; a required sensory view must have been
+    /// produced at exactly `max(0, boundary - observationDelaySteps)`.
+    pub fn validate_against(
+        &self,
+        descriptor: &ViewDescriptor,
+        boundary: Option<u64>,
+    ) -> Result<()> {
+        if self.view_id != descriptor.view_id {
+            return err(format!(
+                "ViewRef: viewId {:?} does not match descriptor {:?}",
+                self.view_id, descriptor.view_id
+            ));
+        }
+        if self.pixels.byte_length != descriptor.frame_bytes() {
+            return err(format!(
+                "ViewRef {}: artifact is {} bytes, rowStride x height is {}",
+                self.view_id,
+                self.pixels.byte_length,
+                descriptor.frame_bytes()
+            ));
+        }
+        if let Some(boundary) = boundary {
+            let expected = descriptor.required_produced_step(boundary);
+            if self.produced_step != expected {
+                return err(format!(
+                    "ViewRef {}: producedStep {} must be max(0, {boundary} - {}) = {expected}",
+                    self.view_id, self.produced_step, descriptor.observation_delay_steps
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
+impl DomainType for ViewRef {
+    const TYPE_NAME: &'static str = "ViewRef";
+
+    fn from_json(value: &Value) -> Result<ViewRef> {
+        let mut f = Fields::new(value, "ViewRef")?;
+        let view_id = f.id("viewId")?;
+        let produced_step = f.u64_string("producedStep")?;
+        let pixels = ArtifactRef::from_json(f.value("pixels")?)?;
+        f.finish()?;
+        let r = ViewRef {
+            view_id,
+            produced_step,
+            pixels,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("viewId", self.view_id.clone().into()),
+            ("producedStep", u64_json(self.produced_step)),
+            ("pixels", self.pixels.to_json()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.view_id) {
+            return err("ViewRef: viewId is not a valid id");
+        }
+        if self.pixels.byte_length == 0 {
+            return err("ViewRef: pixels must have a positive byte length");
+        }
+        Ok(())
+    }
+}
+
+/// `AudioDescriptor`: one native audio stream's shape.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AudioDescriptor {
+    pub stream_id: String,
+    pub sample_rate: u64,
+    pub channels: u64,
+}
+
+impl AudioDescriptor {
+    /// The exact byte length of `frames` interleaved f32 frames.
+    pub fn chunk_bytes(&self, frames: u64) -> u64 {
+        frames * self.channels * 4
+    }
+}
+
+impl DomainType for AudioDescriptor {
+    const TYPE_NAME: &'static str = "AudioDescriptor";
+
+    fn from_json(value: &Value) -> Result<AudioDescriptor> {
+        let mut f = Fields::new(value, "AudioDescriptor")?;
+        let stream_id = f.id("streamId")?;
+        let sample_rate = f.int("sampleRate", 8_000, 192_000)?;
+        let channels = f.int("channels", 1, 8)?;
+        constant(&mut f, "format", "f32le-interleaved")?;
+        f.finish()?;
+        let d = AudioDescriptor {
+            stream_id,
+            sample_rate,
+            channels,
+        };
+        d.validate()?;
+        Ok(d)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("streamId", self.stream_id.clone().into()),
+            ("sampleRate", Value::from(self.sample_rate)),
+            ("channels", Value::from(self.channels)),
+            ("format", "f32le-interleaved".into()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.stream_id) {
+            return err("AudioDescriptor: streamId is not a valid id");
+        }
+        if !(8_000..=192_000).contains(&self.sample_rate) {
+            return err("AudioDescriptor: sampleRate must be an integer 8000..=192000");
+        }
+        if !(1..=8).contains(&self.channels) {
+            return err("AudioDescriptor: channels must be an integer 1..=8");
+        }
+        Ok(())
+    }
+}
+
+/// `AudioRef`: one produced chunk of one audio stream.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AudioRef {
+    pub stream_id: String,
+    pub first_sample: u64,
+    pub sample_frames: u64,
+    pub samples: ArtifactRef,
+    pub discontinuity: bool,
+}
+
+impl AudioRef {
+    /// Byte shape against the descriptor that declared this stream.
+    pub fn validate_against(&self, descriptor: &AudioDescriptor) -> Result<()> {
+        if self.stream_id != descriptor.stream_id {
+            return err(format!(
+                "AudioRef: streamId {:?} does not match descriptor {:?}",
+                self.stream_id, descriptor.stream_id
+            ));
+        }
+        let expected = descriptor.chunk_bytes(self.sample_frames);
+        if self.samples.byte_length != expected {
+            return err(format!(
+                "AudioRef {}: artifact is {} bytes, sampleFrames x channels x 4 is {expected}",
+                self.stream_id, self.samples.byte_length
+            ));
+        }
+        Ok(())
+    }
+
+    /// Within an epoch chunks cannot overlap or go backwards (state-media-v1 section 2).
+    pub fn follows(&self, previous: &AudioRef) -> Result<()> {
+        if self.stream_id != previous.stream_id {
+            return err("AudioRef: chunks of different streams are not ordered against each other");
+        }
+        let expected = previous.first_sample + previous.sample_frames;
+        if self.first_sample < expected {
+            return err(format!(
+                "AudioRef {}: firstSample {} overlaps the previous chunk, which ends at {expected}",
+                self.stream_id, self.first_sample
+            ));
+        }
+        Ok(())
+    }
+}
+
+impl DomainType for AudioRef {
+    const TYPE_NAME: &'static str = "AudioRef";
+
+    fn from_json(value: &Value) -> Result<AudioRef> {
+        let mut f = Fields::new(value, "AudioRef")?;
+        let stream_id = f.id("streamId")?;
+        let first_sample = f.u64_string("firstSample")?;
+        let sample_frames = f.int("sampleFrames", 0, MAX_SAMPLE_FRAMES)?;
+        let samples = ArtifactRef::from_json(f.value("samples")?)?;
+        let discontinuity = f.boolean("discontinuity")?;
+        f.finish()?;
+        let r = AudioRef {
+            stream_id,
+            first_sample,
+            sample_frames,
+            samples,
+            discontinuity,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("streamId", self.stream_id.clone().into()),
+            ("firstSample", u64_json(self.first_sample)),
+            ("sampleFrames", Value::from(self.sample_frames)),
+            ("samples", self.samples.to_json()),
+            ("discontinuity", Value::Bool(self.discontinuity)),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.stream_id) {
+            return err("AudioRef: streamId is not a valid id");
+        }
+        if self.sample_frames > MAX_SAMPLE_FRAMES {
+            return err("AudioRef: sampleFrames must be an integer 0..=192000");
+        }
+        if self.first_sample.checked_add(self.sample_frames).is_none() {
+            return err("AudioRef: firstSample + sampleFrames overflows U64");
+        }
+        Ok(())
+    }
+}
+
+/// Reads a bounded, unique-by-`viewId` list of view refs.
+pub fn view_list(f: &mut Fields<'_>, key: &'static str) -> Result<Vec<ViewRef>> {
+    let views = list(f, key, 0, MAX_VIEWS, ViewRef::from_json)?;
+    require_unique(views.iter().map(|v| v.view_id.as_str()), key)?;
+    Ok(views)
+}
+
+/// Reads a bounded, unique-by-`streamId` list of audio refs.
+pub fn audio_list(f: &mut Fields<'_>, key: &'static str) -> Result<Vec<AudioRef>> {
+    let audio = list(f, key, 0, MAX_AUDIO_STREAMS, AudioRef::from_json)?;
+    require_unique(audio.iter().map(|a| a.stream_id.as_str()), key)?;
+    Ok(audio)
+}
+
+// ---------------------------------------------------------------------------------------------
+// State.* payloads (state-media-v1 section 5)
+
+/// A checkpoint payload artifact: the digest is mandatory on checkpoint payloads
+/// (state-media-v1 section 1).
+fn checkpoint_payload(f: &mut Fields<'_>, key: &'static str) -> Result<ArtifactRef> {
+    let reference = ArtifactRef::from_json(f.value(key)?)?;
+    match &reference.digest {
+        Some(d) if is_digest(d) => Ok(reference),
+        _ => err(format!(
+            "{key}: a checkpoint payload must carry a content digest"
+        )),
+    }
+}
+
+/// `State.Capture` params.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CaptureParams {
+    pub checkpoint_id: String,
+}
+
+impl DomainType for CaptureParams {
+    const TYPE_NAME: &'static str = "CaptureParams";
+
+    fn from_json(value: &Value) -> Result<CaptureParams> {
+        let mut f = Fields::new(value, "CaptureParams")?;
+        let checkpoint_id = f.id("checkpointId")?;
+        f.finish()?;
+        let p = CaptureParams { checkpoint_id };
+        p.validate()?;
+        Ok(p)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![("checkpointId", self.checkpoint_id.clone().into())])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.checkpoint_id) {
+            return err("CaptureParams: checkpointId is not a valid id");
+        }
+        Ok(())
+    }
+}
+
+/// `State.Capture` result.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CaptureResult {
+    pub checkpoint_id: String,
+    pub boundary: u64,
+    pub compatibility_digest: String,
+    pub payload: ArtifactRef,
+}
+
+impl DomainType for CaptureResult {
+    const TYPE_NAME: &'static str = "CaptureResult";
+
+    fn from_json(value: &Value) -> Result<CaptureResult> {
+        let mut f = Fields::new(value, "CaptureResult")?;
+        let checkpoint_id = f.id("checkpointId")?;
+        let boundary = f.u64_string("boundary")?;
+        let compatibility_digest = f.string("compatibilityDigest")?.to_owned();
+        let payload = checkpoint_payload(&mut f, "payload")?;
+        f.finish()?;
+        let r = CaptureResult {
+            checkpoint_id,
+            boundary,
+            compatibility_digest,
+            payload,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("checkpointId", self.checkpoint_id.clone().into()),
+            ("boundary", u64_json(self.boundary)),
+            (
+                "compatibilityDigest",
+                self.compatibility_digest.clone().into(),
+            ),
+            ("payload", self.payload.to_json()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.checkpoint_id) {
+            return err("CaptureResult: checkpointId is not a valid id");
+        }
+        if !is_digest(&self.compatibility_digest) {
+            return err("CaptureResult: compatibilityDigest must be 64 lowercase hex digits");
+        }
+        match &self.payload.digest {
+            Some(d) if is_digest(d) => Ok(()),
+            _ => err("CaptureResult: a checkpoint payload must carry a content digest"),
+        }
+    }
+}
+
+/// `State.StageRestore` params. The scope is the source boundary, under a proposed new epoch.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct StageRestoreParams {
+    pub checkpoint_id: String,
+    pub source_scope: Scope,
+    pub compatibility_digest: String,
+    pub payload: ArtifactRef,
+}
+
+impl DomainType for StageRestoreParams {
+    const TYPE_NAME: &'static str = "StageRestoreParams";
+
+    fn from_json(value: &Value) -> Result<StageRestoreParams> {
+        let mut f = Fields::new(value, "StageRestoreParams")?;
+        let checkpoint_id = f.id("checkpointId")?;
+        let source_scope = Scope::from_json(f.value("sourceScope")?)?;
+        let compatibility_digest = f.string("compatibilityDigest")?.to_owned();
+        let payload = checkpoint_payload(&mut f, "payload")?;
+        f.finish()?;
+        let p = StageRestoreParams {
+            checkpoint_id,
+            source_scope,
+            compatibility_digest,
+            payload,
+        };
+        p.validate()?;
+        Ok(p)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("checkpointId", self.checkpoint_id.clone().into()),
+            ("sourceScope", self.source_scope.to_json()),
+            (
+                "compatibilityDigest",
+                self.compatibility_digest.clone().into(),
+            ),
+            ("payload", self.payload.to_json()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.checkpoint_id) {
+            return err("StageRestoreParams: checkpointId is not a valid id");
+        }
+        self.source_scope.validate()?;
+        if !is_digest(&self.compatibility_digest) {
+            return err("StageRestoreParams: compatibilityDigest must be 64 lowercase hex digits");
+        }
+        match &self.payload.digest {
+            Some(d) if is_digest(d) => Ok(()),
+            _ => err("StageRestoreParams: a checkpoint payload must carry a content digest"),
+        }
+    }
+}
+
+/// `State.StageRestore` result.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct StageRestoreResult {
+    pub checkpoint_id: String,
+    pub restore_token: String,
+}
+
+impl DomainType for StageRestoreResult {
+    const TYPE_NAME: &'static str = "StageRestoreResult";
+
+    fn from_json(value: &Value) -> Result<StageRestoreResult> {
+        let mut f = Fields::new(value, "StageRestoreResult")?;
+        let checkpoint_id = f.id("checkpointId")?;
+        let restore_token = f.id("restoreToken")?;
+        f.finish()?;
+        let r = StageRestoreResult {
+            checkpoint_id,
+            restore_token,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("checkpointId", self.checkpoint_id.clone().into()),
+            ("restoreToken", self.restore_token.clone().into()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.checkpoint_id) || !is_id(&self.restore_token) {
+            return err("StageRestoreResult: checkpointId and restoreToken must be valid ids");
+        }
+        Ok(())
+    }
+}
+
+/// `State.ActivateRestore` params.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ActivateRestoreParams {
+    pub restore_token: String,
+}
+
+impl DomainType for ActivateRestoreParams {
+    const TYPE_NAME: &'static str = "ActivateRestoreParams";
+
+    fn from_json(value: &Value) -> Result<ActivateRestoreParams> {
+        let mut f = Fields::new(value, "ActivateRestoreParams")?;
+        let restore_token = f.id("restoreToken")?;
+        f.finish()?;
+        let p = ActivateRestoreParams { restore_token };
+        p.validate()?;
+        Ok(p)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![("restoreToken", self.restore_token.clone().into())])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.restore_token) {
+            return err("ActivateRestoreParams: restoreToken is not a valid id");
+        }
+        Ok(())
+    }
+}
+
+/// `State.ActivateRestore` result. The observation is required from an environment and null
+/// from an agent (state-media-v1 section 5); which one applies is the caller's role, so the
+/// role-specific check is [`ActivateRestoreResult::validate_for_role`].
+#[derive(Clone, Debug, PartialEq)]
+pub struct ActivateRestoreResult {
+    pub committed_step: u64,
+    pub checkpoint_id: String,
+    pub observation: Option<crate::workers::WorldObservation>,
+}
+
+impl ActivateRestoreResult {
+    pub fn validate_for_role(&self, role: crate::workers::Role) -> Result<()> {
+        self.validate()?;
+        match (role, &self.observation) {
+            (crate::workers::Role::Environment, None) => {
+                err("ActivateRestoreResult: an environment must return its restored observation")
+            }
+            (crate::workers::Role::Agent, Some(_)) => {
+                err("ActivateRestoreResult: an agent returns a null observation")
+            }
+            _ => Ok(()),
+        }
+    }
+}
+
+impl DomainType for ActivateRestoreResult {
+    const TYPE_NAME: &'static str = "ActivateRestoreResult";
+
+    fn from_json(value: &Value) -> Result<ActivateRestoreResult> {
+        let mut f = Fields::new(value, "ActivateRestoreResult")?;
+        let committed_step = f.u64_string("committedStep")?;
+        let checkpoint_id = f.id("checkpointId")?;
+        let observation = match f.value("observation")? {
+            Value::Null => None,
+            v => Some(crate::workers::WorldObservation::from_json(v)?),
+        };
+        f.finish()?;
+        let r = ActivateRestoreResult {
+            committed_step,
+            checkpoint_id,
+            observation,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("committedStep", u64_json(self.committed_step)),
+            ("checkpointId", self.checkpoint_id.clone().into()),
+            (
+                "observation",
+                self.observation
+                    .as_ref()
+                    .map_or(Value::Null, |o| o.to_json()),
+            ),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.checkpoint_id) {
+            return err("ActivateRestoreResult: checkpointId is not a valid id");
+        }
+        if let Some(observation) = &self.observation {
+            observation.validate()?;
+            if observation.boundary != self.committed_step {
+                return err(
+                    "ActivateRestoreResult: the observation boundary must be the committed step",
+                );
+            }
+        }
+        Ok(())
+    }
+}
+
+/// The pixel aspect of a view as a rational, for presentation.
+pub fn pixel_aspect(descriptor: &ViewDescriptor) -> Result<RationalNs> {
+    RationalNs::reduced(
+        u128::from(descriptor.pixel_aspect_numerator),
+        u128::from(descriptor.pixel_aspect_denominator),
+    )
+}
+
+/// Audio presentation timestamp, `firstSample / sampleRate` seconds, as a checked rational.
+pub fn audio_pts(reference: &AudioRef, descriptor: &AudioDescriptor) -> Result<RationalNs> {
+    RationalNs::reduced(
+        u128::from(reference.first_sample),
+        u128::from(descriptor.sample_rate),
+    )
+}
+
+/// Samples must be finite f32 (state-media-v1 section 2). The bytes live in an artifact, so
+/// this is the check a reader runs over a mapped chunk.
+pub fn require_finite_samples(bytes: &[u8]) -> Result<()> {
+    if !bytes.len().is_multiple_of(4) {
+        return err("audio chunk: length must be a multiple of 4");
+    }
+    for (index, chunk) in bytes.chunks_exact(4).enumerate() {
+        let sample = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
+        if !sample.is_finite() {
+            return err(format!("audio chunk: sample {index} is not finite"));
+        }
+    }
+    Ok(())
+}
+
+/// A unit-range helper for presentation code that needs the neutral-in-range rule.
+pub fn require_unit(f: &mut Fields<'_>, key: &'static str) -> Result<f64> {
+    finite_in(f, key, 0.0, 1.0)
+}

services/flysim/crates/fly-session-types/src/publishing.rs

@@ -0,0 +1,450 @@
+//! The publication types of publishing-v1 section 3.
+//!
+//! A descriptor changes rarely and a snapshot changes every boundary; both are published on
+//! the same bus, and a snapshot names the descriptor revision it was shaped by.
+
+use flybus::wire::Fields;
+use serde_json::Value;
+
+use crate::media::{AudioRef, MAX_VIEWS, ViewRef, audio_list, view_list};
+use crate::scalar::{
+    DomainType, RationalNs, Result, SchemaRef, Scope, TypedValue, constant, err, id_list,
+    is_digest, is_id, list, obj, require_unique, u64_json,
+};
+use crate::workers::{
+    AgentTelemetry, AssetRef, EnvironmentDescriptor, MAX_AGENTS, MAX_RATE_ROLES, PortControl,
+};
+
+/// Declared stimulus kinds per agent. Not a stated bound; recorded in the schema set.
+pub const MAX_SUPPORTED_STIMULI: usize = 64;
+/// Installed assets in one descriptor. Not a stated bound; recorded in the schema set.
+pub const MAX_ASSETS: usize = 64;
+/// Scoped event ids in one snapshot. Not a stated bound; recorded in the schema set.
+pub const MAX_SNAPSHOT_EVENTS: usize = 64;
+
+/// One agent's place in the composition.
+#[derive(Clone, Debug, PartialEq)]
+pub struct AgentDescriptor {
+    pub agent_id: String,
+    pub port_id: String,
+    pub profile_digest: String,
+    pub dataset_digest: String,
+    pub index_digest: String,
+    pub neuron_count: u64,
+    pub rate_roles: Vec<String>,
+    pub supported_stimuli: Vec<String>,
+}
+
+/// `SessionDescriptor`: the framework shape of one running session.
+#[derive(Clone, Debug, PartialEq)]
+pub struct SessionDescriptor {
+    pub session_id: String,
+    pub revision: u64,
+    pub composition_digest: String,
+    pub environment: EnvironmentDescriptor,
+    pub task_schema: SchemaRef,
+    pub agents: Vec<AgentDescriptor>,
+    pub assets: Vec<AssetRef>,
+}
+
+impl DomainType for SessionDescriptor {
+    const TYPE_NAME: &'static str = "SessionDescriptor";
+
+    fn from_json(value: &Value) -> Result<SessionDescriptor> {
+        let mut f = Fields::new(value, "SessionDescriptor")?;
+        let session_id = f.id("sessionId")?;
+        let revision = f.u64_string("revision")?;
+        let composition_digest = f.string("compositionDigest")?.to_owned();
+        constant(&mut f, "schedulerId", "lockstep-v1")?;
+        let environment = EnvironmentDescriptor::from_json(f.value("environment")?)?;
+        let task_schema = SchemaRef::from_json(f.value("taskSchema")?)?;
+        let agents = list(&mut f, "agents", 1, MAX_AGENTS, |v| {
+            let mut a = Fields::new(v, "SessionDescriptor.agents")?;
+            let agent_id = a.id("agentId")?;
+            let port_id = a.id("portId")?;
+            let profile_digest = a.string("profileDigest")?.to_owned();
+            let dataset_digest = a.string("datasetDigest")?.to_owned();
+            let index_digest = a.string("indexDigest")?.to_owned();
+            let neuron_count = a.u64_string("neuronCount")?;
+            let rate_roles = id_list(&mut a, "rateRoles", 0, MAX_RATE_ROLES)?;
+            let supported_stimuli = id_list(&mut a, "supportedStimuli", 0, MAX_SUPPORTED_STIMULI)?;
+            a.finish()?;
+            Ok(AgentDescriptor {
+                agent_id,
+                port_id,
+                profile_digest,
+                dataset_digest,
+                index_digest,
+                neuron_count,
+                rate_roles,
+                supported_stimuli,
+            })
+        })?;
+        let assets = list(&mut f, "assets", 0, MAX_ASSETS, AssetRef::from_json)?;
+        f.finish()?;
+        let d = SessionDescriptor {
+            session_id,
+            revision,
+            composition_digest,
+            environment,
+            task_schema,
+            agents,
+            assets,
+        };
+        d.validate()?;
+        Ok(d)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("sessionId", self.session_id.clone().into()),
+            ("revision", u64_json(self.revision)),
+            ("compositionDigest", self.composition_digest.clone().into()),
+            ("schedulerId", "lockstep-v1".into()),
+            ("environment", self.environment.to_json()),
+            ("taskSchema", self.task_schema.to_json()),
+            (
+                "agents",
+                Value::Array(
+                    self.agents
+                        .iter()
+                        .map(|a| {
+                            obj(vec![
+                                ("agentId", a.agent_id.clone().into()),
+                                ("portId", a.port_id.clone().into()),
+                                ("profileDigest", a.profile_digest.clone().into()),
+                                ("datasetDigest", a.dataset_digest.clone().into()),
+                                ("indexDigest", a.index_digest.clone().into()),
+                                ("neuronCount", u64_json(a.neuron_count)),
+                                (
+                                    "rateRoles",
+                                    Value::Array(
+                                        a.rate_roles.iter().map(|r| r.clone().into()).collect(),
+                                    ),
+                                ),
+                                (
+                                    "supportedStimuli",
+                                    Value::Array(
+                                        a.supported_stimuli
+                                            .iter()
+                                            .map(|s| s.clone().into())
+                                            .collect(),
+                                    ),
+                                ),
+                            ])
+                        })
+                        .collect(),
+                ),
+            ),
+            (
+                "assets",
+                Value::Array(self.assets.iter().map(AssetRef::to_json).collect()),
+            ),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.session_id) {
+            return err("SessionDescriptor: sessionId is not a valid id");
+        }
+        if !is_digest(&self.composition_digest) {
+            return err("SessionDescriptor: compositionDigest must be 64 lowercase hex digits");
+        }
+        self.environment.validate()?;
+        self.task_schema.validate()?;
+        if self.agents.is_empty() || self.agents.len() > MAX_AGENTS {
+            return err("SessionDescriptor: 1..=4 agents in the first composition");
+        }
+        require_unique(
+            self.agents.iter().map(|a| a.agent_id.as_str()),
+            "SessionDescriptor.agents agentId",
+        )?;
+        require_unique(
+            self.agents.iter().map(|a| a.port_id.as_str()),
+            "SessionDescriptor.agents portId",
+        )?;
+        for agent in &self.agents {
+            if !is_id(&agent.agent_id) || !is_id(&agent.port_id) {
+                return err("SessionDescriptor: agentId and portId must be valid ids");
+            }
+            for (what, digest) in [
+                ("profileDigest", &agent.profile_digest),
+                ("datasetDigest", &agent.dataset_digest),
+                ("indexDigest", &agent.index_digest),
+            ] {
+                if !is_digest(digest) {
+                    return err(format!(
+                        "SessionDescriptor: agent {what} must be 64 lowercase hex digits"
+                    ));
+                }
+            }
+            if agent.rate_roles.len() > MAX_RATE_ROLES {
+                return err("SessionDescriptor: at most 64 rate roles per agent");
+            }
+            require_unique(
+                agent.rate_roles.iter().map(String::as_str),
+                "SessionDescriptor.agents rateRoles",
+            )?;
+            require_unique(
+                agent.supported_stimuli.iter().map(String::as_str),
+                "SessionDescriptor.agents supportedStimuli",
+            )?;
+            if self.environment.port(&agent.port_id).is_none() {
+                return err(format!(
+                    "SessionDescriptor: agent {:?} is bound to port {:?}, which the environment does not declare",
+                    agent.agent_id, agent.port_id
+                ));
+            }
+        }
+        require_unique(
+            self.assets.iter().map(|a| a.id.as_str()),
+            "SessionDescriptor.assets",
+        )?;
+        for asset in &self.assets {
+            asset.validate()?;
+        }
+        Ok(())
+    }
+}
+
+/// One agent's committed values in a snapshot.
+#[derive(Clone, Debug, PartialEq)]
+pub struct SnapshotAgent {
+    pub agent_id: String,
+    pub telemetry: AgentTelemetry,
+    pub selected_decision: Option<TypedValue>,
+    pub applied_controls: Option<PortControl>,
+}
+
+/// `CommittedSnapshot`: the values of one committed boundary.
+#[derive(Clone, Debug, PartialEq)]
+pub struct CommittedSnapshot {
+    pub descriptor_revision: u64,
+    pub publisher_incarnation: String,
+    pub scope: Scope,
+    pub episode_id: String,
+    pub sequence: u64,
+    pub world_time: RationalNs,
+    pub agents: Vec<SnapshotAgent>,
+    pub progress: TypedValue,
+    pub views: Vec<ViewRef>,
+    pub audio: Vec<AudioRef>,
+    pub event_ids: Vec<String>,
+}
+
+impl CommittedSnapshot {
+    /// Descriptor agreement: the revision, the agent set and the port each control names.
+    pub fn validate_against(&self, descriptor: &SessionDescriptor) -> Result<()> {
+        self.validate()?;
+        if self.descriptor_revision != descriptor.revision {
+            return err("CommittedSnapshot: descriptorRevision does not match the descriptor");
+        }
+        if self.scope.session_id != descriptor.session_id {
+            return err("CommittedSnapshot: sessionId does not match the descriptor");
+        }
+        for agent in &self.agents {
+            let declared = descriptor
+                .agents
+                .iter()
+                .find(|a| a.agent_id == agent.agent_id)
+                .ok_or_else(|| {
+                    crate::scalar::wire_err(format!(
+                        "CommittedSnapshot: agent {:?} is not in the descriptor",
+                        agent.agent_id
+                    ))
+                })?;
+            agent
+                .telemetry
+                .validate_against_roles(&declared.rate_roles)?;
+            if let Some(controls) = &agent.applied_controls {
+                if controls.port_id != declared.port_id {
+                    return err(format!(
+                        "CommittedSnapshot: agent {:?} controls port {:?}, not its assigned {:?}",
+                        agent.agent_id, controls.port_id, declared.port_id
+                    ));
+                }
+                let port = descriptor
+                    .environment
+                    .port(&declared.port_id)
+                    .ok_or_else(|| {
+                        crate::scalar::wire_err("CommittedSnapshot: assigned port is not declared")
+                    })?;
+                controls.validate_against(&port.controls)?;
+            }
+        }
+        Ok(())
+    }
+}
+
+impl DomainType for CommittedSnapshot {
+    const TYPE_NAME: &'static str = "CommittedSnapshot";
+
+    fn from_json(value: &Value) -> Result<CommittedSnapshot> {
+        let mut f = Fields::new(value, "CommittedSnapshot")?;
+        let descriptor_revision = f.u64_string("descriptorRevision")?;
+        let publisher_incarnation = f.id("publisherIncarnation")?;
+        let scope = Scope::from_json(f.value("scope")?)?;
+        let episode_id = f.id("episodeId")?;
+        let sequence = f.u64_string("sequence")?;
+        let world_time = RationalNs::from_json(f.value("worldTime")?)?;
+        let agents = list(&mut f, "agents", 1, MAX_AGENTS, |v| {
+            let mut a = Fields::new(v, "CommittedSnapshot.agents")?;
+            let agent_id = a.id("agentId")?;
+            let telemetry = AgentTelemetry::from_json(a.value("telemetry")?)?;
+            let selected_decision = TypedValue::nullable_from_json(a.value("selectedDecision")?)?;
+            let applied_controls = match a.value("appliedControls")? {
+                Value::Null => None,
+                v => Some(PortControl::from_json(v)?),
+            };
+            a.finish()?;
+            Ok(SnapshotAgent {
+                agent_id,
+                telemetry,
+                selected_decision,
+                applied_controls,
+            })
+        })?;
+        let progress = TypedValue::from_json(f.value("progress")?)?;
+        let (views, audio) = {
+            let v = f.value("media")?;
+            let mut m = Fields::new(v, "CommittedSnapshot.media")?;
+            let views = view_list(&mut m, "views")?;
+            let audio = audio_list(&mut m, "audio")?;
+            m.finish()?;
+            (views, audio)
+        };
+        let event_ids = id_list(&mut f, "eventIds", 0, MAX_SNAPSHOT_EVENTS)?;
+        f.finish()?;
+        let s = CommittedSnapshot {
+            descriptor_revision,
+            publisher_incarnation,
+            scope,
+            episode_id,
+            sequence,
+            world_time,
+            agents,
+            progress,
+            views,
+            audio,
+            event_ids,
+        };
+        s.validate()?;
+        Ok(s)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("descriptorRevision", u64_json(self.descriptor_revision)),
+            (
+                "publisherIncarnation",
+                self.publisher_incarnation.clone().into(),
+            ),
+            ("scope", self.scope.to_json()),
+            ("episodeId", self.episode_id.clone().into()),
+            ("sequence", u64_json(self.sequence)),
+            ("worldTime", self.world_time.to_json()),
+            (
+                "agents",
+                Value::Array(
+                    self.agents
+                        .iter()
+                        .map(|a| {
+                            obj(vec![
+                                ("agentId", a.agent_id.clone().into()),
+                                ("telemetry", a.telemetry.to_json()),
+                                (
+                                    "selectedDecision",
+                                    TypedValue::nullable_to_json(a.selected_decision.as_ref()),
+                                ),
+                                (
+                                    "appliedControls",
+                                    a.applied_controls
+                                        .as_ref()
+                                        .map_or(Value::Null, PortControl::to_json),
+                                ),
+                            ])
+                        })
+                        .collect(),
+                ),
+            ),
+            ("progress", self.progress.to_json()),
+            (
+                "media",
+                obj(vec![
+                    (
+                        "views",
+                        Value::Array(self.views.iter().map(ViewRef::to_json).collect()),
+                    ),
+                    (
+                        "audio",
+                        Value::Array(self.audio.iter().map(AudioRef::to_json).collect()),
+                    ),
+                ]),
+            ),
+            (
+                "eventIds",
+                Value::Array(self.event_ids.iter().map(|e| e.clone().into()).collect()),
+            ),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.publisher_incarnation) || !is_id(&self.episode_id) {
+            return err("CommittedSnapshot: publisherIncarnation and episodeId must be valid ids");
+        }
+        self.scope.validate()?;
+        self.world_time.validate()?;
+        if self.agents.is_empty() || self.agents.len() > MAX_AGENTS {
+            return err("CommittedSnapshot: 1..=4 agents");
+        }
+        require_unique(
+            self.agents.iter().map(|a| a.agent_id.as_str()),
+            "CommittedSnapshot.agents",
+        )?;
+        for agent in &self.agents {
+            if !is_id(&agent.agent_id) {
+                return err("CommittedSnapshot: agentId is not a valid id");
+            }
+            agent.telemetry.validate()?;
+            if let Some(decision) = &agent.selected_decision {
+                decision.validate()?;
+            }
+            if let Some(controls) = &agent.applied_controls {
+                controls.validate()?;
+            }
+            // "Decisions/controls describe the transition ending at that boundary, null at
+            // initial boundary 0." (publishing-v1 section 3)
+            if self.scope.step == 0
+                && (agent.selected_decision.is_some() || agent.applied_controls.is_some())
+            {
+                return err(
+                    "CommittedSnapshot: at boundary 0 selectedDecision and appliedControls are null",
+                );
+            }
+            if self.scope.step > 0
+                && (agent.selected_decision.is_none() || agent.applied_controls.is_none())
+            {
+                return err(
+                    "CommittedSnapshot: past boundary 0 every agent has a decision and applied controls",
+                );
+            }
+        }
+        self.progress.validate()?;
+        if self.views.len() > MAX_VIEWS {
+            return err("CommittedSnapshot: at most 8 views");
+        }
+        require_unique(
+            self.views.iter().map(|v| v.view_id.as_str()),
+            "CommittedSnapshot.media.views",
+        )?;
+        require_unique(
+            self.audio.iter().map(|a| a.stream_id.as_str()),
+            "CommittedSnapshot.media.audio",
+        )?;
+        require_unique(
+            self.event_ids.iter().map(String::as_str),
+            "CommittedSnapshot.eventIds",
+        )?;
+        Ok(())
+    }
+}

services/flysim/crates/fly-session-types/src/rpc.rs

@@ -0,0 +1,420 @@
+//! The domain request/reply envelope of ipc-v1 section 3 and the error codes of section 7.
+//!
+//! A domain reply is the `outcome` object inside a bus `rpc.result`. Bus route or admission
+//! failure is not one of these: it never reaches a handler, so it cannot carry a mutation
+//! certainty.
+
+use flybus::wire::Fields;
+use serde_json::Value;
+
+use crate::canonical;
+use crate::scalar::{
+    DomainRequestId, DomainType, Result, Scope, bounded_string, constant, enumeration, err, is_id,
+    obj,
+};
+use crate::workers::MAX_MESSAGE_CODE_POINTS;
+
+/// The domain error codes of ipc-v1 section 7.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum ErrorCode {
+    /// Invalid schema/range, before mutation.
+    InvalidArgument,
+    /// Missing method or capability.
+    Unsupported,
+    /// Wrong session/profile/port/build/asset identity.
+    IdentityMismatch,
+    StaleEpoch,
+    StaleStep,
+    FutureStep,
+    /// Wrong worker phase.
+    InvalidPhase,
+    /// Existing logical operation with a changed id or body.
+    Conflict,
+    /// The original operation is still executing; this duplicate bus call started no work.
+    InProgress,
+    /// Domain capacity unavailable before admission.
+    Busy,
+    /// Missing, unowned or mismatched artifact, or an invalid media shape.
+    BufferInvalid,
+    /// Safe replay is no longer available; never recompute to replace it.
+    ResultExpired,
+    /// Restore validation failed before activation.
+    IncompatibleState,
+    BackendFailure,
+    Internal,
+}
+
+impl ErrorCode {
+    pub const ALL: &'static [&'static str] = &[
+        "INVALID_ARGUMENT",
+        "UNSUPPORTED",
+        "IDENTITY_MISMATCH",
+        "STALE_EPOCH",
+        "STALE_STEP",
+        "FUTURE_STEP",
+        "INVALID_PHASE",
+        "CONFLICT",
+        "IN_PROGRESS",
+        "BUSY",
+        "BUFFER_INVALID",
+        "RESULT_EXPIRED",
+        "INCOMPATIBLE_STATE",
+        "BACKEND_FAILURE",
+        "INTERNAL",
+    ];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            ErrorCode::InvalidArgument => "INVALID_ARGUMENT",
+            ErrorCode::Unsupported => "UNSUPPORTED",
+            ErrorCode::IdentityMismatch => "IDENTITY_MISMATCH",
+            ErrorCode::StaleEpoch => "STALE_EPOCH",
+            ErrorCode::StaleStep => "STALE_STEP",
+            ErrorCode::FutureStep => "FUTURE_STEP",
+            ErrorCode::InvalidPhase => "INVALID_PHASE",
+            ErrorCode::Conflict => "CONFLICT",
+            ErrorCode::InProgress => "IN_PROGRESS",
+            ErrorCode::Busy => "BUSY",
+            ErrorCode::BufferInvalid => "BUFFER_INVALID",
+            ErrorCode::ResultExpired => "RESULT_EXPIRED",
+            ErrorCode::IncompatibleState => "INCOMPATIBLE_STATE",
+            ErrorCode::BackendFailure => "BACKEND_FAILURE",
+            ErrorCode::Internal => "INTERNAL",
+        }
+    }
+
+    pub fn parse(s: &str) -> Result<ErrorCode> {
+        Ok(match s {
+            "INVALID_ARGUMENT" => ErrorCode::InvalidArgument,
+            "UNSUPPORTED" => ErrorCode::Unsupported,
+            "IDENTITY_MISMATCH" => ErrorCode::IdentityMismatch,
+            "STALE_EPOCH" => ErrorCode::StaleEpoch,
+            "STALE_STEP" => ErrorCode::StaleStep,
+            "FUTURE_STEP" => ErrorCode::FutureStep,
+            "INVALID_PHASE" => ErrorCode::InvalidPhase,
+            "CONFLICT" => ErrorCode::Conflict,
+            "IN_PROGRESS" => ErrorCode::InProgress,
+            "BUSY" => ErrorCode::Busy,
+            "BUFFER_INVALID" => ErrorCode::BufferInvalid,
+            "RESULT_EXPIRED" => ErrorCode::ResultExpired,
+            "INCOMPATIBLE_STATE" => ErrorCode::IncompatibleState,
+            "BACKEND_FAILURE" => ErrorCode::BackendFailure,
+            "INTERNAL" => ErrorCode::Internal,
+            _ => return err("code is not one of the fifteen domain error codes"),
+        })
+    }
+
+    /// The codes that are raised strictly before any mutation, so their certainty is `none`.
+    pub fn is_before_mutation(self) -> bool {
+        matches!(
+            self,
+            ErrorCode::InvalidArgument
+                | ErrorCode::Unsupported
+                | ErrorCode::IdentityMismatch
+                | ErrorCode::StaleEpoch
+                | ErrorCode::StaleStep
+                | ErrorCode::FutureStep
+                | ErrorCode::InvalidPhase
+                | ErrorCode::Conflict
+                | ErrorCode::InProgress
+                | ErrorCode::Busy
+                | ErrorCode::BufferInvalid
+                | ErrorCode::IncompatibleState
+        )
+    }
+}
+
+/// How certain the responder is that the operation mutated state.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum MutationCertainty {
+    /// Nothing was applied.
+    None,
+    /// The mutation completed.
+    Applied,
+    /// Completion is not established. "Errors after partial mutation use unknown unless
+    /// completion is established." (ipc-v1 section 7)
+    Unknown,
+}
+
+impl MutationCertainty {
+    pub const ALL: &'static [&'static str] = &["none", "applied", "unknown"];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            MutationCertainty::None => "none",
+            MutationCertainty::Applied => "applied",
+            MutationCertainty::Unknown => "unknown",
+        }
+    }
+
+    pub fn parse(s: &str) -> Result<MutationCertainty> {
+        match s {
+            "none" => Ok(MutationCertainty::None),
+            "applied" => Ok(MutationCertainty::Applied),
+            "unknown" => Ok(MutationCertainty::Unknown),
+            _ => err("mutation must be none, applied or unknown"),
+        }
+    }
+}
+
+/// `SessionRpcRequest`: one domain operation, independent of the bus callId that carries it.
+#[derive(Clone, Debug, PartialEq)]
+pub struct SessionRpcRequest {
+    pub request_id: DomainRequestId,
+    pub scope: Option<Scope>,
+    pub params: Value,
+}
+
+impl SessionRpcRequest {
+    /// The canonical body digest of this request under `method` (ipc-v1 section 5).
+    pub fn body_digest(&self, method: &str) -> Result<String> {
+        canonical::body_digest(method, self.scope.as_ref(), &self.params)
+    }
+
+    /// The operation key of a step mutation issued by `worker_id` under `method`. Lifecycle
+    /// calls with a null scope have no step operation key.
+    pub fn operation_key(&self, method: &str, worker_id: &str) -> Result<canonical::OperationKey> {
+        let scope = self
+            .scope
+            .clone()
+            .ok_or_else(|| crate::scalar::wire_err("operation key: a step mutation has a scope"))?;
+        canonical::OperationKey::new(scope, method, worker_id)
+    }
+}
+
+impl DomainType for SessionRpcRequest {
+    const TYPE_NAME: &'static str = "SessionRpcRequest";
+
+    fn from_json(value: &Value) -> Result<SessionRpcRequest> {
+        let mut f = Fields::new(value, "SessionRpcRequest")?;
+        let request_id = DomainRequestId::read(&mut f, "requestId")?;
+        let scope = Scope::nullable_from_json(f.value("scope")?)?;
+        let params = f.object("params")?.clone();
+        f.finish()?;
+        let r = SessionRpcRequest {
+            request_id,
+            scope,
+            params: Value::Object(params),
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("requestId", self.request_id.to_json()),
+            ("scope", Scope::nullable_to_json(self.scope.as_ref())),
+            ("params", self.params.clone()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !self.params.is_object() {
+            return err("SessionRpcRequest: params must be an object");
+        }
+        if let Some(scope) = &self.scope {
+            scope.validate()?;
+        }
+        canonical::reject_bus_identities(&self.params)
+    }
+}
+
+/// `SessionRpcSuccess`: a terminal domain success, echoing the request scope.
+#[derive(Clone, Debug, PartialEq)]
+pub struct SessionRpcSuccess {
+    pub request_id: DomainRequestId,
+    pub worker_id: String,
+    pub incarnation_id: String,
+    pub scope: Option<Scope>,
+    pub result: Value,
+}
+
+impl DomainType for SessionRpcSuccess {
+    const TYPE_NAME: &'static str = "SessionRpcSuccess";
+
+    fn from_json(value: &Value) -> Result<SessionRpcSuccess> {
+        let mut f = Fields::new(value, "SessionRpcSuccess")?;
+        constant(&mut f, "type", "result")?;
+        let request_id = DomainRequestId::read(&mut f, "requestId")?;
+        let worker_id = f.id("workerId")?;
+        let incarnation_id = f.id("incarnationId")?;
+        let scope = Scope::nullable_from_json(f.value("scope")?)?;
+        let result = f.object("result")?.clone();
+        f.finish()?;
+        let s = SessionRpcSuccess {
+            request_id,
+            worker_id,
+            incarnation_id,
+            scope,
+            result: Value::Object(result),
+        };
+        s.validate()?;
+        Ok(s)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("type", "result".into()),
+            ("requestId", self.request_id.to_json()),
+            ("workerId", self.worker_id.clone().into()),
+            ("incarnationId", self.incarnation_id.clone().into()),
+            ("scope", Scope::nullable_to_json(self.scope.as_ref())),
+            ("result", self.result.clone()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.worker_id) || !is_id(&self.incarnation_id) {
+            return err("SessionRpcSuccess: workerId and incarnationId must be valid ids");
+        }
+        if !self.result.is_object() {
+            return err("SessionRpcSuccess: result must be an object");
+        }
+        if let Some(scope) = &self.scope {
+            scope.validate()?;
+        }
+        Ok(())
+    }
+}
+
+/// `SessionRpcFailure`: a terminal domain error with an explicit mutation certainty.
+#[derive(Clone, Debug, PartialEq)]
+pub struct SessionRpcFailure {
+    pub request_id: DomainRequestId,
+    pub worker_id: String,
+    pub incarnation_id: String,
+    pub scope: Option<Scope>,
+    pub code: ErrorCode,
+    pub message: String,
+    pub mutation: MutationCertainty,
+}
+
+impl DomainType for SessionRpcFailure {
+    const TYPE_NAME: &'static str = "SessionRpcFailure";
+
+    fn from_json(value: &Value) -> Result<SessionRpcFailure> {
+        let mut f = Fields::new(value, "SessionRpcFailure")?;
+        constant(&mut f, "type", "error")?;
+        let request_id = DomainRequestId::read(&mut f, "requestId")?;
+        let worker_id = f.id("workerId")?;
+        let incarnation_id = f.id("incarnationId")?;
+        let scope = Scope::nullable_from_json(f.value("scope")?)?;
+        let (code, message, mutation) = {
+            let v = f.value("error")?;
+            let mut e = Fields::new(v, "SessionRpcFailure.error")?;
+            let code = ErrorCode::parse(&enumeration(&mut e, "code", ErrorCode::ALL)?)?;
+            let message = bounded_string(&mut e, "message", MAX_MESSAGE_CODE_POINTS)?;
+            let mutation = MutationCertainty::parse(&enumeration(
+                &mut e,
+                "mutation",
+                MutationCertainty::ALL,
+            )?)?;
+            e.finish()?;
+            (code, message, mutation)
+        };
+        f.finish()?;
+        let failure = SessionRpcFailure {
+            request_id,
+            worker_id,
+            incarnation_id,
+            scope,
+            code,
+            message,
+            mutation,
+        };
+        failure.validate()?;
+        Ok(failure)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("type", "error".into()),
+            ("requestId", self.request_id.to_json()),
+            ("workerId", self.worker_id.clone().into()),
+            ("incarnationId", self.incarnation_id.clone().into()),
+            ("scope", Scope::nullable_to_json(self.scope.as_ref())),
+            (
+                "error",
+                obj(vec![
+                    ("code", self.code.as_str().into()),
+                    ("message", self.message.clone().into()),
+                    ("mutation", self.mutation.as_str().into()),
+                ]),
+            ),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.worker_id) || !is_id(&self.incarnation_id) {
+            return err("SessionRpcFailure: workerId and incarnationId must be valid ids");
+        }
+        if self.message.chars().count() > MAX_MESSAGE_CODE_POINTS {
+            return err("SessionRpcFailure: message is at most 512 code points");
+        }
+        if self.code.is_before_mutation() && self.mutation != MutationCertainty::None {
+            return err(format!(
+                "SessionRpcFailure: {} is raised before mutation, so mutation is \"none\"",
+                self.code.as_str()
+            ));
+        }
+        if let Some(scope) = &self.scope {
+            scope.validate()?;
+        }
+        Ok(())
+    }
+}
+
+/// A terminal domain outcome: success or failure.
+#[derive(Clone, Debug, PartialEq)]
+pub enum SessionRpcOutcome {
+    Success(SessionRpcSuccess),
+    Failure(SessionRpcFailure),
+}
+
+impl SessionRpcOutcome {
+    pub fn request_id(&self) -> &DomainRequestId {
+        match self {
+            SessionRpcOutcome::Success(s) => &s.request_id,
+            SessionRpcOutcome::Failure(f) => &f.request_id,
+        }
+    }
+
+    /// Replies echo the original scope (ipc-v1 section 3).
+    pub fn echoes(&self, request: &SessionRpcRequest) -> bool {
+        let scope = match self {
+            SessionRpcOutcome::Success(s) => &s.scope,
+            SessionRpcOutcome::Failure(f) => &f.scope,
+        };
+        self.request_id() == &request.request_id && scope == &request.scope
+    }
+}
+
+impl DomainType for SessionRpcOutcome {
+    const TYPE_NAME: &'static str = "SessionRpcOutcome";
+
+    fn from_json(value: &Value) -> Result<SessionRpcOutcome> {
+        let kind = value
+            .get("type")
+            .and_then(Value::as_str)
+            .ok_or_else(|| crate::scalar::wire_err("SessionRpcOutcome: missing type"))?;
+        match kind {
+            "result" => SessionRpcSuccess::from_json(value).map(SessionRpcOutcome::Success),
+            "error" => SessionRpcFailure::from_json(value).map(SessionRpcOutcome::Failure),
+            _ => err("SessionRpcOutcome: type must be \"result\" or \"error\""),
+        }
+    }
+
+    fn to_json(&self) -> Value {
+        match self {
+            SessionRpcOutcome::Success(s) => s.to_json(),
+            SessionRpcOutcome::Failure(f) => f.to_json(),
+        }
+    }
+
+    fn validate(&self) -> Result<()> {
+        match self {
+            SessionRpcOutcome::Success(s) => s.validate(),
+            SessionRpcOutcome::Failure(f) => f.validate(),
+        }
+    }
+}

services/flysim/crates/fly-session-types/src/scalar.rs

@@ -0,0 +1,725 @@
+//! The domain scalars of ipc-v1 section 2, and the four identities that must never be confused.
+//!
+//! `Id`, `U64` and `Digest` are the bus encodings: this module calls straight into
+//! [`flybus::wire`] instead of restating the regular expressions, and
+//! `tests/encodings.rs` pins that the two agree. Everything else here is domain-only:
+//! `Scope`, `RationalNs` (reduced, positive denominator, zero as `0/1`, checked arithmetic),
+//! `SchemaRef` and `TypedValue` with its 32-KiB canonical-JSON cap.
+
+use std::cmp::Ordering;
+
+use flybus::wire::{self, Fields, WireError};
+use serde_json::{Map, Value};
+
+use crate::canonical;
+
+pub type Result<T> = std::result::Result<T, WireError>;
+
+/// Every parsed domain type re-validates itself, so a value built in Rust and a value read
+/// from JSON are held to the same rules.
+pub trait DomainType: Sized {
+    /// The name this type has in the canonical schema set.
+    const TYPE_NAME: &'static str;
+
+    /// Reads and validates one JSON value. Unknown fields are refused.
+    fn from_json(value: &Value) -> Result<Self>;
+
+    /// The canonical JSON shape of this value.
+    fn to_json(&self) -> Value;
+
+    /// The rules that are not expressible as one field read: ranges that depend on another
+    /// field, uniqueness, ordering and size caps.
+    fn validate(&self) -> Result<()>;
+}
+
+pub fn err<T>(message: impl Into<String>) -> Result<T> {
+    Err(WireError(message.into()))
+}
+
+pub fn wire_err(message: impl Into<String>) -> WireError {
+    WireError(message.into())
+}
+
+pub(crate) fn obj(pairs: Vec<(&str, Value)>) -> Value {
+    let mut map = Map::new();
+    for (key, value) in pairs {
+        map.insert(key.to_owned(), value);
+    }
+    Value::Object(map)
+}
+
+/// A `U64` field: the decimal string encoding, never a JSON number.
+pub fn u64_json(n: u64) -> Value {
+    Value::String(n.to_string())
+}
+
+/// `Id`: `^[a-z0-9][a-z0-9._-]{0,63}$`, exactly the bus encoding.
+pub fn is_id(s: &str) -> bool {
+    wire::is_id(s)
+}
+
+/// `Digest`: 64 lowercase hexadecimal digits, exactly the bus encoding.
+pub fn is_digest(s: &str) -> bool {
+    wire::is_digest(s)
+}
+
+/// `U64`: `"0"` or `[1-9][0-9]*` up to `u64::MAX`, exactly the bus encoding.
+pub fn parse_u64(s: &str) -> Option<u64> {
+    wire::parse_u64(s)
+}
+
+// ---------------------------------------------------------------------------------------------
+// Field readers the bus reader does not have
+
+/// A finite JSON number. NaN and infinities never survive strict parsing; this also refuses
+/// integers outside the exactly representable double range, which canonical JSON cannot encode.
+pub fn finite(f: &mut Fields<'_>, key: &'static str) -> Result<f64> {
+    let value = f.value(key)?;
+    match value {
+        Value::Number(n) => canonical::finite_double(n)
+            .ok_or_else(|| wire_err(format!("{key} must be a finite JSON number"))),
+        _ => err(format!("{key} must be a finite JSON number")),
+    }
+}
+
+/// A finite JSON number inside `lo..=hi`, refused rather than clamped.
+pub fn finite_in(f: &mut Fields<'_>, key: &'static str, lo: f64, hi: f64) -> Result<f64> {
+    let n = finite(f, key)?;
+    if n < lo || n > hi {
+        return err(format!("{key} must be in [{lo}, {hi}]"));
+    }
+    Ok(n)
+}
+
+/// A JSON integer in `i32` range, the seed encoding Agent.Initialize uses.
+pub fn i32_field(f: &mut Fields<'_>, key: &'static str) -> Result<i32> {
+    let value = f.value(key)?;
+    match value.as_i64() {
+        Some(n) if i64::from(i32::MIN) <= n && n <= i64::from(i32::MAX) => Ok(n as i32),
+        _ => err(format!("{key} must be a signed 32-bit integer")),
+    }
+}
+
+/// One member of a closed string enum.
+pub fn enumeration(f: &mut Fields<'_>, key: &'static str, allowed: &[&str]) -> Result<String> {
+    let s = f.string(key)?;
+    if allowed.contains(&s) {
+        Ok(s.to_owned())
+    } else {
+        err(format!("{key} must be one of {}", allowed.join(", ")))
+    }
+}
+
+/// A string constant: a field whose only legal value is `expected`.
+pub fn constant(f: &mut Fields<'_>, key: &'static str, expected: &str) -> Result<()> {
+    let s = f.string(key)?;
+    if s == expected {
+        Ok(())
+    } else {
+        err(format!("{key} must be {expected:?}"))
+    }
+}
+
+/// A `true` constant.
+pub fn constant_true(f: &mut Fields<'_>, key: &'static str) -> Result<()> {
+    if f.boolean(key)? {
+        Ok(())
+    } else {
+        err(format!("{key} must be true"))
+    }
+}
+
+/// A string of at most `max` Unicode code points.
+pub fn bounded_string(f: &mut Fields<'_>, key: &'static str, max: usize) -> Result<String> {
+    let s = f.string(key)?;
+    if s.chars().count() > max {
+        return err(format!("{key} must be at most {max} code points"));
+    }
+    Ok(s.to_owned())
+}
+
+/// `null`, or a string of at most `max` code points.
+pub fn nullable_bounded_string(
+    f: &mut Fields<'_>,
+    key: &'static str,
+    max: usize,
+) -> Result<Option<String>> {
+    match f.value(key)? {
+        Value::Null => Ok(None),
+        _ => bounded_string(f, key, max).map(Some),
+    }
+}
+
+/// Reads an array of `lo..=hi` items through `read`, keeping the supplied order.
+pub fn list<T>(
+    f: &mut Fields<'_>,
+    key: &'static str,
+    lo: usize,
+    hi: usize,
+    read: impl Fn(&Value) -> Result<T>,
+) -> Result<Vec<T>> {
+    let items = f.array(key, lo, hi)?;
+    let mut out = Vec::with_capacity(items.len());
+    for item in items {
+        out.push(read(item).map_err(|e| wire_err(format!("{key}: {e}")))?);
+    }
+    Ok(out)
+}
+
+/// An array of `lo..=hi` `Id`s.
+pub fn id_list(f: &mut Fields<'_>, key: &'static str, lo: usize, hi: usize) -> Result<Vec<String>> {
+    list(f, key, lo, hi, |v| match v.as_str() {
+        Some(s) if is_id(s) => Ok(s.to_owned()),
+        _ => err("every entry must be an id"),
+    })
+}
+
+/// Fails on the first repeated key, naming it.
+pub fn require_unique<'a>(keys: impl IntoIterator<Item = &'a str>, what: &str) -> Result<()> {
+    let mut seen: Vec<&str> = Vec::new();
+    for key in keys {
+        if seen.contains(&key) {
+            return err(format!("{what}: duplicate {key:?}"));
+        }
+        seen.push(key);
+    }
+    Ok(())
+}
+
+/// Fails unless `actual` is exactly `expected`, in that order: descriptor order is part of
+/// the contract, not a set membership test.
+pub fn require_same_order<'a>(
+    actual: impl IntoIterator<Item = &'a str>,
+    expected: impl IntoIterator<Item = &'a str>,
+    what: &str,
+) -> Result<()> {
+    let actual: Vec<&str> = actual.into_iter().collect();
+    let expected: Vec<&str> = expected.into_iter().collect();
+    if actual != expected {
+        return err(format!(
+            "{what}: must list [{}] in that order, found [{}]",
+            expected.join(", "),
+            actual.join(", ")
+        ));
+    }
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------------------------
+// Scope
+
+/// `Scope`: the simulation timeline identity. Never the bus route or store incarnation.
+#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct Scope {
+    pub session_id: String,
+    pub epoch: String,
+    pub step: u64,
+}
+
+impl Scope {
+    pub fn new(session_id: &str, epoch: &str, step: u64) -> Result<Scope> {
+        let scope = Scope {
+            session_id: session_id.to_owned(),
+            epoch: epoch.to_owned(),
+            step,
+        };
+        scope.validate()?;
+        Ok(scope)
+    }
+
+    /// `null`, or a scope.
+    pub fn nullable_from_json(value: &Value) -> Result<Option<Scope>> {
+        match value {
+            Value::Null => Ok(None),
+            _ => Scope::from_json(value).map(Some),
+        }
+    }
+
+    pub fn nullable_to_json(scope: Option<&Scope>) -> Value {
+        scope.map_or(Value::Null, Scope::to_json)
+    }
+}
+
+impl DomainType for Scope {
+    const TYPE_NAME: &'static str = "Scope";
+
+    fn from_json(value: &Value) -> Result<Scope> {
+        let mut f = Fields::new(value, "Scope")?;
+        let session_id = f.id("sessionId")?;
+        let epoch = f.id("epoch")?;
+        let step = f.u64_string("step")?;
+        f.finish()?;
+        let scope = Scope {
+            session_id,
+            epoch,
+            step,
+        };
+        scope.validate()?;
+        Ok(scope)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("sessionId", self.session_id.clone().into()),
+            ("epoch", self.epoch.clone().into()),
+            ("step", u64_json(self.step)),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.session_id) {
+            return err("Scope: sessionId is not a valid id");
+        }
+        if !is_id(&self.epoch) {
+            return err("Scope: epoch is not a valid id");
+        }
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------------------------
+// RationalNs
+
+/// A nanosecond rational: reduced, positive denominator, zero encoded `0/1`.
+///
+/// ipc-v1 section 2: "Fractions are reduced, denominators positive, durations positive; zero
+/// is encoded 0/1. Arithmetic is checked." Durations are checked with
+/// [`RationalNs::require_positive`] by the fields that are durations; `worldTime` and a tick
+/// remainder are legitimately zero.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub struct RationalNs {
+    pub numerator: u64,
+    pub denominator: u64,
+}
+
+fn gcd(a: u64, b: u64) -> u64 {
+    let (mut a, mut b) = (a, b);
+    while b != 0 {
+        let t = a % b;
+        a = b;
+        b = t;
+    }
+    a
+}
+
+fn gcd128(a: u128, b: u128) -> u128 {
+    let (mut a, mut b) = (a, b);
+    while b != 0 {
+        let t = a % b;
+        a = b;
+        b = t;
+    }
+    a
+}
+
+impl RationalNs {
+    pub const ZERO: RationalNs = RationalNs {
+        numerator: 0,
+        denominator: 1,
+    };
+
+    /// Exactly the supplied pair, which must already be in canonical form.
+    pub fn new(numerator: u64, denominator: u64) -> Result<RationalNs> {
+        let r = RationalNs {
+            numerator,
+            denominator,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    /// Reduces first, then validates: the constructor for arithmetic results.
+    pub fn reduced(numerator: u128, denominator: u128) -> Result<RationalNs> {
+        if denominator == 0 {
+            return err("RationalNs: denominator must be positive");
+        }
+        let (n, d) = if numerator == 0 {
+            (0u128, 1u128)
+        } else {
+            let g = gcd128(numerator, denominator);
+            (numerator / g, denominator / g)
+        };
+        if n > u128::from(u64::MAX) || d > u128::from(u64::MAX) {
+            return err("RationalNs: reduced value does not fit U64");
+        }
+        RationalNs::new(n as u64, d as u64)
+    }
+
+    pub fn is_zero(&self) -> bool {
+        self.numerator == 0
+    }
+
+    /// Durations must be positive (ipc-v1 section 2).
+    pub fn require_positive(&self, what: &str) -> Result<()> {
+        if self.is_zero() {
+            return err(format!("{what}: duration must be positive"));
+        }
+        Ok(())
+    }
+
+    pub fn checked_add(&self, other: &RationalNs) -> Result<RationalNs> {
+        let n = u128::from(self.numerator) * u128::from(other.denominator)
+            + u128::from(other.numerator) * u128::from(self.denominator);
+        let d = u128::from(self.denominator) * u128::from(other.denominator);
+        RationalNs::reduced(n, d)
+    }
+
+    pub fn checked_sub(&self, other: &RationalNs) -> Result<RationalNs> {
+        let left = u128::from(self.numerator) * u128::from(other.denominator);
+        let right = u128::from(other.numerator) * u128::from(self.denominator);
+        if right > left {
+            return err("RationalNs: subtraction would be negative");
+        }
+        let d = u128::from(self.denominator) * u128::from(other.denominator);
+        RationalNs::reduced(left - right, d)
+    }
+
+    pub fn checked_mul_u64(&self, k: u64) -> Result<RationalNs> {
+        let n = u128::from(self.numerator)
+            .checked_mul(u128::from(k))
+            .ok_or_else(|| wire_err("RationalNs: multiplication overflowed"))?;
+        RationalNs::reduced(n, u128::from(self.denominator))
+    }
+
+    /// The step-v1 section 5 accumulator: `ticks = floor(self / tick)` and the remainder
+    /// `self - ticks * tick`, which is always `>= 0` and `< tick`.
+    pub fn divide_floor(&self, tick: &RationalNs) -> Result<(u64, RationalNs)> {
+        tick.require_positive("RationalNs::divide_floor tick")?;
+        let n = u128::from(self.numerator) * u128::from(tick.denominator);
+        let d = u128::from(self.denominator) * u128::from(tick.numerator);
+        let ticks = n / d;
+        if ticks > u128::from(u64::MAX) {
+            return err("RationalNs: tick count does not fit U64");
+        }
+        let ticks = ticks as u64;
+        let remainder = self.checked_sub(&tick.checked_mul_u64(ticks)?)?;
+        Ok((ticks, remainder))
+    }
+}
+
+impl PartialOrd for RationalNs {
+    fn partial_cmp(&self, other: &RationalNs) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for RationalNs {
+    fn cmp(&self, other: &RationalNs) -> Ordering {
+        let left = u128::from(self.numerator) * u128::from(other.denominator);
+        let right = u128::from(other.numerator) * u128::from(self.denominator);
+        left.cmp(&right)
+    }
+}
+
+impl DomainType for RationalNs {
+    const TYPE_NAME: &'static str = "RationalNs";
+
+    fn from_json(value: &Value) -> Result<RationalNs> {
+        let mut f = Fields::new(value, "RationalNs")?;
+        let numerator = f.u64_string("numerator")?;
+        let denominator = f.u64_string("denominator")?;
+        f.finish()?;
+        RationalNs::new(numerator, denominator)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("numerator", u64_json(self.numerator)),
+            ("denominator", u64_json(self.denominator)),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if self.denominator == 0 {
+            return err("RationalNs: denominator must be positive");
+        }
+        if self.numerator == 0 && self.denominator != 1 {
+            return err("RationalNs: zero is encoded 0/1");
+        }
+        if self.numerator != 0 && gcd(self.numerator, self.denominator) != 1 {
+            return err("RationalNs: fraction must be reduced");
+        }
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------------------------
+// SchemaRef and TypedValue
+
+/// `SchemaRef`: the identity of a registered typed payload schema. Version is 1..=65535.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct SchemaRef {
+    pub id: String,
+    pub version: u16,
+    pub digest: String,
+}
+
+impl SchemaRef {
+    pub fn new(id: &str, version: u16, digest: &str) -> Result<SchemaRef> {
+        let r = SchemaRef {
+            id: id.to_owned(),
+            version,
+            digest: digest.to_owned(),
+        };
+        r.validate()?;
+        Ok(r)
+    }
+}
+
+impl DomainType for SchemaRef {
+    const TYPE_NAME: &'static str = "SchemaRef";
+
+    fn from_json(value: &Value) -> Result<SchemaRef> {
+        let mut f = Fields::new(value, "SchemaRef")?;
+        let id = f.id("id")?;
+        let version = f.int("version", 1, 65_535)? as u16;
+        let digest = f.string("digest")?.to_owned();
+        f.finish()?;
+        let r = SchemaRef {
+            id,
+            version,
+            digest,
+        };
+        r.validate()?;
+        Ok(r)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("id", self.id.clone().into()),
+            ("version", Value::from(u64::from(self.version))),
+            ("digest", self.digest.clone().into()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        if !is_id(&self.id) {
+            return err("SchemaRef: id is not a valid id");
+        }
+        if self.version == 0 {
+            return err("SchemaRef: version must be 1..=65535");
+        }
+        if !is_digest(&self.digest) {
+            return err("SchemaRef: digest must be 64 lowercase hex digits");
+        }
+        Ok(())
+    }
+}
+
+/// The canonical-JSON size limit of one `TypedValue` (ipc-v1 section 2, workers-v1 section 1).
+pub const MAX_TYPED_VALUE_BYTES: usize = 32 * 1024;
+
+/// `TypedValue`: a schema identity plus an object, capped at 32 KiB of canonical JSON.
+#[derive(Clone, Debug, PartialEq)]
+pub struct TypedValue {
+    pub schema: SchemaRef,
+    pub value: Value,
+}
+
+impl TypedValue {
+    pub fn new(schema: SchemaRef, value: Value) -> Result<TypedValue> {
+        let t = TypedValue { schema, value };
+        t.validate()?;
+        Ok(t)
+    }
+
+    pub fn nullable_from_json(value: &Value) -> Result<Option<TypedValue>> {
+        match value {
+            Value::Null => Ok(None),
+            _ => TypedValue::from_json(value).map(Some),
+        }
+    }
+
+    pub fn nullable_to_json(value: Option<&TypedValue>) -> Value {
+        value.map_or(Value::Null, TypedValue::to_json)
+    }
+
+    /// The canonical JSON byte length of the whole typed value.
+    pub fn canonical_len(&self) -> Result<usize> {
+        canonical::canonicalize(&self.to_json()).map(|s| s.len())
+    }
+}
+
+impl DomainType for TypedValue {
+    const TYPE_NAME: &'static str = "TypedValue";
+
+    fn from_json(value: &Value) -> Result<TypedValue> {
+        let mut f = Fields::new(value, "TypedValue")?;
+        let schema = SchemaRef::from_json(f.value("schema")?)?;
+        let inner = f.value("value")?.clone();
+        f.finish()?;
+        let t = TypedValue {
+            schema,
+            value: inner,
+        };
+        t.validate()?;
+        Ok(t)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("schema", self.schema.to_json()),
+            ("value", self.value.clone()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        self.schema.validate()?;
+        if !self.value.is_object() {
+            return err("TypedValue: value must be an object");
+        }
+        let len = self.canonical_len()?;
+        if len > MAX_TYPED_VALUE_BYTES {
+            return err(format!(
+                "TypedValue: {len} bytes of canonical JSON exceeds the {MAX_TYPED_VALUE_BYTES}-byte limit"
+            ));
+        }
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------------------------
+// The four identities
+
+/// A bus RPC correlation id, `call-<U64>` (bus-v1 section 6). It is not a domain operation id:
+/// a safe domain retry keeps its [`DomainRequestId`] and gets a new `BusCallId`.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct BusCallId(String);
+
+/// A domain operation id, `req-` plus a canonical `U64` serial (ipc-v1 section 5).
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct DomainRequestId(String);
+
+/// The identity of an immutable artifact: store incarnation, artifact id and generation.
+/// Not an address, not authority to read, and not an [`crate::workers::AssetRef`].
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct ArtifactIdentity {
+    pub store_id: String,
+    pub artifact_id: String,
+    pub generation: u64,
+}
+
+/// Which kind of ownership root a token names.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum OwnerKind {
+    /// One recipient's delivery, `dlv-<U64>`.
+    Delivery,
+    /// An explicit artifact hold, `own-<U64>`.
+    Hold,
+}
+
+/// A delivery or explicit-hold owner token. Connection-private: it never appears in a domain
+/// payload or a canonical body digest.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct OwnerToken {
+    token: String,
+    kind: OwnerKind,
+}
+
+macro_rules! serial_identity {
+    ($type:ty, $prefix:literal, $what:literal) => {
+        impl $type {
+            /// Parses the canonical `prefix-<U64>` form; any other prefix is refused, which is
+            /// what keeps the four identities from being swapped for one another.
+            pub fn parse(s: &str) -> Result<Self> {
+                match wire::parse_serial_id($prefix, s) {
+                    Some(_) => Ok(Self(s.to_owned())),
+                    None => err(concat!($what, " must be canonical ", $prefix, "-<U64>")),
+                }
+            }
+
+            pub fn from_serial(serial: u64) -> Self {
+                Self(wire::serial_id($prefix, serial))
+            }
+
+            pub fn serial(&self) -> u64 {
+                wire::parse_serial_id($prefix, &self.0).expect("validated on construction")
+            }
+
+            pub fn as_str(&self) -> &str {
+                &self.0
+            }
+
+            pub fn read(f: &mut Fields<'_>, key: &'static str) -> Result<Self> {
+                let s = f.string(key)?;
+                Self::parse(s).map_err(|e| wire_err(format!("{key}: {e}")))
+            }
+
+            pub fn read_nullable(f: &mut Fields<'_>, key: &'static str) -> Result<Option<Self>> {
+                match f.value(key)? {
+                    Value::Null => Ok(None),
+                    _ => Self::read(f, key).map(Some),
+                }
+            }
+
+            pub fn to_json(&self) -> Value {
+                Value::String(self.0.clone())
+            }
+        }
+    };
+}
+
+serial_identity!(BusCallId, "call", "a bus callId");
+serial_identity!(DomainRequestId, "req", "a domain requestId");
+
+impl OwnerToken {
+    pub fn parse(s: &str) -> Result<OwnerToken> {
+        if wire::parse_serial_id("dlv", s).is_some() {
+            return Ok(OwnerToken {
+                token: s.to_owned(),
+                kind: OwnerKind::Delivery,
+            });
+        }
+        if wire::parse_serial_id("own", s).is_some() {
+            return Ok(OwnerToken {
+                token: s.to_owned(),
+                kind: OwnerKind::Hold,
+            });
+        }
+        err("an owner token must be canonical dlv-<U64> or own-<U64>")
+    }
+
+    pub fn delivery(serial: u64) -> OwnerToken {
+        OwnerToken {
+            token: wire::serial_id("dlv", serial),
+            kind: OwnerKind::Delivery,
+        }
+    }
+
+    pub fn hold(serial: u64) -> OwnerToken {
+        OwnerToken {
+            token: wire::serial_id("own", serial),
+            kind: OwnerKind::Hold,
+        }
+    }
+
+    pub fn kind(&self) -> OwnerKind {
+        self.kind
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.token
+    }
+}
+
+impl ArtifactIdentity {
+    /// The identity half of a bus `ArtifactRef`: the parts that name the bytes, without the
+    /// byte length, content type or optional digest.
+    pub fn of(reference: &flybus::wire::ArtifactRef) -> ArtifactIdentity {
+        ArtifactIdentity {
+            store_id: reference.store_id.clone(),
+            artifact_id: reference.artifact_id.clone(),
+            generation: reference.generation,
+        }
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        if !is_id(&self.store_id) {
+            return err("ArtifactIdentity: storeId is not a valid id");
+        }
+        if !is_id(&self.artifact_id) {
+            return err("ArtifactIdentity: artifactId is not a valid id");
+        }
+        Ok(())
+    }
+}

services/flysim/crates/fly-session-types/src/seed.rs

@@ -0,0 +1,70 @@
+//! `seed-derivation-v1`: independent per-agent seeds from one recorded master seed.
+//!
+//! The specification is `docs/design/session-framework/seed-derivation-v1.md`; this is its
+//! reference implementation, and `fixtures/seed-vectors.json` its test vectors, which the
+//! TypeScript package reproduces.
+
+use sha2::{Digest as _, Sha256};
+
+use crate::canonical;
+use crate::scalar::{Result, err, is_id};
+
+/// The algorithm identity. It is part of composition identity: changing any byte of the
+/// derivation requires a new id.
+pub const ALGORITHM: &str = "seed-derivation-v1";
+
+/// The domain separation prefix hashed before the inputs.
+pub const PREFIX: &str = "flybrain/seed-derivation-v1";
+
+/// The SHA-256 of the derivation material for one agent, lowercase hex.
+pub fn material_digest(master_seed: u64, agent_id: &str) -> Result<String> {
+    Ok(canonical::sha256_hex(&material(master_seed, agent_id)?))
+}
+
+/// The exact bytes hashed: the prefix, the master seed as a canonical `U64` decimal string and
+/// the agent id, each followed by one `\n`.
+pub fn material(master_seed: u64, agent_id: &str) -> Result<Vec<u8>> {
+    if !is_id(agent_id) {
+        return err("seed derivation: agentId is not a valid id");
+    }
+    Ok(format!("{PREFIX}\n{master_seed}\n{agent_id}\n").into_bytes())
+}
+
+/// The signed 32-bit seed `Agent.Initialize` takes for `agent_id`.
+///
+/// The digest is read as eight big-endian `u32` lanes; the first nonzero lane becomes the
+/// seed, reinterpreted as two's-complement `i32`. Skipping zero lanes keeps the seed usable
+/// by an xorshift generator, whose state must not be zero. If every lane were zero the
+/// material is rehashed with a counter suffix, which no observed input has needed.
+pub fn agent_seed(master_seed: u64, agent_id: &str) -> Result<i32> {
+    let mut material = material(master_seed, agent_id)?;
+    for round in 0u32..4 {
+        if round > 0 {
+            material.extend_from_slice(format!("{round}\n").as_bytes());
+        }
+        let digest = Sha256::digest(&material);
+        for lane in digest.chunks_exact(4) {
+            let word = u32::from_be_bytes([lane[0], lane[1], lane[2], lane[3]]);
+            if word != 0 {
+                return Ok(word as i32);
+            }
+        }
+    }
+    err("seed derivation: every lane of four digests was zero")
+}
+
+/// The seeds of a whole composition, in the order the agent ids are given.
+///
+/// Equal ids deliberately derive equal seeds: "Identical explicit seeds are allowed only when
+/// the experiment intentionally declares them" (workers-v1 section 2), so a composition with a
+/// repeated agent id is refused here rather than silently sharing a seed.
+pub fn composition_seeds(master_seed: u64, agent_ids: &[String]) -> Result<Vec<i32>> {
+    crate::scalar::require_unique(
+        agent_ids.iter().map(String::as_str),
+        "seed derivation: agentIds",
+    )?;
+    agent_ids
+        .iter()
+        .map(|id| agent_seed(master_seed, id))
+        .collect()
+}

services/flysim/crates/fly-session-types/src/trace.rs

@@ -0,0 +1,486 @@
+//! The trace format of step-v1 section 8, split into behaviour and operational metadata.
+//!
+//! Section 8 requires a record, for every transition, of the scope, the Prepare request ids,
+//! the agent/profile ids, tick counts and remainders, decision digests, the complete batch id
+//! and control digest, the acknowledged world boundary, observation producing boundaries, task
+//! event/outcome ids in order, every Commit acknowledgment and the published boundary. It then
+//! requires that sequential, concurrent and reversed runs "match, excluding wall time, request
+//! ids and other explicitly operational metadata".
+//!
+//! So this record has two halves. [`TraceBehaviour`] is what must match: it is ordered by
+//! agent id rather than by completion order, so a reversed dispatch produces an identical
+//! value. [`TraceOperational`] is what section 8 requires recording but excludes from the
+//! comparison: wall time, the domain request ids, the bus callIds and the delivery ids.
+//! [`TransitionTrace::behaviour_equals`] compares only the first half, and
+//! [`TransitionTrace::behaviour_diff`] names the fields that differ.
+
+use flybus::wire::Fields;
+use serde_json::Value;
+
+use crate::canonical;
+use crate::scalar::{
+    BusCallId, DomainRequestId, DomainType, OwnerToken, RationalNs, Result, Scope, err, is_digest,
+    is_id, list, obj, require_unique, u64_json,
+};
+use crate::workers::{MAX_AGENTS, MAX_RATE_ROLES};
+
+/// One agent's behaviour in one transition.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TraceAgent {
+    pub agent_id: String,
+    pub profile_digest: String,
+    pub ticks_advanced: u64,
+    pub brain_ticks: u64,
+    pub remainder: RationalNs,
+    pub decision_digest: String,
+    /// The boundary this agent acknowledged in its Commit reply.
+    pub committed_step: u64,
+}
+
+/// One view's producing boundary, as observed in this transition.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TraceObservation {
+    pub view_id: String,
+    pub produced_step: u64,
+}
+
+/// The fields two runs of the same transition must agree on.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TraceBehaviour {
+    pub scope: Scope,
+    /// Sorted by agent id, never by completion order.
+    pub agents: Vec<TraceAgent>,
+    pub batch_id: String,
+    pub control_digest: String,
+    pub acknowledged_boundary: u64,
+    /// Sorted by view id.
+    pub observation_boundaries: Vec<TraceObservation>,
+    /// Task outcome ids in task order.
+    pub outcome_ids: Vec<String>,
+    /// Task event ids in task order.
+    pub event_ids: Vec<String>,
+    pub published_boundary: u64,
+}
+
+impl TraceBehaviour {
+    /// Sorts the order-free collections, so a trace recorded in completion order compares
+    /// equal to one recorded in dispatch order.
+    pub fn normalized(&self) -> TraceBehaviour {
+        let mut out = self.clone();
+        out.agents.sort_by(|a, b| a.agent_id.cmp(&b.agent_id));
+        out.observation_boundaries
+            .sort_by(|a, b| a.view_id.cmp(&b.view_id));
+        out
+    }
+
+    pub fn digest(&self) -> Result<String> {
+        canonical::digest_of(&self.normalized().to_json())
+    }
+}
+
+impl DomainType for TraceBehaviour {
+    const TYPE_NAME: &'static str = "TraceBehaviour";
+
+    fn from_json(value: &Value) -> Result<TraceBehaviour> {
+        let mut f = Fields::new(value, "TraceBehaviour")?;
+        let scope = Scope::from_json(f.value("scope")?)?;
+        let agents = list(&mut f, "agents", 1, MAX_AGENTS, |v| {
+            let mut a = Fields::new(v, "TraceBehaviour.agents")?;
+            let agent_id = a.id("agentId")?;
+            let profile_digest = a.string("profileDigest")?.to_owned();
+            let ticks_advanced = a.u64_string("ticksAdvanced")?;
+            let brain_ticks = a.u64_string("brainTicks")?;
+            let remainder = RationalNs::from_json(a.value("remainder")?)?;
+            let decision_digest = a.string("decisionDigest")?.to_owned();
+            let committed_step = a.u64_string("committedStep")?;
+            a.finish()?;
+            Ok(TraceAgent {
+                agent_id,
+                profile_digest,
+                ticks_advanced,
+                brain_ticks,
+                remainder,
+                decision_digest,
+                committed_step,
+            })
+        })?;
+        let batch_id = f.id("batchId")?;
+        let control_digest = f.string("controlDigest")?.to_owned();
+        let acknowledged_boundary = f.u64_string("acknowledgedBoundary")?;
+        let observation_boundaries = list(
+            &mut f,
+            "observationBoundaries",
+            0,
+            crate::media::MAX_VIEWS * 2,
+            |v| {
+                let mut o = Fields::new(v, "TraceBehaviour.observationBoundaries")?;
+                let view_id = o.id("viewId")?;
+                let produced_step = o.u64_string("producedStep")?;
+                o.finish()?;
+                Ok(TraceObservation {
+                    view_id,
+                    produced_step,
+                })
+            },
+        )?;
+        let outcome_ids = crate::scalar::id_list(&mut f, "outcomeIds", 0, MAX_RATE_ROLES)?;
+        let event_ids = crate::scalar::id_list(&mut f, "eventIds", 0, MAX_RATE_ROLES)?;
+        let published_boundary = f.u64_string("publishedBoundary")?;
+        f.finish()?;
+        let b = TraceBehaviour {
+            scope,
+            agents,
+            batch_id,
+            control_digest,
+            acknowledged_boundary,
+            observation_boundaries,
+            outcome_ids,
+            event_ids,
+            published_boundary,
+        };
+        b.validate()?;
+        Ok(b)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("scope", self.scope.to_json()),
+            (
+                "agents",
+                Value::Array(
+                    self.agents
+                        .iter()
+                        .map(|a| {
+                            obj(vec![
+                                ("agentId", a.agent_id.clone().into()),
+                                ("profileDigest", a.profile_digest.clone().into()),
+                                ("ticksAdvanced", u64_json(a.ticks_advanced)),
+                                ("brainTicks", u64_json(a.brain_ticks)),
+                                ("remainder", a.remainder.to_json()),
+                                ("decisionDigest", a.decision_digest.clone().into()),
+                                ("committedStep", u64_json(a.committed_step)),
+                            ])
+                        })
+                        .collect(),
+                ),
+            ),
+            ("batchId", self.batch_id.clone().into()),
+            ("controlDigest", self.control_digest.clone().into()),
+            ("acknowledgedBoundary", u64_json(self.acknowledged_boundary)),
+            (
+                "observationBoundaries",
+                Value::Array(
+                    self.observation_boundaries
+                        .iter()
+                        .map(|o| {
+                            obj(vec![
+                                ("viewId", o.view_id.clone().into()),
+                                ("producedStep", u64_json(o.produced_step)),
+                            ])
+                        })
+                        .collect(),
+                ),
+            ),
+            (
+                "outcomeIds",
+                Value::Array(self.outcome_ids.iter().map(|i| i.clone().into()).collect()),
+            ),
+            (
+                "eventIds",
+                Value::Array(self.event_ids.iter().map(|i| i.clone().into()).collect()),
+            ),
+            ("publishedBoundary", u64_json(self.published_boundary)),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        self.scope.validate()?;
+        if self.agents.is_empty() || self.agents.len() > MAX_AGENTS {
+            return err("TraceBehaviour: 1..=4 agents");
+        }
+        require_unique(
+            self.agents.iter().map(|a| a.agent_id.as_str()),
+            "TraceBehaviour.agents",
+        )?;
+        for agent in &self.agents {
+            if !is_id(&agent.agent_id) {
+                return err("TraceBehaviour: agentId is not a valid id");
+            }
+            if !is_digest(&agent.profile_digest) || !is_digest(&agent.decision_digest) {
+                return err("TraceBehaviour: agent digests must be 64 lowercase hex digits");
+            }
+            agent.remainder.validate()?;
+            if agent.committed_step != self.scope.step + 1 {
+                return err(
+                    "TraceBehaviour: every commit acknowledgment is the transition's next boundary",
+                );
+            }
+        }
+        if !is_id(&self.batch_id) {
+            return err("TraceBehaviour: batchId is not a valid id");
+        }
+        if !is_digest(&self.control_digest) {
+            return err("TraceBehaviour: controlDigest must be 64 lowercase hex digits");
+        }
+        if self.acknowledged_boundary != self.scope.step + 1 {
+            return err("TraceBehaviour: the acknowledged boundary is scope.step + 1");
+        }
+        if self.published_boundary != self.acknowledged_boundary {
+            return err(
+                "TraceBehaviour: the published boundary is the boundary every agent committed",
+            );
+        }
+        require_unique(
+            self.observation_boundaries
+                .iter()
+                .map(|o| o.view_id.as_str()),
+            "TraceBehaviour.observationBoundaries",
+        )?;
+        require_unique(
+            self.event_ids.iter().map(String::as_str),
+            "TraceBehaviour.eventIds",
+        )?;
+        require_unique(
+            self.outcome_ids.iter().map(String::as_str),
+            "TraceBehaviour.outcomeIds",
+        )?;
+        Ok(())
+    }
+}
+
+/// One agent's domain request id for one phase.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TraceRequest {
+    pub agent_id: String,
+    pub request_id: DomainRequestId,
+}
+
+/// What step-v1 section 8 records but excludes from the comparison.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TraceOperational {
+    /// Wall time is for pacing, health and presentation only (step-v1 section 5).
+    pub wall_time_ns: u64,
+    pub prepare_request_ids: Vec<TraceRequest>,
+    pub advance_request_id: DomainRequestId,
+    pub commit_request_ids: Vec<TraceRequest>,
+    /// The transport correlation ids this transition happened to use. A safe retry changes
+    /// these and nothing in [`TraceBehaviour`].
+    pub bus_call_ids: Vec<BusCallId>,
+    pub delivery_ids: Vec<OwnerToken>,
+}
+
+impl DomainType for TraceOperational {
+    const TYPE_NAME: &'static str = "TraceOperational";
+
+    fn from_json(value: &Value) -> Result<TraceOperational> {
+        let mut f = Fields::new(value, "TraceOperational")?;
+        let wall_time_ns = f.u64_string("wallTimeNs")?;
+        let read_requests = |v: &Value| -> Result<TraceRequest> {
+            let mut r = Fields::new(v, "TraceOperational request")?;
+            let agent_id = r.id("agentId")?;
+            let request_id = DomainRequestId::read(&mut r, "requestId")?;
+            r.finish()?;
+            Ok(TraceRequest {
+                agent_id,
+                request_id,
+            })
+        };
+        let prepare_request_ids = list(&mut f, "prepareRequestIds", 1, MAX_AGENTS, read_requests)?;
+        let advance_request_id = DomainRequestId::read(&mut f, "advanceRequestId")?;
+        let commit_request_ids = list(&mut f, "commitRequestIds", 1, MAX_AGENTS, read_requests)?;
+        let bus_call_ids = list(&mut f, "busCallIds", 0, 64, |v| match v.as_str() {
+            Some(s) => BusCallId::parse(s),
+            None => err("every busCallId must be a string"),
+        })?;
+        let delivery_ids = list(&mut f, "deliveryIds", 0, 64, |v| match v.as_str() {
+            Some(s) => OwnerToken::parse(s),
+            None => err("every deliveryId must be a string"),
+        })?;
+        f.finish()?;
+        let o = TraceOperational {
+            wall_time_ns,
+            prepare_request_ids,
+            advance_request_id,
+            commit_request_ids,
+            bus_call_ids,
+            delivery_ids,
+        };
+        o.validate()?;
+        Ok(o)
+    }
+
+    fn to_json(&self) -> Value {
+        let requests = |items: &[TraceRequest]| {
+            Value::Array(
+                items
+                    .iter()
+                    .map(|r| {
+                        obj(vec![
+                            ("agentId", r.agent_id.clone().into()),
+                            ("requestId", r.request_id.to_json()),
+                        ])
+                    })
+                    .collect(),
+            )
+        };
+        obj(vec![
+            ("wallTimeNs", u64_json(self.wall_time_ns)),
+            ("prepareRequestIds", requests(&self.prepare_request_ids)),
+            ("advanceRequestId", self.advance_request_id.to_json()),
+            ("commitRequestIds", requests(&self.commit_request_ids)),
+            (
+                "busCallIds",
+                Value::Array(self.bus_call_ids.iter().map(BusCallId::to_json).collect()),
+            ),
+            (
+                "deliveryIds",
+                Value::Array(
+                    self.delivery_ids
+                        .iter()
+                        .map(|t| Value::String(t.as_str().to_owned()))
+                        .collect(),
+                ),
+            ),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        require_unique(
+            self.prepare_request_ids.iter().map(|r| r.agent_id.as_str()),
+            "TraceOperational.prepareRequestIds",
+        )?;
+        require_unique(
+            self.commit_request_ids.iter().map(|r| r.agent_id.as_str()),
+            "TraceOperational.commitRequestIds",
+        )?;
+        require_unique(
+            self.bus_call_ids.iter().map(BusCallId::as_str),
+            "TraceOperational.busCallIds",
+        )?;
+        require_unique(
+            self.delivery_ids.iter().map(OwnerToken::as_str),
+            "TraceOperational.deliveryIds",
+        )?;
+        Ok(())
+    }
+}
+
+/// One transition's trace: behaviour plus operational metadata.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TransitionTrace {
+    pub behaviour: TraceBehaviour,
+    pub operational: TraceOperational,
+}
+
+impl TransitionTrace {
+    /// Behaviour equality: the comparison step-v1 section 8 asks for.
+    pub fn behaviour_equals(&self, other: &TransitionTrace) -> bool {
+        self.behaviour.normalized() == other.behaviour.normalized()
+    }
+
+    /// The behaviour fields that differ, named. Empty when [`Self::behaviour_equals`] holds.
+    pub fn behaviour_diff(&self, other: &TransitionTrace) -> Vec<String> {
+        let (a, b) = (self.behaviour.normalized(), other.behaviour.normalized());
+        let mut out = Vec::new();
+        if a.scope != b.scope {
+            out.push(format!("scope: {:?} vs {:?}", a.scope, b.scope));
+        }
+        if a.batch_id != b.batch_id {
+            out.push(format!("batchId: {} vs {}", a.batch_id, b.batch_id));
+        }
+        if a.control_digest != b.control_digest {
+            out.push("controlDigest differs".to_owned());
+        }
+        if a.acknowledged_boundary != b.acknowledged_boundary {
+            out.push(format!(
+                "acknowledgedBoundary: {} vs {}",
+                a.acknowledged_boundary, b.acknowledged_boundary
+            ));
+        }
+        if a.published_boundary != b.published_boundary {
+            out.push(format!(
+                "publishedBoundary: {} vs {}",
+                a.published_boundary, b.published_boundary
+            ));
+        }
+        if a.observation_boundaries != b.observation_boundaries {
+            out.push("observationBoundaries differ".to_owned());
+        }
+        if a.outcome_ids != b.outcome_ids {
+            out.push("outcomeIds differ".to_owned());
+        }
+        if a.event_ids != b.event_ids {
+            out.push("eventIds differ".to_owned());
+        }
+        let ids_a: Vec<&str> = a.agents.iter().map(|x| x.agent_id.as_str()).collect();
+        let ids_b: Vec<&str> = b.agents.iter().map(|x| x.agent_id.as_str()).collect();
+        if ids_a != ids_b {
+            out.push(format!(
+                "agents: [{}] vs [{}]",
+                ids_a.join(", "),
+                ids_b.join(", ")
+            ));
+        } else {
+            for (left, right) in a.agents.iter().zip(&b.agents) {
+                if left != right {
+                    out.push(format!("agent {}: behaviour differs", left.agent_id));
+                }
+            }
+        }
+        out
+    }
+
+    /// Two whole runs agree on behaviour, transition by transition.
+    pub fn runs_equal(left: &[TransitionTrace], right: &[TransitionTrace]) -> bool {
+        left.len() == right.len() && left.iter().zip(right).all(|(a, b)| a.behaviour_equals(b))
+    }
+}
+
+impl DomainType for TransitionTrace {
+    const TYPE_NAME: &'static str = "TransitionTrace";
+
+    fn from_json(value: &Value) -> Result<TransitionTrace> {
+        let mut f = Fields::new(value, "TransitionTrace")?;
+        let behaviour = TraceBehaviour::from_json(f.value("behaviour")?)?;
+        let operational = TraceOperational::from_json(f.value("operational")?)?;
+        f.finish()?;
+        let t = TransitionTrace {
+            behaviour,
+            operational,
+        };
+        t.validate()?;
+        Ok(t)
+    }
+
+    fn to_json(&self) -> Value {
+        obj(vec![
+            ("behaviour", self.behaviour.to_json()),
+            ("operational", self.operational.to_json()),
+        ])
+    }
+
+    fn validate(&self) -> Result<()> {
+        self.behaviour.validate()?;
+        self.operational.validate()?;
+        let behaviour_agents: Vec<&str> = self
+            .behaviour
+            .agents
+            .iter()
+            .map(|a| a.agent_id.as_str())
+            .collect();
+        for phase in [
+            &self.operational.prepare_request_ids,
+            &self.operational.commit_request_ids,
+        ] {
+            for request in phase {
+                if !behaviour_agents.contains(&request.agent_id.as_str()) {
+                    return err(format!(
+                        "TransitionTrace: request recorded for {:?}, which is not in the transition",
+                        request.agent_id
+                    ));
+                }
+            }
+        }
+        Ok(())
+    }
+}