lazaret/crates/lazaret-core/tests/schema_records.rs

use std::collections::{HashMap, HashSet};
use std::fmt::Debug;
use std::num::NonZeroU16;

use lazaret_core::codec::{CodecError, MAX_SCHEMA_RECORD_BYTES, decode, encode};
use lazaret_core::identity::{
    BootId, BundleDigest, CapabilityId, CertificateDigest, ClientId, ConfigurationFingerprint,
    ContainerId, CoordinatorId, DerivedPolicyDigest, DeviceId, DomainId, FileDescriptor,
    GenerationId, GuestProcessGroupId, GuestProcessId, GuestProcessStartTimeTicks, HostPath,
    HostPidLocator, IdMapDigest, InodeId, LegacyUnverifiedProcessIdentityV1, MountId,
    MountIdentity, NamespaceIdentity, NetworkSidecarId, PidfsDeviceId, PidfsInodeId,
    PidfsProcessAuthorityV1, PidfsProcessIdentityV1, PolicyId, PrepareOperationId,
    PreparedArtifactId, ProcessStartTimeTicks, ProtocolCredentialId, ReadinessFingerprint,
    RequestDigest, RequestId, SandboxId, SelectedManifestDigest, SessionAnchorId, SessionId,
    SourceIdentity, StreamDigest, TrustManifestDigest,
};
use lazaret_core::records::*;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

#[derive(Deserialize, Serialize)]
struct RawParticipant {
    generation_id: GenerationId,
    role: ParticipantRole,
    state: ParticipantState,
    ownership: Option<ParticipantOwnership>,
}

#[derive(Deserialize, Serialize)]
struct RawTlsCredential {
    generation_identity: GenerationIdentity,
    role: TlsRole,
    subject_id: TlsSubjectId,
    certificate_der: Vec<u8>,
    certificate_sha256: CertificateDigest,
}

#[derive(Deserialize, Serialize)]
struct RawProtocolCredential {
    protocol_version: ProtocolVersion,
    credential_id: ProtocolCredentialId,
    generation_identity: GenerationIdentity,
    role: ProtocolCredentialRole,
    subject_id: ProtocolSubjectId,
    public_key: Vec<u8>,
    issuer_signature: Vec<u8>,
}

#[derive(Deserialize, Serialize)]
struct RawSandboxCredentialBundle {
    protocol_version: ProtocolVersion,
    generation_identity: GenerationIdentity,
    trust_manifest: TrustManifest,
    coordinator_protocol_credential: ProtocolCredential,
    sandbox_protocol_credential: ProtocolCredential,
    sandbox_private_key: Vec<u8>,
}

#[derive(Deserialize, Serialize)]
struct RawMetadata {
    schema_id: String,
    schema_version: u32,
}

#[derive(Serialize)]
struct MissingVersion {
    schema_id: String,
}

#[derive(Serialize)]
struct RawDomainStore {
    metadata: StoreMetadata,
    domain: Domain,
    generations: Vec<Generation>,
    participants: Vec<Participant>,
    sessions: Vec<Session>,
    session_terminals: Vec<SessionTerminal>,
    mailboxes: Vec<Mailbox>,
}

#[derive(Serialize)]
struct RawStorageCatalog {
    metadata: StoreMetadata,
    prepared_artifacts: Vec<PreparedArtifact>,
    artifact_leases: Vec<ArtifactLease>,
    prepare_operations: Vec<PrepareOperation>,
}

#[derive(Serialize)]
struct RawTerminalResize {
    session_id: SessionId,
    rows: u16,
    columns: u16,
    pixel_width: u16,
    pixel_height: u16,
}

#[derive(Serialize)]
struct RawSessionTerminal {
    generation_id: GenerationId,
    session_id: SessionId,
    state: SessionTerminalState,
    container_id: Option<ContainerId>,
    host_coordinator: PersistedHostProcessOwnership,
    pty_master: Option<PtyIdentity>,
    exec_client: PersistedHostProcessOwnership,
    guest_process: Option<GuestProcessIdentity>,
}

#[derive(Deserialize, Serialize)]
struct InvalidSessionExit {
    session_id: SessionId,
    outcome: SessionExitOutcome,
    exit_code: Option<i32>,
}

#[derive(Serialize)]
struct RawDerivedNetworkPolicy {
    schema_id: String,
    address_families: Vec<AddressFamilyDisposition>,
    default_actions: Vec<DefaultAction>,
    prefix_rules: Vec<PrefixRule>,
    resolvers: Vec<Resolver>,
}

#[derive(Serialize)]
struct RawNetworkReadinessAttestation {
    schema_id: String,
    domain_id: DomainId,
    generation_id: GenerationId,
    configuration_fingerprint: ConfigurationFingerprint,
    network_sidecar_id: NetworkSidecarId,
    derived_policy_digest: DerivedPolicyDigest,
    enforcement_namespace: NamespaceIdentity,
    runtime_supervisor: PidfsProcessIdentityV1,
    connectivity_helper: PidfsProcessIdentityV1,
    observation: NetworkObservation,
}

#[derive(Serialize)]
struct RawNetworkReadinessIdentity {
    schema_id: String,
    domain_id: DomainId,
    generation_id: GenerationId,
    configuration_fingerprint: ConfigurationFingerprint,
    network_sidecar_id: NetworkSidecarId,
    derived_policy_digest: DerivedPolicyDigest,
    enforcement_namespace: NamespaceIdentity,
}

fn opaque(byte: u8) -> [u8; 32] {
    [byte; 32]
}

fn domain_id() -> DomainId {
    DomainId::new(opaque(1))
}

fn generation_id() -> GenerationId {
    GenerationId::new(7)
}

fn coordinator_id() -> CoordinatorId {
    CoordinatorId::new(opaque(2))
}

fn sidecar_id() -> NetworkSidecarId {
    NetworkSidecarId::new(opaque(3))
}

fn sandbox_id() -> SandboxId {
    SandboxId::new(opaque(4))
}

fn session_id() -> SessionId {
    SessionId::new(opaque(5))
}

fn process(seed: u32) -> PidfsProcessIdentityV1 {
    PidfsProcessIdentityV1::new(
        PidfsProcessAuthorityV1 {
            boot_id: BootId::new([0x42; 16]),
            pidfs_st_dev: PidfsDeviceId::new(7),
            pidfs_st_ino: PidfsInodeId::new(u64::from(seed)),
        },
        HostPidLocator::new(seed),
        ProcessStartTimeTicks::new(u64::from(seed) * 10),
    )
}

fn owned_process(seed: u32) -> PersistedHostProcessOwnership {
    PersistedHostProcessOwnership::committed(process(seed))
}

fn namespace() -> NamespaceIdentity {
    NamespaceIdentity {
        device: DeviceId::new(8),
        inode: InodeId::new(9),
    }
}

fn mount() -> MountIdentity {
    MountIdentity {
        mount_id: MountId::new(10),
        device: DeviceId::new(11),
        inode: InodeId::new(12),
    }
}

fn generation_identity() -> GenerationIdentity {
    GenerationIdentity {
        agent_profile: "trusted".into(),
        domain_id: domain_id(),
        generation_id: generation_id(),
        configuration_fingerprint: ConfigurationFingerprint::new(opaque(6)),
        coordinator_id: coordinator_id(),
        sandbox_id: sandbox_id(),
    }
}

fn trust_manifest() -> TrustManifest {
    TrustManifest {
        protocol_version: ProtocolVersion::V1,
        generation_identity: generation_identity(),
        tls_ca_sha256: CertificateDigest::new(opaque(7)),
        protocol_issuer_public_key: vec![8; 32],
        issuer_signature: vec![9; 64],
    }
}

fn protocol_credential(role: ProtocolCredentialRole) -> ProtocolCredential {
    match role {
        ProtocolCredentialRole::SessionCoordinator => ProtocolCredential::session_coordinator(
            ProtocolCredentialId::new(opaque(10)),
            generation_identity(),
            coordinator_id(),
            vec![11; 32],
            vec![12; 64],
        ),
        ProtocolCredentialRole::AgentSandbox => ProtocolCredential::agent_sandbox(
            ProtocolCredentialId::new(opaque(10)),
            generation_identity(),
            sandbox_id(),
            vec![11; 32],
            vec![12; 64],
        ),
    }
}

fn session_binding() -> SessionBinding {
    SessionBinding {
        generation_identity: generation_identity(),
        cli_id: ClientId::new(opaque(13)),
        session_id: session_id(),
        session_anchor_public_key: vec![14; 32],
    }
}

fn session_capability() -> SessionRequestCapability {
    SessionRequestCapability {
        protocol_version: ProtocolVersion::V1,
        capability_id: CapabilityId::new(opaque(15)),
        session_binding: session_binding(),
        coordinator_credential_id: ProtocolCredentialId::new(opaque(16)),
        sandbox_credential_id: ProtocolCredentialId::new(opaque(17)),
        coordinator_signature: vec![18; 64],
    }
}

fn network_observation() -> NetworkObservation {
    NetworkObservation {
        policy_delivery: PolicyDelivery::SealedMemfd,
        helper_ready: true,
        observed_routes: vec![IpNetwork {
            address: IpAddress::V4([10, 0, 0, 0]),
            prefix_length: 8,
        }],
        routes_attested: true,
    }
}

fn readiness_attestation() -> NetworkReadinessAttestation {
    NetworkReadinessAttestation::v1(NetworkReadinessAttestationFields {
        domain_id: domain_id(),
        generation_id: generation_id(),
        configuration_fingerprint: ConfigurationFingerprint::new(opaque(19)),
        network_sidecar_id: sidecar_id(),
        derived_policy_digest: DerivedPolicyDigest::new(opaque(20)),
        enforcement_namespace: namespace(),
        runtime_supervisor: process(21),
        connectivity_helper: process(22),
        observation: network_observation(),
    })
}

fn readiness_identity() -> NetworkReadinessIdentity {
    NetworkReadinessIdentity::v1(NetworkReadinessIdentityFields {
        domain_id: domain_id(),
        generation_id: generation_id(),
        configuration_fingerprint: ConfigurationFingerprint::new(opaque(19)),
        network_sidecar_id: sidecar_id(),
        derived_policy_digest: DerivedPolicyDigest::new(opaque(20)),
        enforcement_namespace: namespace(),
    })
}

fn reachability_probe() -> NetworkReachabilityProbe {
    NetworkReachabilityProbe {
        domain_id: domain_id(),
        generation_id: generation_id(),
        enforcement_namespace: namespace(),
        denied_destinations: vec![DeniedDestinationObservation {
            destination: IpAddress::V4([10, 1, 2, 3]),
            unreachable: true,
        }],
        permitted_destinations: vec![PermittedDestinationObservation {
            destination: IpAddress::V4([1, 1, 1, 1]),
            reachable: true,
        }],
        dns_observations: vec![DnsObservation {
            resolver: IpAddress::V4([9, 9, 9, 9]),
            query_name: "example.invalid".into(),
            resolved_addresses: vec![IpAddress::V6([1; 16])],
        }],
    }
}

fn readiness_evidence() -> NetworkReadinessEvidence {
    NetworkReadinessEvidence {
        domain_id: domain_id(),
        generation_id: generation_id(),
        network_sidecar_id: sidecar_id(),
        readiness_fingerprint: ReadinessFingerprint::new(opaque(23)),
        attestation: readiness_attestation(),
        readiness_identity: readiness_identity(),
        reachability_probe: reachability_probe(),
    }
}

fn prepared_ownership() -> PreparedLowerOwnership {
    PreparedLowerOwnership {
        prepared_artifact_id: Some(PreparedArtifactId::new(opaque(24))),
        source_identity: Some(SourceIdentity::new(vec![25, 26])),
        lease_state: Some(ParticipantState::LeaseHeld),
        path: Some(HostPath::new(vec![b'/', 0xff])),
        device: Some(DeviceId::new(27)),
        inode: Some(InodeId::new(28)),
    }
}

fn network_ownership() -> NetworkOwnership {
    NetworkOwnership {
        derived_policy_digest: Some(DerivedPolicyDigest::new(opaque(20))),
        readiness_fingerprint: Some(ReadinessFingerprint::new(opaque(23))),
        readiness_evidence: Some(readiness_evidence()),
        namespace: Some(namespace()),
        runtime_supervisor: owned_process(21),
    }
}

fn mutate_network_ownership(mutate: impl FnOnce(&mut NetworkOwnership)) -> NetworkOwnership {
    let mut ownership = network_ownership();
    mutate(&mut ownership);
    ownership
}

fn state_anchor_ownership(
    daemon: PersistedHostProcessOwnership,
    mount_identity: Option<MountIdentity>,
) -> StateAnchorOwnership {
    StateAnchorOwnership {
        lower_artifact_id: Some(PreparedArtifactId::new(opaque(24))),
        upper_path: Some(HostPath::new(b"upper".to_vec())),
        work_path: Some(HostPath::new(b"work".to_vec())),
        merged_path: Some(HostPath::new(b"merged".to_vec())),
        daemon,
        mount: mount_identity,
    }
}

fn pty() -> PtyIdentity {
    PtyIdentity {
        master_fd: FileDescriptor::new(31),
        device: DeviceId::new(32),
        inode: InodeId::new(33),
    }
}

fn guest_process() -> GuestProcessIdentity {
    GuestProcessIdentity {
        leader_pid: GuestProcessId::new(35),
        leader_start_time: GuestProcessStartTimeTicks::new(36),
        process_group_id: GuestProcessGroupId::new(37),
    }
}

fn active_state(role: ParticipantRole) -> ParticipantState {
    match role {
        ParticipantRole::PreparedLower => ParticipantState::LeaseHeld,
        ParticipantRole::StateAnchor | ParticipantRole::GitRouting => ParticipantState::Prepared,
        ParticipantRole::Network
        | ParticipantRole::Helper
        | ParticipantRole::Sandbox
        | ParticipantRole::Publication => ParticipantState::Committed,
    }
}

fn complete_ownership(role: ParticipantRole, state: ParticipantState) -> ParticipantOwnership {
    match role {
        ParticipantRole::PreparedLower => {
            let mut ownership = prepared_ownership();
            ownership.lease_state = Some(state);
            ParticipantOwnership::PreparedLower(ownership)
        }
        ParticipantRole::StateAnchor => ParticipantOwnership::StateAnchor(state_anchor_ownership(
            owned_process(41),
            Some(mount()),
        )),
        ParticipantRole::GitRouting => ParticipantOwnership::GitRouting(GitRoutingOwnership {
            mountpoint: Some(HostPath::new(b"workspace".to_vec())),
            host_root: Some(HostPath::new(b"host".to_vec())),
            isolated_root: Some(HostPath::new(b"isolated".to_vec())),
            daemon: owned_process(42),
            mount: Some(mount()),
        }),
        ParticipantRole::Network => ParticipantOwnership::Network(Box::new(network_ownership())),
        ParticipantRole::Helper => ParticipantOwnership::Helper(HelperOwnership {
            network_sidecar_id: Some(sidecar_id()),
            connectivity_helper: owned_process(22),
            runtime_supervisor: owned_process(21),
        }),
        ParticipantRole::Sandbox => ParticipantOwnership::Sandbox(SandboxOwnership {
            container_id: Some(ContainerId::new(opaque(29))),
            runtime_root: Some(HostPath::new(b"runtime".to_vec())),
            bundle_digest: Some(BundleDigest::new(opaque(43))),
            host_coordinator: owned_process(30),
            runtime_supervisor: owned_process(21),
            attached_runsc: owned_process(44),
        }),
        ParticipantRole::Publication => ParticipantOwnership::Publication(PublicationOwnership {
            socket_path: Some(HostPath::new(b"socket".to_vec())),
            identity_path: Some(HostPath::new(b"identity".to_vec())),
        }),
    }
}

fn release_ownership(role: ParticipantRole) -> ParticipantOwnership {
    let state = if role == ParticipantRole::PreparedLower {
        ParticipantState::ReleaseIntent
    } else {
        active_state(role)
    };
    complete_ownership(role, state)
}

fn terminal_snapshot(
    state: SessionTerminalState,
    host_coordinator: PersistedHostProcessOwnership,
    exec_client: PersistedHostProcessOwnership,
    has_physical_evidence: bool,
) -> Result<SessionTerminal, RecordInvariantError> {
    SessionTerminal::snapshot(SessionTerminalFields {
        generation_id: generation_id(),
        session_id: session_id(),
        state,
        container_id: has_physical_evidence.then(|| ContainerId::new(opaque(29))),
        host_coordinator,
        pty_master: has_physical_evidence.then_some(pty()),
        exec_client,
        guest_process: has_physical_evidence.then_some(guest_process()),
    })
}

#[derive(Clone, Copy, Debug)]
enum ParticipantProcessEdge {
    StateAnchorDaemon,
    GitRoutingDaemon,
    NetworkRuntimeSupervisor,
    HelperConnectivityHelper,
    HelperRuntimeSupervisor,
    SandboxHostCoordinator,
    SandboxRuntimeSupervisor,
    SandboxAttachedRunsc,
}

fn participant_edge_snapshot(
    edge: ParticipantProcessEdge,
    process_ownership: PersistedHostProcessOwnership,
) -> Participant {
    let (role, ownership) = match edge {
        ParticipantProcessEdge::StateAnchorDaemon => (
            ParticipantRole::StateAnchor,
            ParticipantOwnership::StateAnchor(state_anchor_ownership(process_ownership, None)),
        ),
        ParticipantProcessEdge::GitRoutingDaemon => (
            ParticipantRole::GitRouting,
            ParticipantOwnership::GitRouting(GitRoutingOwnership {
                mountpoint: Some(HostPath::new(b"workspace".to_vec())),
                host_root: Some(HostPath::new(b"host".to_vec())),
                isolated_root: Some(HostPath::new(b"isolated".to_vec())),
                daemon: process_ownership,
                mount: None,
            }),
        ),
        ParticipantProcessEdge::NetworkRuntimeSupervisor => (
            ParticipantRole::Network,
            ParticipantOwnership::Network(Box::new(NetworkOwnership {
                derived_policy_digest: Some(DerivedPolicyDigest::new(opaque(20))),
                readiness_fingerprint: None,
                readiness_evidence: None,
                namespace: None,
                runtime_supervisor: process_ownership,
            })),
        ),
        ParticipantProcessEdge::HelperConnectivityHelper => (
            ParticipantRole::Helper,
            ParticipantOwnership::Helper(HelperOwnership {
                network_sidecar_id: Some(sidecar_id()),
                connectivity_helper: process_ownership,
                runtime_supervisor: owned_process(121),
            }),
        ),
        ParticipantProcessEdge::HelperRuntimeSupervisor => (
            ParticipantRole::Helper,
            ParticipantOwnership::Helper(HelperOwnership {
                network_sidecar_id: Some(sidecar_id()),
                connectivity_helper: PersistedHostProcessOwnership::unassigned(),
                runtime_supervisor: process_ownership,
            }),
        ),
        ParticipantProcessEdge::SandboxHostCoordinator => (
            ParticipantRole::Sandbox,
            ParticipantOwnership::Sandbox(SandboxOwnership {
                container_id: Some(ContainerId::new(opaque(29))),
                runtime_root: None,
                bundle_digest: None,
                host_coordinator: process_ownership,
                runtime_supervisor: PersistedHostProcessOwnership::unassigned(),
                attached_runsc: PersistedHostProcessOwnership::unassigned(),
            }),
        ),
        ParticipantProcessEdge::SandboxRuntimeSupervisor => (
            ParticipantRole::Sandbox,
            ParticipantOwnership::Sandbox(SandboxOwnership {
                container_id: Some(ContainerId::new(opaque(29))),
                runtime_root: None,
                bundle_digest: None,
                host_coordinator: owned_process(122),
                runtime_supervisor: process_ownership,
                attached_runsc: PersistedHostProcessOwnership::unassigned(),
            }),
        ),
        ParticipantProcessEdge::SandboxAttachedRunsc => (
            ParticipantRole::Sandbox,
            ParticipantOwnership::Sandbox(SandboxOwnership {
                container_id: Some(ContainerId::new(opaque(29))),
                runtime_root: None,
                bundle_digest: None,
                host_coordinator: owned_process(122),
                runtime_supervisor: owned_process(123),
                attached_runsc: process_ownership,
            }),
        ),
    };
    Participant::snapshot(
        generation_id(),
        role,
        ParticipantState::Intent,
        Some(ownership),
    )
    .unwrap()
}

fn verify_shape<T>(record: T, seen: &mut HashSet<&'static str>)
where
    T: Clone + Debug + DeserializeOwned + Eq + SchemaType + Serialize,
{
    assert!(seen.insert(T::NAME), "duplicate test for {}", T::NAME);
    let encoded = encode(&record).expect("representative record encodes");
    assert_eq!(encode(&record).unwrap(), encoded);
    assert_eq!(decode::<T>(&encoded).unwrap(), record);

    let mut truncated = encoded.clone();
    truncated.pop();
    assert!(decode::<T>(&truncated).is_err());

    let mut trailing = encoded;
    trailing.push(0);
    assert!(matches!(
        decode::<T>(&trailing),
        Err(CodecError::TrailingPayload)
    ));
    drop(record);
}

#[test]
#[allow(clippy::too_many_lines)]
fn every_schema_shape_round_trips_and_rejects_malformed_or_unknown_versions() {
    let terminal = SessionTerminal::snapshot(SessionTerminalFields {
        generation_id: generation_id(),
        session_id: session_id(),
        state: SessionTerminalState::Running,
        container_id: Some(ContainerId::new(opaque(29))),
        host_coordinator: owned_process(30),
        pty_master: Some(PtyIdentity {
            master_fd: FileDescriptor::new(31),
            device: DeviceId::new(32),
            inode: InodeId::new(33),
        }),
        exec_client: owned_process(34),
        guest_process: Some(GuestProcessIdentity {
            leader_pid: GuestProcessId::new(35),
            leader_start_time: GuestProcessStartTimeTicks::new(36),
            process_group_id: GuestProcessGroupId::new(37),
        }),
    })
    .unwrap();
    let domain = Domain {
        domain_id: domain_id(),
        profile_name: "trusted".into(),
        canonical_workspace: HostPath::new(vec![b'/', b'w', 0xfe]),
    };
    let generation = Generation {
        domain_id: domain_id(),
        generation_id: generation_id(),
        configuration_fingerprint: ConfigurationFingerprint::new(opaque(6)),
        coordinator_id: coordinator_id(),
        network_sidecar_id: sidecar_id(),
        sandbox_id: sandbox_id(),
        durable_decision: DurableDecision::Preparing,
    };
    let participant = Participant::snapshot(
        generation_id(),
        ParticipantRole::PreparedLower,
        ParticipantState::LeaseHeld,
        Some(ParticipantOwnership::PreparedLower(prepared_ownership())),
    )
    .unwrap();
    let session = Session {
        generation_id: generation_id(),
        session_id: session_id(),
        cli_id: ClientId::new(opaque(13)),
        session_anchor_id: SessionAnchorId::new(opaque(38)),
        session_phase: SessionPhase::Attached,
    };
    let mailbox = Mailbox {
        generation_id: generation_id(),
        identity_path: HostPath::new(b"identity".to_vec()),
        inbox_path: HostPath::new(b"inbox".to_vec()),
        outbox_path: HostPath::new(b"outbox".to_vec()),
        state: ParticipantState::Committed,
    };
    let metadata = StoreMetadata::domain_store_v2();
    let tls_credential = TlsCredential::session_coordinator(
        generation_identity(),
        coordinator_id(),
        vec![39, 40],
        CertificateDigest::new(opaque(41)),
    );
    let request = MailboxRequest {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: RequestId::new(opaque(42)),
        session_capability: session_capability(),
        sandbox_credential_id: ProtocolCredentialId::new(opaque(17)),
        command_argv: vec![b"printf".to_vec(), vec![0xff]],
        justification: "build dependency".into(),
    };
    let prepared_artifact = PreparedArtifact {
        prepared_artifact_id: PreparedArtifactId::new(opaque(24)),
        source_identity: SourceIdentity::new(vec![25, 26]),
        selected_manifest_digest: SelectedManifestDigest::new(opaque(43)),
        policy_id: PolicyId::new(opaque(44)),
        id_map_digest: IdMapDigest::new(opaque(45)),
        path: HostPath::new(b"artifact".to_vec()),
        device: DeviceId::new(46),
        inode: InodeId::new(47),
        state: ArtifactState::Available,
    };
    let mut seen = HashSet::new();

    verify_shape(metadata.clone(), &mut seen);
    verify_shape(domain.clone(), &mut seen);
    verify_shape(generation.clone(), &mut seen);
    verify_shape(participant.clone(), &mut seen);
    verify_shape(ParticipantState::ReleaseIntent, &mut seen);
    verify_shape(
        ParticipantOwnership::Network(Box::new(network_ownership())),
        &mut seen,
    );
    verify_shape(session.clone(), &mut seen);
    verify_shape(terminal.clone(), &mut seen);
    verify_shape(SessionTerminalState::ReleaseIntent, &mut seen);
    verify_shape(mailbox.clone(), &mut seen);
    verify_shape(terminal.pty_master().unwrap(), &mut seen);
    let pidfs_identity = process(48);
    let staged = StagedProcessOwnership::v1(
        domain_id(),
        generation_id(),
        coordinator_id(),
        HostProcessOwnershipSlot::SandboxRuntimeSupervisor,
        pidfs_identity,
    );
    verify_shape(pidfs_identity.authority(), &mut seen);
    verify_shape(pidfs_identity, &mut seen);
    verify_shape(
        LegacyUnverifiedProcessIdentityV1::new(
            HostPidLocator::new(48),
            ProcessStartTimeTicks::new(480),
        ),
        &mut seen,
    );
    verify_shape(HostProcessIdentity::pidfs_v1(pidfs_identity), &mut seen);
    verify_shape(
        PersistedHostProcessOwnership::staged(pidfs_identity),
        &mut seen,
    );
    verify_shape(
        HostProcessOwnershipSlot::SessionTerminalExecClient(session_id()),
        &mut seen,
    );
    verify_shape(ProcessOwnershipMessageVersion::V1, &mut seen);
    verify_shape(staged, &mut seen);
    verify_shape(ProcessOwnershipCommitAck::v1(staged), &mut seen);
    verify_shape(terminal.guest_process().unwrap(), &mut seen);
    verify_shape(
        SessionTerminalEvent::Input(TerminalInput {
            session_id: session_id(),
            bytes: vec![49],
        }),
        &mut seen,
    );
    verify_shape(
        TerminalInput {
            session_id: session_id(),
            bytes: vec![50],
        },
        &mut seen,
    );
    verify_shape(
        TerminalResize {
            session_id: session_id(),
            rows: NonZeroU16::new(24).unwrap(),
            columns: NonZeroU16::new(80).unwrap(),
            pixel_width: 640,
            pixel_height: 480,
        },
        &mut seen,
    );
    verify_shape(
        TerminalSignal {
            session_id: session_id(),
            signal: TerminalSignalKind::Suspend,
        },
        &mut seen,
    );
    verify_shape(TerminalSignalKind::Interrupt, &mut seen);
    verify_shape(
        SessionClose {
            session_id: session_id(),
        },
        &mut seen,
    );
    verify_shape(
        SessionTerminalMessage::Output(TerminalOutput {
            session_id: session_id(),
            bytes: vec![51],
        }),
        &mut seen,
    );
    verify_shape(
        TerminalOutput {
            session_id: session_id(),
            bytes: vec![52],
        },
        &mut seen,
    );
    verify_shape(SessionExit::exited(session_id(), 3), &mut seen);
    verify_shape(SessionExitOutcome::RuntimeLost, &mut seen);
    verify_shape(generation_identity(), &mut seen);
    verify_shape(trust_manifest(), &mut seen);
    verify_shape(tls_credential.clone(), &mut seen);
    verify_shape(
        ProtocolPublicationIdentity {
            protocol_version: ProtocolVersion::V1,
            generation_identity: generation_identity(),
            tls_ca_certificate_der: vec![53],
            trust_manifest_sha256: TrustManifestDigest::new(opaque(54)),
            coordinator_tls_certificate_sha256: CertificateDigest::new(opaque(55)),
        },
        &mut seen,
    );
    verify_shape(
        SandboxCredentialBundle::new(
            generation_identity(),
            trust_manifest(),
            protocol_credential(ProtocolCredentialRole::SessionCoordinator),
            protocol_credential(ProtocolCredentialRole::AgentSandbox),
            vec![56; 32],
        )
        .unwrap(),
        &mut seen,
    );
    verify_shape(TlsRole::ClientCli, &mut seen);
    verify_shape(
        protocol_credential(ProtocolCredentialRole::AgentSandbox),
        &mut seen,
    );
    verify_shape(ProtocolCredentialRole::SessionCoordinator, &mut seen);
    verify_shape(session_binding(), &mut seen);
    verify_shape(session_capability(), &mut seen);
    verify_shape(
        SignedProtocolRecord {
            payload_bytes: vec![57],
            signature: vec![58; 64],
        },
        &mut seen,
    );
    verify_shape(request.clone(), &mut seen);
    verify_shape(
        MailboxCancellation {
            protocol_version: ProtocolVersion::V1,
            session_binding: session_binding(),
            request_id: RequestId::new(opaque(42)),
            session_capability: session_capability(),
            sandbox_credential_id: ProtocolCredentialId::new(opaque(17)),
        },
        &mut seen,
    );
    verify_shape(
        MailboxResultAck {
            protocol_version: ProtocolVersion::V1,
            session_binding: session_binding(),
            request_id: RequestId::new(opaque(42)),
            request_sha256: RequestDigest::new(opaque(59)),
            session_capability: session_capability(),
            sandbox_credential_id: ProtocolCredentialId::new(opaque(17)),
            highest_contiguous_sequence: 4,
        },
        &mut seen,
    );
    verify_shape(
        DeveloperDecision {
            protocol_version: ProtocolVersion::V1,
            session_binding: session_binding(),
            request_id: RequestId::new(opaque(42)),
            request_sha256: RequestDigest::new(opaque(59)),
            decision: DeveloperDecisionKind::Approve,
            approved_command_argv: request.command_argv.clone(),
        },
        &mut seen,
    );
    verify_shape(DeveloperDecisionKind::Deny, &mut seen);
    verify_shape(
        MailboxResult {
            protocol_version: ProtocolVersion::V1,
            session_binding: session_binding(),
            request_id: RequestId::new(opaque(42)),
            request_sha256: RequestDigest::new(opaque(59)),
            coordinator_credential_id: ProtocolCredentialId::new(opaque(16)),
            outcome: EscalationOutcome::Completed,
            exit_code: Some(0),
            chunk_count: 2,
            stdout_sha256: StreamDigest::new(opaque(60)),
            stderr_sha256: StreamDigest::new(opaque(61)),
        },
        &mut seen,
    );
    verify_shape(
        MailboxResultChunk {
            protocol_version: ProtocolVersion::V1,
            session_binding: session_binding(),
            request_id: RequestId::new(opaque(42)),
            request_sha256: RequestDigest::new(opaque(59)),
            coordinator_credential_id: ProtocolCredentialId::new(opaque(16)),
            sequence: 1,
            stream: ResultStream::Stderr,
            bytes: vec![62],
        },
        &mut seen,
    );
    verify_shape(ResultStream::Stdout, &mut seen);
    verify_shape(EscalationOutcome::TimedOut, &mut seen);
    verify_shape(namespace(), &mut seen);
    verify_shape(mount(), &mut seen);
    verify_shape(prepared_ownership(), &mut seen);
    verify_shape(
        StateAnchorOwnership {
            lower_artifact_id: Some(PreparedArtifactId::new(opaque(24))),
            upper_path: Some(HostPath::new(b"upper".to_vec())),
            work_path: Some(HostPath::new(b"work".to_vec())),
            merged_path: Some(HostPath::new(b"merged".to_vec())),
            daemon: owned_process(63),
            mount: Some(mount()),
        },
        &mut seen,
    );
    verify_shape(
        GitRoutingOwnership {
            mountpoint: Some(HostPath::new(b"workspace".to_vec())),
            host_root: Some(HostPath::new(b"host".to_vec())),
            isolated_root: Some(HostPath::new(b"isolated".to_vec())),
            daemon: owned_process(64),
            mount: Some(mount()),
        },
        &mut seen,
    );
    verify_shape(network_ownership(), &mut seen);
    verify_shape(
        HelperOwnership {
            network_sidecar_id: Some(sidecar_id()),
            connectivity_helper: owned_process(65),
            runtime_supervisor: owned_process(21),
        },
        &mut seen,
    );
    verify_shape(
        SandboxOwnership {
            container_id: Some(ContainerId::new(opaque(29))),
            runtime_root: Some(HostPath::new(b"runtime".to_vec())),
            bundle_digest: Some(BundleDigest::new(opaque(66))),
            host_coordinator: owned_process(30),
            runtime_supervisor: owned_process(21),
            attached_runsc: owned_process(67),
        },
        &mut seen,
    );
    verify_shape(
        PublicationOwnership {
            socket_path: Some(HostPath::new(b"socket".to_vec())),
            identity_path: Some(HostPath::new(b"identity".to_vec())),
        },
        &mut seen,
    );
    verify_shape(prepared_artifact.clone(), &mut seen);
    verify_shape(ArtifactState::Reclaiming, &mut seen);
    verify_shape(
        ArtifactLease {
            prepared_artifact_id: PreparedArtifactId::new(opaque(24)),
            domain_id: domain_id(),
            generation_id: generation_id(),
        },
        &mut seen,
    );
    verify_shape(
        PrepareOperation {
            operation_id: PrepareOperationId::new(opaque(68)),
            expected_artifact_id: PreparedArtifactId::new(opaque(24)),
            source_identity: SourceIdentity::new(vec![25, 26]),
            stage_path: HostPath::new(b"stage".to_vec()),
            device: DeviceId::new(69),
            inode: InodeId::new(70),
        },
        &mut seen,
    );
    verify_shape(
        AddressFamilyDisposition {
            family: AddressFamily::Ipv6,
            disposition: AddressFamilyDispositionValue::Disabled,
        },
        &mut seen,
    );
    verify_shape(
        DefaultAction {
            family: AddressFamily::Ipv4,
            action: PolicyAction::Allow,
        },
        &mut seen,
    );
    verify_shape(
        PrefixRule {
            family: AddressFamily::Ipv4,
            network: IpNetwork {
                address: IpAddress::V4([192, 168, 0, 0]),
                prefix_length: 16,
            },
            action: PolicyAction::Deny,
        },
        &mut seen,
    );
    verify_shape(
        Resolver {
            family: AddressFamily::Ipv4,
            address: IpAddress::V4([9, 9, 9, 9]),
        },
        &mut seen,
    );
    verify_shape(AddressFamily::Ipv4, &mut seen);
    verify_shape(PolicyAction::Deny, &mut seen);
    verify_shape(
        DerivedNetworkPolicy::v1(DerivedNetworkPolicyFields {
            address_families: vec![AddressFamilyDisposition {
                family: AddressFamily::Ipv4,
                disposition: AddressFamilyDispositionValue::Enabled,
            }],
            default_actions: vec![DefaultAction {
                family: AddressFamily::Ipv4,
                action: PolicyAction::Allow,
            }],
            prefix_rules: vec![],
            resolvers: vec![Resolver {
                family: AddressFamily::Ipv4,
                address: IpAddress::V4([9, 9, 9, 9]),
            }],
        }),
        &mut seen,
    );
    verify_shape(readiness_evidence(), &mut seen);
    verify_shape(readiness_attestation(), &mut seen);
    verify_shape(network_observation(), &mut seen);
    verify_shape(PolicyDelivery::SealedMemfd, &mut seen);
    verify_shape(readiness_identity(), &mut seen);
    verify_shape(reachability_probe(), &mut seen);
    verify_shape(
        DeniedDestinationObservation {
            destination: IpAddress::V4([10, 0, 0, 1]),
            unreachable: true,
        },
        &mut seen,
    );
    verify_shape(
        PermittedDestinationObservation {
            destination: IpAddress::V4([1, 0, 0, 1]),
            reachable: true,
        },
        &mut seen,
    );
    verify_shape(
        DnsObservation {
            resolver: IpAddress::V4([9, 9, 9, 9]),
            query_name: "example.invalid".into(),
            resolved_addresses: vec![IpAddress::V4([203, 0, 113, 1])],
        },
        &mut seen,
    );
    verify_shape(
        DomainStore {
            metadata: DomainStoreMetadata::v2(),
            domain,
            generations: vec![generation],
            participants: vec![participant],
            sessions: vec![session],
            session_terminals: vec![terminal],
            mailboxes: vec![mailbox],
        },
        &mut seen,
    );
    verify_shape(
        StorageCatalog {
            metadata: StorageCatalogMetadata::v1(),
            prepared_artifacts: vec![prepared_artifact],
            artifact_leases: vec![],
            prepare_operations: vec![],
        },
        &mut seen,
    );

    assert_eq!(seen.len(), IMPLEMENTED_SERIALIZED_SCHEMA_TYPE_NAMES.len());
    assert!(
        IMPLEMENTED_SERIALIZED_SCHEMA_TYPE_NAMES
            .iter()
            .all(|name| seen.contains(name))
    );
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum SchemaCategory {
    Serialized,
    Runtime,
}

fn parse_schema_categories(
    schema: &str,
    serialized_registry: &HashSet<&str>,
    runtime_registry: &HashSet<&str>,
) -> Result<HashMap<String, SchemaCategory>, String> {
    if let Some(name) = serialized_registry.intersection(runtime_registry).next() {
        return Err(format!("schema registries overlap at {name}"));
    }

    let mut documented = HashMap::new();
    let mut current: Option<(&str, Option<SchemaCategory>)> = None;

    for line in schema.lines() {
        let trimmed = line.trim();
        if let Some(declaration) = trimmed.strip_prefix("class ") {
            if current.is_some() {
                return Err("nested class declaration".into());
            }
            let name = declaration
                .split(|character: char| character.is_whitespace() || character == '{')
                .next()
                .filter(|name| !name.is_empty())
                .ok_or_else(|| "class declaration has no name".to_owned())?;
            if !declaration.contains('{') {
                return Err(format!("class {name} has no stereotype block"));
            }
            if !serialized_registry.contains(name) && !runtime_registry.contains(name) {
                return Err(format!("unknown schema class {name}"));
            }
            current = Some((name, None));
            continue;
        }

        let Some((name, category)) = current.as_mut() else {
            continue;
        };
        if trimmed == "}" {
            let category = category.ok_or_else(|| format!("class {name} is unmarked"))?;
            match documented.insert((*name).to_owned(), category) {
                Some(existing) if existing != category => {
                    return Err(format!("class {name} has conflicting categories"));
                }
                _ => {}
            }
            current = None;
            continue;
        }
        let Some(stereotype) = trimmed
            .strip_prefix("<<")
            .and_then(|value| value.strip_suffix(">>"))
        else {
            continue;
        };
        let parsed = match stereotype {
            "serialized" => Some(SchemaCategory::Serialized),
            "runtime" => Some(SchemaCategory::Runtime),
            "enumeration" | "union" => None,
            _ => return Err(format!("class {name} has unknown stereotype {stereotype}")),
        };
        if let Some(parsed) = parsed
            && category.replace(parsed).is_some()
        {
            return Err(format!("class {name} has multiple category markers"));
        }
    }

    if let Some((name, _)) = current {
        return Err(format!("class {name} has no closing brace"));
    }
    for name in serialized_registry.union(runtime_registry) {
        if !documented.contains_key(*name) {
            return Err(format!("registered schema class {name} is undocumented"));
        }
    }
    Ok(documented)
}

#[test]
fn schema_categories_match_exact_disjoint_registries() {
    let serialized = DECLARED_SERIALIZED_SCHEMA_TYPE_NAMES
        .iter()
        .copied()
        .collect::<HashSet<_>>();
    let runtime = DECLARED_RUNTIME_SCHEMA_TYPE_NAMES
        .iter()
        .copied()
        .collect::<HashSet<_>>();
    assert_eq!(
        serialized.len(),
        DECLARED_SERIALIZED_SCHEMA_TYPE_NAMES.len()
    );
    assert_eq!(runtime.len(), DECLARED_RUNTIME_SCHEMA_TYPE_NAMES.len());

    let documented = parse_schema_categories(
        include_str!("../../../docs/schema.md"),
        &serialized,
        &runtime,
    )
    .unwrap();

    let documented_serialized = documented
        .iter()
        .filter_map(|(name, category)| {
            (*category == SchemaCategory::Serialized).then_some(name.as_str())
        })
        .collect::<HashSet<_>>();
    let documented_runtime = documented
        .iter()
        .filter_map(|(name, category)| {
            (*category == SchemaCategory::Runtime).then_some(name.as_str())
        })
        .collect::<HashSet<_>>();

    assert_eq!(documented_serialized, serialized);
    assert_eq!(documented_runtime, runtime);
}

#[test]
fn schema_category_parser_fails_closed() {
    let serialized = HashSet::from(["Known"]);
    let runtime = HashSet::from(["Runtime"]);
    let overlapping = HashSet::from(["Known"]);

    for invalid in [
        "class Known {\nfield\n}",
        "class Known {\n<<serialized>>\n<<serialized>>\n}",
        "class Known {\n<<serialized>>\n<<runtime>>\n}",
        "class Unknown {\n<<serialized>>\n}",
        "class Known {\n<<unexpected>>\n}",
    ] {
        assert!(parse_schema_categories(invalid, &serialized, &runtime).is_err());
    }
    assert!(parse_schema_categories("", &serialized, &runtime).is_err());
    assert!(parse_schema_categories("", &serialized, &overlapping).is_err());

    assert!(
        parse_schema_categories(
            "class Known {\n<<serialized>>\n}\nclass Runtime {\n<<runtime>>\n}",
            &serialized,
            &runtime,
        )
        .is_ok()
    );
}

fn assert_unknown_protocol_version<T>(record: &T)
where
    T: DeserializeOwned + Serialize,
{
    let mut encoded = encode(record).unwrap();
    encoded[0] = u8::MAX;
    assert!(matches!(
        decode::<T>(&encoded),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
fn trailing_and_nonminimal_postcard_encodings_are_rejected() {
    assert!(matches!(
        decode::<ParticipantState>(&[0, 0]),
        Err(CodecError::TrailingPayload)
    ));
    assert!(matches!(
        decode::<ParticipantState>(&[0x80, 0]),
        Err(CodecError::NonMinimalPayload)
    ));

    let input = TerminalInput {
        session_id: session_id(),
        bytes: vec![1],
    };
    let mut overlong_length = encode(&input).unwrap();
    overlong_length[32] = 0x81;
    overlong_length.insert(33, 0);
    assert!(matches!(
        decode::<TerminalInput>(&overlong_length),
        Err(CodecError::NonMinimalPayload)
    ));

    let mut overlong_protocol_version = encode(&trust_manifest()).unwrap();
    overlong_protocol_version[0] = 0x80;
    overlong_protocol_version.insert(1, 0);
    assert!(matches!(
        decode::<TrustManifest>(&overlong_protocol_version),
        Err(CodecError::NonMinimalPayload)
    ));

    let mut overlong_store_id = encode(&StoreMetadata::domain_store_v2()).unwrap();
    overlong_store_id[0] |= 0x80;
    overlong_store_id.insert(1, 0);
    assert!(matches!(
        decode::<StoreMetadata>(&overlong_store_id),
        Err(CodecError::NonMinimalPayload)
    ));
}

#[test]
fn unknown_enums_and_protocol_versions_are_rejected() {
    assert!(matches!(
        decode::<ParticipantState>(&[u8::MAX]),
        Err(CodecError::MalformedPayload(_))
    ));

    assert_unknown_protocol_version(&trust_manifest());
    assert_unknown_protocol_version(&ProtocolPublicationIdentity {
        protocol_version: ProtocolVersion::V1,
        generation_identity: generation_identity(),
        tls_ca_certificate_der: vec![1],
        trust_manifest_sha256: TrustManifestDigest::new(opaque(2)),
        coordinator_tls_certificate_sha256: CertificateDigest::new(opaque(3)),
    });
    assert_unknown_protocol_version(
        &SandboxCredentialBundle::new(
            generation_identity(),
            trust_manifest(),
            protocol_credential(ProtocolCredentialRole::SessionCoordinator),
            protocol_credential(ProtocolCredentialRole::AgentSandbox),
            vec![4; 32],
        )
        .unwrap(),
    );
    assert_unknown_protocol_version(&protocol_credential(
        ProtocolCredentialRole::SessionCoordinator,
    ));
    assert_unknown_protocol_version(&session_capability());

    let request = MailboxRequest {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: RequestId::new(opaque(5)),
        session_capability: session_capability(),
        sandbox_credential_id: ProtocolCredentialId::new(opaque(6)),
        command_argv: vec![b"true".to_vec()],
        justification: "test".into(),
    };
    assert_unknown_protocol_version(&request);
    assert_unknown_protocol_version(&MailboxCancellation {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: request.request_id,
        session_capability: session_capability(),
        sandbox_credential_id: ProtocolCredentialId::new(opaque(6)),
    });
    assert_unknown_protocol_version(&MailboxResultAck {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: request.request_id,
        request_sha256: RequestDigest::new(opaque(7)),
        session_capability: session_capability(),
        sandbox_credential_id: ProtocolCredentialId::new(opaque(6)),
        highest_contiguous_sequence: 0,
    });
    assert_unknown_protocol_version(&DeveloperDecision {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: request.request_id,
        request_sha256: RequestDigest::new(opaque(7)),
        decision: DeveloperDecisionKind::Deny,
        approved_command_argv: vec![],
    });
    assert_unknown_protocol_version(&MailboxResult {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: request.request_id,
        request_sha256: RequestDigest::new(opaque(7)),
        coordinator_credential_id: ProtocolCredentialId::new(opaque(8)),
        outcome: EscalationOutcome::Denied,
        exit_code: None,
        chunk_count: 0,
        stdout_sha256: StreamDigest::new(opaque(9)),
        stderr_sha256: StreamDigest::new(opaque(10)),
    });
    assert_unknown_protocol_version(&MailboxResultChunk {
        protocol_version: ProtocolVersion::V1,
        session_binding: session_binding(),
        request_id: request.request_id,
        request_sha256: RequestDigest::new(opaque(7)),
        coordinator_credential_id: ProtocolCredentialId::new(opaque(8)),
        sequence: 0,
        stream: ResultStream::Stdout,
        bytes: vec![],
    });

    let staged = StagedProcessOwnership::v1(
        domain_id(),
        generation_id(),
        coordinator_id(),
        HostProcessOwnershipSlot::SandboxAttachedRunsc,
        process(90),
    );
    assert_unknown_protocol_version(&staged);
    assert_unknown_protocol_version(&ProcessOwnershipCommitAck::v1(staged));
    assert!(matches!(
        decode::<HostProcessIdentity>(&[u8::MAX]),
        Err(CodecError::MalformedPayload(_))
    ));
    assert!(matches!(
        decode::<PersistedHostProcessOwnership>(&[u8::MAX]),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
fn pidfs_authority_is_separate_from_locator_and_negative_evidence() {
    let maximum = PidfsProcessIdentityV1::new(
        PidfsProcessAuthorityV1 {
            boot_id: BootId::new([u8::MAX; 16]),
            pidfs_st_dev: PidfsDeviceId::new(u64::MAX),
            pidfs_st_ino: PidfsInodeId::new(u64::MAX),
        },
        HostPidLocator::new(u32::MAX),
        ProcessStartTimeTicks::new(u64::MAX),
    );
    assert_eq!(
        decode::<PidfsProcessIdentityV1>(&encode(&maximum).unwrap()).unwrap(),
        maximum
    );
    assert_eq!(maximum.authority().pidfs_st_dev.get(), u64::MAX);
    assert_eq!(maximum.authority().pidfs_st_ino.get(), u64::MAX);

    let changed_locator = PidfsProcessIdentityV1::new(
        maximum.authority(),
        HostPidLocator::new(1),
        ProcessStartTimeTicks::new(2),
    );
    assert!(maximum.has_same_authority(&changed_locator));
    assert_ne!(maximum, changed_locator);

    for unequal_authority in [
        PidfsProcessAuthorityV1 {
            boot_id: BootId::new([0; 16]),
            ..maximum.authority()
        },
        PidfsProcessAuthorityV1 {
            pidfs_st_dev: PidfsDeviceId::new(u64::MAX - 1),
            ..maximum.authority()
        },
        PidfsProcessAuthorityV1 {
            pidfs_st_ino: PidfsInodeId::new(u64::MAX - 1),
            ..maximum.authority()
        },
    ] {
        let unequal = PidfsProcessIdentityV1::new(
            unequal_authority,
            maximum.host_pid(),
            maximum.start_time_ticks(),
        );
        assert!(!maximum.has_same_authority(&unequal));
        assert_ne!(
            maximum.authority().cmp(&unequal.authority()),
            std::cmp::Ordering::Equal
        );
    }
}

#[test]
fn process_slot_separates_cleanup_authority_from_serviceability() {
    let current = process(91);
    assert_eq!(
        PersistedHostProcessOwnership::committed(current)
            .serviceable_pidfs()
            .unwrap(),
        &current
    );
    assert_eq!(
        PersistedHostProcessOwnership::staged(current).serviceable_pidfs(),
        Err(ProcessOwnershipRefusal::Staged)
    );
    assert_eq!(
        PersistedHostProcessOwnership::staged(current)
            .authoritative_pidfs()
            .unwrap(),
        &current
    );
    assert_eq!(
        PersistedHostProcessOwnership::unassigned().authoritative_pidfs(),
        Err(ProcessAuthorityRefusal::Unassigned)
    );

    let legacy = LegacyUnverifiedProcessIdentityV1::new(
        HostPidLocator::new(91),
        ProcessStartTimeTicks::new(910),
    );
    assert_eq!(
        HostProcessIdentity::legacy_unverified_v1(legacy).verified_pidfs(),
        Err(LegacyProcessIdentityRefusal)
    );
    assert_eq!(
        PersistedHostProcessOwnership::legacy_unverified(legacy).serviceable_pidfs(),
        Err(ProcessOwnershipRefusal::LegacyUnverified)
    );
    assert_eq!(
        PersistedHostProcessOwnership::legacy_unverified(legacy).authoritative_pidfs(),
        Err(ProcessAuthorityRefusal::LegacyUnverified)
    );

    let staged_anchor = ParticipantOwnership::StateAnchor(state_anchor_ownership(
        PersistedHostProcessOwnership::staged(current),
        Some(mount()),
    ));
    assert_eq!(
        Participant::snapshot(
            generation_id(),
            ParticipantRole::StateAnchor,
            ParticipantState::Prepared,
            Some(staged_anchor.clone()),
        ),
        Err(RecordInvariantError::ParticipantSnapshotState)
    );
    assert!(matches!(
        decode::<Participant>(
            &encode(&RawParticipant {
                generation_id: generation_id(),
                role: ParticipantRole::StateAnchor,
                state: ParticipantState::Prepared,
                ownership: Some(staged_anchor),
            })
            .unwrap()
        ),
        Err(CodecError::MalformedPayload(_))
    ));

    assert_eq!(
        SessionTerminal::snapshot(SessionTerminalFields {
            generation_id: generation_id(),
            session_id: session_id(),
            state: SessionTerminalState::Running,
            container_id: Some(ContainerId::new(opaque(9))),
            host_coordinator: owned_process(10),
            pty_master: Some(pty()),
            exec_client: PersistedHostProcessOwnership::staged(current),
            guest_process: Some(guest_process()),
        }),
        Err(RecordInvariantError::SessionTerminalSnapshotState)
    );
    assert!(matches!(
        decode::<SessionTerminal>(
            &encode(&RawSessionTerminal {
                generation_id: generation_id(),
                session_id: session_id(),
                state: SessionTerminalState::Running,
                container_id: Some(ContainerId::new(opaque(9))),
                host_coordinator: owned_process(10),
                pty_master: Some(pty()),
                exec_client: PersistedHostProcessOwnership::staged(current),
                guest_process: Some(guest_process()),
            })
            .unwrap()
        ),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
#[allow(clippy::too_many_lines)]
fn participant_snapshot_matrix_is_phase_complete_and_strictly_decoded() {
    let roles = [
        ParticipantRole::PreparedLower,
        ParticipantRole::StateAnchor,
        ParticipantRole::GitRouting,
        ParticipantRole::Network,
        ParticipantRole::Helper,
        ParticipantRole::Sandbox,
        ParticipantRole::Publication,
    ];
    let active_states = [
        ParticipantState::LeaseHeld,
        ParticipantState::Prepared,
        ParticipantState::Committed,
    ];

    for role in roles {
        assert_eq!(
            Participant::bare_intent(generation_id(), role).ownership(),
            None
        );
        assert!(
            Participant::snapshot(
                generation_id(),
                role,
                ParticipantState::Intent,
                Some(complete_ownership(role, ParticipantState::Intent)),
            )
            .is_ok()
        );
        assert!(
            Participant::snapshot(
                generation_id(),
                role,
                ParticipantState::ReleaseIntent,
                Some(release_ownership(role)),
            )
            .is_ok()
        );
        assert!(
            Participant::snapshot(
                generation_id(),
                role,
                ParticipantState::Clean,
                Some(ParticipantOwnership::empty(role)),
            )
            .is_ok()
        );
        assert_eq!(
            Participant::snapshot(
                generation_id(),
                role,
                ParticipantState::Intent,
                Some(ParticipantOwnership::empty(role)),
            ),
            Err(RecordInvariantError::ParticipantSnapshotState)
        );

        for state in active_states {
            let candidate = Participant::snapshot(
                generation_id(),
                role,
                state,
                Some(complete_ownership(role, state)),
            );
            assert_eq!(candidate.is_ok(), state == active_state(role));
        }
        for state in [
            ParticipantState::LeaseHeld,
            ParticipantState::Prepared,
            ParticipantState::Committed,
            ParticipantState::ReleaseIntent,
            ParticipantState::Clean,
        ] {
            assert_eq!(
                Participant::snapshot(generation_id(), role, state, None),
                Err(RecordInvariantError::ParticipantSnapshotState)
            );
        }
    }

    for raw in [
        RawParticipant {
            generation_id: generation_id(),
            role: ParticipantRole::StateAnchor,
            state: ParticipantState::Intent,
            ownership: Some(ParticipantOwnership::empty(ParticipantRole::StateAnchor)),
        },
        RawParticipant {
            generation_id: generation_id(),
            role: ParticipantRole::StateAnchor,
            state: ParticipantState::Prepared,
            ownership: None,
        },
        RawParticipant {
            generation_id: generation_id(),
            role: ParticipantRole::Network,
            state: ParticipantState::Prepared,
            ownership: Some(complete_ownership(
                ParticipantRole::StateAnchor,
                ParticipantState::Prepared,
            )),
        },
    ] {
        assert!(matches!(
            decode::<Participant>(&encode(&raw).unwrap()),
            Err(CodecError::MalformedPayload(_))
        ));
    }
}

#[test]
fn active_process_owners_require_complete_current_pidfs_evidence() {
    let current = process(94);
    let legacy = LegacyUnverifiedProcessIdentityV1::new(
        HostPidLocator::new(94),
        ProcessStartTimeTicks::new(940),
    );
    let invalid_slots = [
        PersistedHostProcessOwnership::unassigned(),
        PersistedHostProcessOwnership::staged(current),
        PersistedHostProcessOwnership::legacy_unverified(legacy),
    ];
    for daemon in invalid_slots {
        assert_eq!(
            Participant::snapshot(
                generation_id(),
                ParticipantRole::StateAnchor,
                ParticipantState::Prepared,
                Some(ParticipantOwnership::StateAnchor(state_anchor_ownership(
                    daemon,
                    Some(mount()),
                ))),
            ),
            Err(RecordInvariantError::ParticipantSnapshotState)
        );
    }
    assert_eq!(
        Participant::snapshot(
            generation_id(),
            ParticipantRole::StateAnchor,
            ParticipantState::Prepared,
            Some(ParticipantOwnership::StateAnchor(state_anchor_ownership(
                owned_process(94),
                None,
            ))),
        ),
        Err(RecordInvariantError::ParticipantSnapshotState)
    );

    let mut inconsistent_network = network_ownership();
    inconsistent_network.runtime_supervisor = owned_process(95);
    assert_eq!(
        Participant::snapshot(
            generation_id(),
            ParticipantRole::Network,
            ParticipantState::Committed,
            Some(ParticipantOwnership::Network(Box::new(
                inconsistent_network
            ))),
        ),
        Err(RecordInvariantError::ParticipantSnapshotState)
    );
    assert_eq!(
        Participant::snapshot(
            GenerationId::new(generation_id().get() + 1),
            ParticipantRole::Network,
            ParticipantState::Committed,
            Some(ParticipantOwnership::Network(Box::new(network_ownership()))),
        ),
        Err(RecordInvariantError::ParticipantSnapshotState)
    );
}

#[test]
#[allow(clippy::too_many_lines)]
fn active_network_rejects_every_mismatched_readiness_relation() {
    let other_domain = DomainId::new(opaque(80));
    let other_generation = GenerationId::new(81);
    let other_sidecar = NetworkSidecarId::new(opaque(82));
    let other_configuration = ConfigurationFingerprint::new(opaque(83));
    let other_policy = DerivedPolicyDigest::new(opaque(84));
    let other_fingerprint = ReadinessFingerprint::new(opaque(85));
    let other_namespace = NamespaceIdentity {
        device: DeviceId::new(86),
        inode: InodeId::new(87),
    };
    let mismatches = [
        (
            "enclosing generation",
            other_generation,
            network_ownership(),
        ),
        (
            "readiness fingerprint",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership.readiness_fingerprint = Some(other_fingerprint);
            }),
        ),
        (
            "attestation domain",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .attestation
                    .domain_id = other_domain;
            }),
        ),
        (
            "readiness identity domain",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .readiness_identity
                    .domain_id = other_domain;
            }),
        ),
        (
            "reachability domain",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .reachability_probe
                    .domain_id = other_domain;
            }),
        ),
        (
            "attestation generation",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .attestation
                    .generation_id = other_generation;
            }),
        ),
        (
            "readiness identity generation",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .readiness_identity
                    .generation_id = other_generation;
            }),
        ),
        (
            "reachability generation",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .reachability_probe
                    .generation_id = other_generation;
            }),
        ),
        (
            "attestation sidecar",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .attestation
                    .network_sidecar_id = other_sidecar;
            }),
        ),
        (
            "readiness identity sidecar",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .readiness_identity
                    .network_sidecar_id = other_sidecar;
            }),
        ),
        (
            "configuration fingerprint",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .readiness_identity
                    .configuration_fingerprint = other_configuration;
            }),
        ),
        (
            "attestation policy",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .attestation
                    .derived_policy_digest = other_policy;
            }),
        ),
        (
            "readiness identity policy",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .readiness_identity
                    .derived_policy_digest = other_policy;
            }),
        ),
        (
            "attestation namespace",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .attestation
                    .enforcement_namespace = other_namespace;
            }),
        ),
        (
            "readiness identity namespace",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .readiness_identity
                    .enforcement_namespace = other_namespace;
            }),
        ),
        (
            "reachability namespace",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership
                    .readiness_evidence
                    .as_mut()
                    .unwrap()
                    .reachability_probe
                    .enforcement_namespace = other_namespace;
            }),
        ),
        (
            "runtime supervisor",
            generation_id(),
            mutate_network_ownership(|ownership| {
                ownership.runtime_supervisor = owned_process(88);
            }),
        ),
    ];

    for (name, enclosing_generation, ownership) in mismatches {
        assert_eq!(
            Participant::snapshot(
                enclosing_generation,
                ParticipantRole::Network,
                ParticipantState::Committed,
                Some(ParticipantOwnership::Network(Box::new(ownership.clone()))),
            ),
            Err(RecordInvariantError::ParticipantSnapshotState),
            "{name}"
        );
        let encoded = encode(&RawParticipant {
            generation_id: enclosing_generation,
            role: ParticipantRole::Network,
            state: ParticipantState::Committed,
            ownership: Some(ParticipantOwnership::Network(Box::new(ownership))),
        })
        .unwrap();
        assert!(
            matches!(
                decode::<Participant>(&encoded),
                Err(CodecError::MalformedPayload(_))
            ),
            "{name}"
        );
    }
}

#[test]
#[allow(clippy::too_many_lines)]
fn participant_transitions_preserve_and_clear_exact_evidence() {
    let bare = Participant::bare_intent(generation_id(), ParticipantRole::StateAnchor);
    let initial_ownership = ParticipantOwnership::StateAnchor(state_anchor_ownership(
        PersistedHostProcessOwnership::unassigned(),
        None,
    ));
    let initial = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(initial_ownership),
    )
    .unwrap();
    assert_eq!(bare.validate_transition_to(&initial), Ok(()));

    let staged_identity = process(96);
    let staged_ownership = ParticipantOwnership::StateAnchor(state_anchor_ownership(
        PersistedHostProcessOwnership::staged(staged_identity),
        None,
    ));
    let staged = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(staged_ownership.clone()),
    )
    .unwrap();
    assert_eq!(
        staged.validate_transition_to(&initial),
        Err(RecordTransitionError::EvidenceLost)
    );
    assert_eq!(initial.validate_transition_to(&staged), Ok(()));

    let committed_ownership = ParticipantOwnership::StateAnchor(state_anchor_ownership(
        PersistedHostProcessOwnership::committed(staged_identity),
        None,
    ));
    let committed = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(committed_ownership),
    )
    .unwrap();
    assert_eq!(staged.validate_transition_to(&committed), Ok(()));

    let mounted_ownership = ParticipantOwnership::StateAnchor(state_anchor_ownership(
        PersistedHostProcessOwnership::committed(staged_identity),
        Some(mount()),
    ));
    let mounted = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(mounted_ownership.clone()),
    )
    .unwrap();
    assert_eq!(committed.validate_transition_to(&mounted), Ok(()));
    let active = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Prepared,
        Some(mounted_ownership.clone()),
    )
    .unwrap();
    assert_eq!(mounted.validate_transition_to(&active), Ok(()));
    assert_eq!(
        bare.validate_transition_to(&active),
        Err(RecordTransitionError::PriorAcquisitionSnapshotRequired)
    );

    let release = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::ReleaseIntent,
        Some(mounted_ownership),
    )
    .unwrap();
    assert_eq!(active.validate_transition_to(&release), Ok(()));
    let clean = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Clean,
        Some(ParticipantOwnership::empty(ParticipantRole::StateAnchor)),
    )
    .unwrap();
    assert_eq!(
        release.validate_transition_to(&clean),
        Err(RecordTransitionError::AbsenceProofRequired)
    );
    assert_eq!(
        release.validate_transition_to_clean_after_absence(&clean),
        Ok(())
    );
    assert!(clean.ownership().unwrap().is_empty());

    let bare_clean = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Clean,
        Some(ParticipantOwnership::empty(ParticipantRole::StateAnchor)),
    )
    .unwrap();
    assert_eq!(
        bare.validate_transition_to_clean_after_absence(&bare_clean),
        Ok(())
    );
    assert_eq!(
        initial.validate_transition_to_clean_after_absence(&bare_clean),
        Err(RecordTransitionError::StateRegression)
    );
}

#[test]
#[allow(clippy::too_many_lines)]
fn newly_recorded_existing_processes_do_not_bypass_child_staging() {
    let existing_edges = [
        ParticipantProcessEdge::NetworkRuntimeSupervisor,
        ParticipantProcessEdge::HelperRuntimeSupervisor,
        ParticipantProcessEdge::SandboxHostCoordinator,
    ];
    for (index, edge) in existing_edges.into_iter().enumerate() {
        let current = process(130 + u32::try_from(index).unwrap());
        let unassigned =
            participant_edge_snapshot(edge, PersistedHostProcessOwnership::unassigned());
        let committed =
            participant_edge_snapshot(edge, PersistedHostProcessOwnership::committed(current));
        let bare = Participant::bare_intent(generation_id(), committed.role());
        assert_eq!(bare.validate_transition_to(&committed), Ok(()), "{edge:?}");
        assert_eq!(
            unassigned.validate_transition_to(&committed),
            Ok(()),
            "{edge:?}"
        );

        let legacy = participant_edge_snapshot(
            edge,
            PersistedHostProcessOwnership::legacy_unverified(
                LegacyUnverifiedProcessIdentityV1::new(
                    current.host_pid(),
                    current.start_time_ticks(),
                ),
            ),
        );
        assert_eq!(
            unassigned.validate_transition_to(&legacy),
            Err(RecordTransitionError::ProcessStageSkipped),
            "{edge:?}"
        );
    }

    let acquired_edges = [
        ParticipantProcessEdge::StateAnchorDaemon,
        ParticipantProcessEdge::GitRoutingDaemon,
        ParticipantProcessEdge::HelperConnectivityHelper,
        ParticipantProcessEdge::SandboxRuntimeSupervisor,
        ParticipantProcessEdge::SandboxAttachedRunsc,
    ];
    for (index, edge) in acquired_edges.into_iter().enumerate() {
        let current = process(140 + u32::try_from(index).unwrap());
        let unassigned =
            participant_edge_snapshot(edge, PersistedHostProcessOwnership::unassigned());
        let staged =
            participant_edge_snapshot(edge, PersistedHostProcessOwnership::staged(current));
        let committed =
            participant_edge_snapshot(edge, PersistedHostProcessOwnership::committed(current));
        assert_eq!(
            unassigned.validate_transition_to(&staged),
            Ok(()),
            "{edge:?}"
        );
        assert_eq!(
            staged.validate_transition_to(&committed),
            Ok(()),
            "{edge:?}"
        );
        assert_eq!(
            unassigned.validate_transition_to(&committed),
            Err(RecordTransitionError::ProcessStageSkipped),
            "{edge:?}"
        );
        let wrong_commit = participant_edge_snapshot(
            edge,
            PersistedHostProcessOwnership::committed(process(150 + u32::try_from(index).unwrap())),
        );
        assert_eq!(
            staged.validate_transition_to(&wrong_commit),
            Err(RecordTransitionError::EvidenceChanged),
            "{edge:?}"
        );
    }

    let terminal_unassigned = terminal_snapshot(
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::unassigned(),
        PersistedHostProcessOwnership::unassigned(),
        false,
    )
    .unwrap();
    let terminal_existing_hc = terminal_snapshot(
        SessionTerminalState::Intent,
        owned_process(101),
        PersistedHostProcessOwnership::unassigned(),
        false,
    )
    .unwrap();
    assert_eq!(
        terminal_unassigned.validate_transition_to(&terminal_existing_hc),
        Ok(())
    );
    let terminal_legacy_hc = terminal_snapshot(
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::legacy_unverified(LegacyUnverifiedProcessIdentityV1::new(
            HostPidLocator::new(101),
            ProcessStartTimeTicks::new(1_010),
        )),
        PersistedHostProcessOwnership::unassigned(),
        false,
    )
    .unwrap();
    assert_eq!(
        terminal_unassigned.validate_transition_to(&terminal_legacy_hc),
        Err(RecordTransitionError::ProcessStageSkipped)
    );

    let terminal_exec_staged = terminal_snapshot(
        SessionTerminalState::Intent,
        owned_process(101),
        PersistedHostProcessOwnership::staged(process(102)),
        false,
    )
    .unwrap();
    let terminal_exec_committed = terminal_snapshot(
        SessionTerminalState::Intent,
        owned_process(101),
        owned_process(102),
        false,
    )
    .unwrap();
    assert_eq!(
        terminal_existing_hc.validate_transition_to(&terminal_exec_staged),
        Ok(())
    );
    assert_eq!(
        terminal_exec_staged.validate_transition_to(&terminal_exec_committed),
        Ok(())
    );
    assert_eq!(
        terminal_existing_hc.validate_transition_to(&terminal_exec_committed),
        Err(RecordTransitionError::ProcessStageSkipped)
    );

    let terminal_wrong_exec = terminal_snapshot(
        SessionTerminalState::Intent,
        owned_process(101),
        owned_process(103),
        false,
    )
    .unwrap();
    assert_eq!(
        terminal_exec_staged.validate_transition_to(&terminal_wrong_exec),
        Err(RecordTransitionError::EvidenceChanged)
    );
}

#[test]
#[allow(clippy::too_many_lines)]
fn participant_transitions_reject_field_loss_replacement_and_stage_skips() {
    let current = process(97);
    let initial_ownership =
        state_anchor_ownership(PersistedHostProcessOwnership::unassigned(), None);
    let initial = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(ParticipantOwnership::StateAnchor(initial_ownership.clone())),
    )
    .unwrap();

    let direct_commit = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(ParticipantOwnership::StateAnchor(state_anchor_ownership(
            PersistedHostProcessOwnership::committed(current),
            None,
        ))),
    )
    .unwrap();
    assert_eq!(
        initial.validate_transition_to(&direct_commit),
        Err(RecordTransitionError::ProcessStageSkipped)
    );
    let staged = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(ParticipantOwnership::StateAnchor(state_anchor_ownership(
            PersistedHostProcessOwnership::staged(current),
            None,
        ))),
    )
    .unwrap();
    assert_eq!(
        direct_commit.validate_transition_to(&staged),
        Err(RecordTransitionError::EvidenceLost)
    );
    let same_authority_different_locator = PidfsProcessIdentityV1::new(
        current.authority(),
        HostPidLocator::new(current.host_pid().get() + 1),
        current.start_time_ticks(),
    );
    let replaced = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(ParticipantOwnership::StateAnchor(state_anchor_ownership(
            PersistedHostProcessOwnership::committed(same_authority_different_locator),
            None,
        ))),
    )
    .unwrap();
    assert!(current.has_same_authority(&same_authority_different_locator));
    assert_eq!(
        staged.validate_transition_to(&replaced),
        Err(RecordTransitionError::EvidenceChanged)
    );

    let mut losses = Vec::new();
    let mut lost = initial_ownership.clone();
    lost.lower_artifact_id = None;
    losses.push(lost);
    let mut lost = initial_ownership.clone();
    lost.upper_path = None;
    losses.push(lost);
    let mut lost = initial_ownership.clone();
    lost.work_path = None;
    losses.push(lost);
    let mut lost = initial_ownership;
    lost.merged_path = None;
    losses.push(lost);
    for lost in losses {
        let lost = Participant::snapshot(
            generation_id(),
            ParticipantRole::StateAnchor,
            ParticipantState::Intent,
            Some(ParticipantOwnership::StateAnchor(lost)),
        )
        .unwrap();
        assert_eq!(
            initial.validate_transition_to(&lost),
            Err(RecordTransitionError::EvidenceLost)
        );
    }

    let partial_release = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::ReleaseIntent,
        initial.ownership().cloned(),
    )
    .unwrap();
    assert_eq!(initial.validate_transition_to(&partial_release), Ok(()));
    assert_eq!(partial_release.ownership(), initial.ownership());

    let staged_release = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::ReleaseIntent,
        staged.ownership().cloned(),
    )
    .unwrap();
    assert_eq!(staged.validate_transition_to(&staged_release), Ok(()));
    let ParticipantOwnership::StateAnchor(staged_release_ownership) =
        staged_release.ownership().unwrap()
    else {
        panic!("wrong ownership role")
    };
    assert_eq!(
        staged_release_ownership
            .daemon
            .authoritative_pidfs()
            .unwrap(),
        &current
    );
}

#[test]
#[allow(clippy::too_many_lines)]
fn participant_and_terminal_transition_tables_reject_every_regression() {
    let intent = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Intent,
        Some(complete_ownership(
            ParticipantRole::StateAnchor,
            ParticipantState::Intent,
        )),
    )
    .unwrap();
    let active = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Prepared,
        Some(complete_ownership(
            ParticipantRole::StateAnchor,
            ParticipantState::Prepared,
        )),
    )
    .unwrap();
    let release = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::ReleaseIntent,
        Some(release_ownership(ParticipantRole::StateAnchor)),
    )
    .unwrap();
    let clean = Participant::snapshot(
        generation_id(),
        ParticipantRole::StateAnchor,
        ParticipantState::Clean,
        Some(ParticipantOwnership::empty(ParticipantRole::StateAnchor)),
    )
    .unwrap();
    let states = [&intent, &active, &release, &clean];
    let expected = [
        [
            Ok(()),
            Ok(()),
            Ok(()),
            Err(RecordTransitionError::AbsenceProofRequired),
        ],
        [
            Err(RecordTransitionError::StateRegression),
            Ok(()),
            Ok(()),
            Err(RecordTransitionError::AbsenceProofRequired),
        ],
        [
            Err(RecordTransitionError::StateRegression),
            Err(RecordTransitionError::StateRegression),
            Ok(()),
            Err(RecordTransitionError::AbsenceProofRequired),
        ],
        [
            Err(RecordTransitionError::StateRegression),
            Err(RecordTransitionError::StateRegression),
            Err(RecordTransitionError::StateRegression),
            Ok(()),
        ],
    ];
    for (old_index, old) in states.into_iter().enumerate() {
        for (next_index, next) in states.into_iter().enumerate() {
            assert_eq!(
                old.validate_transition_to(next),
                expected[old_index][next_index]
            );
        }
    }

    let terminal_intent = terminal_snapshot(
        SessionTerminalState::Intent,
        owned_process(101),
        owned_process(102),
        true,
    )
    .unwrap();
    let terminal_running = terminal_snapshot(
        SessionTerminalState::Running,
        owned_process(101),
        owned_process(102),
        true,
    )
    .unwrap();
    let terminal_release = terminal_snapshot(
        SessionTerminalState::ReleaseIntent,
        owned_process(101),
        owned_process(102),
        true,
    )
    .unwrap();
    let terminal_clean = terminal_snapshot(
        SessionTerminalState::Clean,
        PersistedHostProcessOwnership::unassigned(),
        PersistedHostProcessOwnership::unassigned(),
        false,
    )
    .unwrap();
    let terminal_states = [
        &terminal_intent,
        &terminal_running,
        &terminal_release,
        &terminal_clean,
    ];
    for (old_index, old) in terminal_states.into_iter().enumerate() {
        for (next_index, next) in terminal_states.into_iter().enumerate() {
            assert_eq!(
                old.validate_transition_to(next),
                expected[old_index][next_index]
            );
        }
    }
}

#[test]
fn prepared_lower_release_changes_only_the_lease_phase() {
    let bare = Participant::bare_intent(generation_id(), ParticipantRole::PreparedLower);
    let intent = Participant::snapshot(
        generation_id(),
        ParticipantRole::PreparedLower,
        ParticipantState::Intent,
        Some(complete_ownership(
            ParticipantRole::PreparedLower,
            ParticipantState::Intent,
        )),
    )
    .unwrap();
    assert_eq!(bare.validate_transition_to(&intent), Ok(()));
    let active = Participant::snapshot(
        generation_id(),
        ParticipantRole::PreparedLower,
        ParticipantState::LeaseHeld,
        Some(complete_ownership(
            ParticipantRole::PreparedLower,
            ParticipantState::LeaseHeld,
        )),
    )
    .unwrap();
    assert_eq!(intent.validate_transition_to(&active), Ok(()));
    assert_eq!(
        bare.validate_transition_to(&active),
        Err(RecordTransitionError::PriorAcquisitionSnapshotRequired)
    );
    let release = Participant::snapshot(
        generation_id(),
        ParticipantRole::PreparedLower,
        ParticipantState::ReleaseIntent,
        Some(release_ownership(ParticipantRole::PreparedLower)),
    )
    .unwrap();
    assert_eq!(active.validate_transition_to(&release), Ok(()));

    let ParticipantOwnership::PreparedLower(mut changed) =
        release_ownership(ParticipantRole::PreparedLower)
    else {
        unreachable!()
    };
    changed.path = Some(HostPath::new(b"changed".to_vec()));
    let changed = Participant::snapshot(
        generation_id(),
        ParticipantRole::PreparedLower,
        ParticipantState::ReleaseIntent,
        Some(ParticipantOwnership::PreparedLower(changed)),
    )
    .unwrap();
    assert_eq!(
        active.validate_transition_to(&changed),
        Err(RecordTransitionError::EvidenceChanged)
    );
}

#[test]
#[allow(clippy::too_many_lines)]
fn session_terminal_snapshot_and_transition_matrix_is_fail_closed() {
    let current_hc = process(101);
    let current_exec = process(102);
    let legacy = LegacyUnverifiedProcessIdentityV1::new(
        HostPidLocator::new(102),
        ProcessStartTimeTicks::new(1_020),
    );
    let bare = terminal_snapshot(
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::unassigned(),
        PersistedHostProcessOwnership::unassigned(),
        false,
    )
    .unwrap();
    for invalid in [
        PersistedHostProcessOwnership::unassigned(),
        PersistedHostProcessOwnership::staged(current_exec),
        PersistedHostProcessOwnership::legacy_unverified(legacy),
    ] {
        assert_eq!(
            terminal_snapshot(
                SessionTerminalState::Running,
                PersistedHostProcessOwnership::committed(current_hc),
                invalid,
                true,
            ),
            Err(RecordInvariantError::SessionTerminalSnapshotState)
        );
    }

    let staged = terminal_snapshot(
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::staged(current_hc),
        PersistedHostProcessOwnership::staged(current_exec),
        false,
    )
    .unwrap();
    assert_eq!(bare.validate_transition_to(&staged), Ok(()));
    let running = terminal_snapshot(
        SessionTerminalState::Running,
        PersistedHostProcessOwnership::committed(current_hc),
        PersistedHostProcessOwnership::committed(current_exec),
        true,
    )
    .unwrap();
    assert_eq!(staged.validate_transition_to(&running), Ok(()));

    let direct_commit = terminal_snapshot(
        SessionTerminalState::Intent,
        PersistedHostProcessOwnership::committed(current_hc),
        PersistedHostProcessOwnership::committed(current_exec),
        false,
    )
    .unwrap();
    assert_eq!(
        bare.validate_transition_to(&direct_commit),
        Err(RecordTransitionError::ProcessStageSkipped)
    );

    let release = terminal_snapshot(
        SessionTerminalState::ReleaseIntent,
        PersistedHostProcessOwnership::committed(current_hc),
        PersistedHostProcessOwnership::committed(current_exec),
        true,
    )
    .unwrap();
    assert_eq!(running.validate_transition_to(&release), Ok(()));
    let clean = terminal_snapshot(
        SessionTerminalState::Clean,
        PersistedHostProcessOwnership::unassigned(),
        PersistedHostProcessOwnership::unassigned(),
        false,
    )
    .unwrap();
    assert_eq!(
        release.validate_transition_to(&clean),
        Err(RecordTransitionError::AbsenceProofRequired)
    );
    assert_eq!(
        release.validate_transition_to_clean_after_absence(&clean),
        Ok(())
    );
    assert_eq!(
        bare.validate_transition_to_clean_after_absence(&clean),
        Ok(())
    );
    assert!(clean.host_coordinator().is_unassigned());
    assert!(clean.exec_client().is_unassigned());
    assert_eq!(clean.container_id(), None);
    assert_eq!(clean.pty_master(), None);
    assert_eq!(clean.guest_process(), None);

    let malformed_clean = RawSessionTerminal {
        generation_id: generation_id(),
        session_id: session_id(),
        state: SessionTerminalState::Clean,
        container_id: Some(ContainerId::new(opaque(29))),
        host_coordinator: PersistedHostProcessOwnership::unassigned(),
        pty_master: None,
        exec_client: PersistedHostProcessOwnership::unassigned(),
        guest_process: None,
    };
    assert!(matches!(
        decode::<SessionTerminal>(&encode(&malformed_clean).unwrap()),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
#[allow(clippy::too_many_lines)]
fn commit_acknowledges_only_the_exact_staged_acquisition() {
    let staged = StagedProcessOwnership::v1(
        domain_id(),
        generation_id(),
        coordinator_id(),
        HostProcessOwnershipSlot::HelperConnectivityHelper,
        process(92),
    );
    let acknowledgment = ProcessOwnershipCommitAck::v1(staged);
    assert!(acknowledgment.acknowledges(&staged));

    let identity = staged.identity();
    let same_authority_different_locator = PidfsProcessIdentityV1::new(
        identity.authority(),
        HostPidLocator::new(identity.host_pid().get() + 1),
        identity.start_time_ticks(),
    );
    assert!(identity.has_same_authority(&same_authority_different_locator));

    let mismatches = [
        StagedProcessOwnership::v1(
            DomainId::new(opaque(99)),
            staged.generation_id(),
            staged.coordinator_id(),
            staged.slot(),
            identity,
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            GenerationId::new(staged.generation_id().get() + 1),
            staged.coordinator_id(),
            staged.slot(),
            identity,
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            CoordinatorId::new(opaque(98)),
            staged.slot(),
            identity,
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            staged.coordinator_id(),
            HostProcessOwnershipSlot::HelperRuntimeSupervisor,
            identity,
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            staged.coordinator_id(),
            staged.slot(),
            PidfsProcessIdentityV1::new(
                PidfsProcessAuthorityV1 {
                    boot_id: BootId::new([0x43; 16]),
                    ..identity.authority()
                },
                identity.host_pid(),
                identity.start_time_ticks(),
            ),
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            staged.coordinator_id(),
            staged.slot(),
            PidfsProcessIdentityV1::new(
                PidfsProcessAuthorityV1 {
                    pidfs_st_dev: PidfsDeviceId::new(identity.authority().pidfs_st_dev.get() + 1),
                    ..identity.authority()
                },
                identity.host_pid(),
                identity.start_time_ticks(),
            ),
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            staged.coordinator_id(),
            staged.slot(),
            PidfsProcessIdentityV1::new(
                PidfsProcessAuthorityV1 {
                    pidfs_st_ino: PidfsInodeId::new(identity.authority().pidfs_st_ino.get() + 1),
                    ..identity.authority()
                },
                identity.host_pid(),
                identity.start_time_ticks(),
            ),
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            staged.coordinator_id(),
            staged.slot(),
            same_authority_different_locator,
        ),
        StagedProcessOwnership::v1(
            staged.domain_id(),
            staged.generation_id(),
            staged.coordinator_id(),
            staged.slot(),
            PidfsProcessIdentityV1::new(
                identity.authority(),
                identity.host_pid(),
                ProcessStartTimeTicks::new(identity.start_time_ticks().get() + 1),
            ),
        ),
    ];
    for mismatch in mismatches {
        assert!(!acknowledgment.acknowledges(&mismatch));
    }
}

#[test]
fn participant_and_credential_role_mismatches_are_rejected() {
    let participant = Participant::snapshot(
        generation_id(),
        ParticipantRole::PreparedLower,
        ParticipantState::LeaseHeld,
        Some(ParticipantOwnership::PreparedLower(prepared_ownership())),
    )
    .unwrap();
    assert_eq!(participant.role(), ParticipantRole::PreparedLower);
    let mismatched_participant = encode(&RawParticipant {
        generation_id: generation_id(),
        role: ParticipantRole::Network,
        state: ParticipantState::LeaseHeld,
        ownership: Some(ParticipantOwnership::PreparedLower(prepared_ownership())),
    })
    .unwrap();
    assert!(matches!(
        decode::<Participant>(&mismatched_participant),
        Err(CodecError::MalformedPayload(_))
    ));

    let tls = TlsCredential::session_coordinator(
        generation_identity(),
        coordinator_id(),
        vec![1],
        CertificateDigest::new(opaque(2)),
    );
    assert_eq!(tls.role(), TlsRole::SessionCoordinator);
    let mismatched_tls = encode(&RawTlsCredential {
        generation_identity: generation_identity(),
        role: TlsRole::SessionCoordinator,
        subject_id: TlsSubjectId::ClientCli(ClientId::new(opaque(3))),
        certificate_der: vec![1],
        certificate_sha256: CertificateDigest::new(opaque(2)),
    })
    .unwrap();
    assert!(matches!(
        decode::<TlsCredential>(&mismatched_tls),
        Err(CodecError::MalformedPayload(_))
    ));

    let mismatched_protocol = encode(&RawProtocolCredential {
        protocol_version: ProtocolVersion::V1,
        credential_id: ProtocolCredentialId::new(opaque(4)),
        generation_identity: generation_identity(),
        role: ProtocolCredentialRole::AgentSandbox,
        subject_id: ProtocolSubjectId::SessionCoordinator(coordinator_id()),
        public_key: vec![5; 32],
        issuer_signature: vec![6; 64],
    })
    .unwrap();
    assert!(matches!(
        decode::<ProtocolCredential>(&mismatched_protocol),
        Err(CodecError::MalformedPayload(_))
    ));

    let coordinator = protocol_credential(ProtocolCredentialRole::SessionCoordinator);
    let sandbox = protocol_credential(ProtocolCredentialRole::AgentSandbox);
    assert!(
        SandboxCredentialBundle::new(
            generation_identity(),
            trust_manifest(),
            sandbox.clone(),
            coordinator.clone(),
            vec![7; 32],
        )
        .is_err()
    );

    let swapped_bundle = encode(&RawSandboxCredentialBundle {
        protocol_version: ProtocolVersion::V1,
        generation_identity: generation_identity(),
        trust_manifest: trust_manifest(),
        coordinator_protocol_credential: sandbox,
        sandbox_protocol_credential: coordinator,
        sandbox_private_key: vec![7; 32],
    })
    .unwrap();
    assert!(matches!(
        decode::<SandboxCredentialBundle>(&swapped_bundle),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
fn store_identity_and_versions_dispatch_before_rows() {
    for metadata in [
        RawMetadata {
            schema_id: String::new(),
            schema_version: DOMAIN_STORE_SCHEMA_VERSION,
        },
        RawMetadata {
            schema_id: DOMAIN_STORE_SCHEMA_ID.into(),
            schema_version: 0,
        },
        RawMetadata {
            schema_id: DOMAIN_STORE_SCHEMA_ID.into(),
            schema_version: DOMAIN_STORE_SCHEMA_VERSION + 1,
        },
        RawMetadata {
            schema_id: "unknown-store".into(),
            schema_version: DOMAIN_STORE_SCHEMA_VERSION,
        },
    ] {
        assert!(matches!(
            decode::<StoreMetadata>(&encode(&metadata).unwrap()),
            Err(CodecError::MalformedPayload(_))
        ));
    }

    assert!(matches!(
        decode::<StoreMetadata>(
            &postcard::to_stdvec(&MissingVersion {
                schema_id: DOMAIN_STORE_SCHEMA_ID.into(),
            })
            .unwrap()
        ),
        Err(CodecError::MalformedPayload(_))
    ));

    assert!(matches!(
        decode::<DomainStoreMetadata>(&encode(&StoreMetadata::storage_catalog_v1()).unwrap()),
        Err(CodecError::MalformedPayload(_))
    ));
    for domain_metadata in [
        StoreMetadata::domain_store_v2(),
        StoreMetadata::legacy_domain_store_v1(),
    ] {
        assert!(matches!(
            decode::<StorageCatalogMetadata>(&encode(&domain_metadata).unwrap()),
            Err(CodecError::MalformedPayload(_))
        ));
    }
    let legacy = StoreMetadata::legacy_domain_store_v1();
    assert_eq!(
        decode::<LegacyDomainStoreMetadata>(&encode(&legacy).unwrap()).unwrap(),
        LegacyDomainStoreMetadata::v1()
    );
    assert!(matches!(
        decode::<DomainStoreMetadata>(&encode(&legacy).unwrap()),
        Err(CodecError::MalformedPayload(_))
    ));

    let wrong_domain_store = postcard::to_stdvec(&RawDomainStore {
        metadata: StoreMetadata::storage_catalog_v1(),
        domain: Domain {
            domain_id: domain_id(),
            profile_name: "trusted".into(),
            canonical_workspace: HostPath::new(b"workspace".to_vec()),
        },
        generations: vec![],
        participants: vec![],
        sessions: vec![],
        session_terminals: vec![],
        mailboxes: vec![],
    })
    .unwrap();
    assert!(matches!(
        decode::<DomainStore>(&wrong_domain_store),
        Err(CodecError::MalformedPayload(_))
    ));

    let wrong_catalog = postcard::to_stdvec(&RawStorageCatalog {
        metadata: StoreMetadata::domain_store_v2(),
        prepared_artifacts: vec![],
        artifact_leases: vec![],
        prepare_operations: vec![],
    })
    .unwrap();
    assert!(matches!(
        decode::<StorageCatalog>(&wrong_catalog),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
fn network_document_schema_identities_are_fixed_to_v1() {
    let invalid_ids = |schema_id: &str, wrong_document_id: &str| {
        [
            String::new(),
            schema_id.replace("-v1", "-v0"),
            schema_id.replace("-v1", "-v2"),
            wrong_document_id.into(),
        ]
    };

    let policy = DerivedNetworkPolicy::v1(DerivedNetworkPolicyFields {
        address_families: vec![],
        default_actions: vec![],
        prefix_rules: vec![],
        resolvers: vec![],
    });
    assert_eq!(policy.schema_id(), DERIVED_NETWORK_POLICY_SCHEMA_ID);
    for schema_id in invalid_ids(
        DERIVED_NETWORK_POLICY_SCHEMA_ID,
        NETWORK_READINESS_ATTESTATION_SCHEMA_ID,
    ) {
        let encoded = postcard::to_stdvec(&RawDerivedNetworkPolicy {
            schema_id,
            address_families: vec![],
            default_actions: vec![],
            prefix_rules: vec![],
            resolvers: vec![],
        })
        .unwrap();
        assert!(matches!(
            decode::<DerivedNetworkPolicy>(&encoded),
            Err(CodecError::MalformedPayload(_))
        ));
    }

    let attestation = readiness_attestation();
    assert_eq!(
        attestation.schema_id(),
        NETWORK_READINESS_ATTESTATION_SCHEMA_ID
    );
    for schema_id in invalid_ids(
        NETWORK_READINESS_ATTESTATION_SCHEMA_ID,
        NETWORK_READINESS_IDENTITY_SCHEMA_ID,
    ) {
        let encoded = postcard::to_stdvec(&RawNetworkReadinessAttestation {
            schema_id,
            domain_id: domain_id(),
            generation_id: generation_id(),
            configuration_fingerprint: ConfigurationFingerprint::new(opaque(1)),
            network_sidecar_id: sidecar_id(),
            derived_policy_digest: DerivedPolicyDigest::new(opaque(2)),
            enforcement_namespace: namespace(),
            runtime_supervisor: process(3),
            connectivity_helper: process(4),
            observation: network_observation(),
        })
        .unwrap();
        assert!(matches!(
            decode::<NetworkReadinessAttestation>(&encoded),
            Err(CodecError::MalformedPayload(_))
        ));
    }

    let identity = readiness_identity();
    assert_eq!(identity.schema_id(), NETWORK_READINESS_IDENTITY_SCHEMA_ID);
    for schema_id in invalid_ids(
        NETWORK_READINESS_IDENTITY_SCHEMA_ID,
        NETWORK_READINESS_ATTESTATION_SCHEMA_ID,
    ) {
        let encoded = postcard::to_stdvec(&RawNetworkReadinessIdentity {
            schema_id,
            domain_id: domain_id(),
            generation_id: generation_id(),
            configuration_fingerprint: ConfigurationFingerprint::new(opaque(1)),
            network_sidecar_id: sidecar_id(),
            derived_policy_digest: DerivedPolicyDigest::new(opaque(2)),
            enforcement_namespace: namespace(),
        })
        .unwrap();
        assert!(matches!(
            decode::<NetworkReadinessIdentity>(&encoded),
            Err(CodecError::MalformedPayload(_))
        ));
    }
}

#[test]
fn zero_terminal_dimensions_are_rejected() {
    assert!(NonZeroU16::new(0).is_none());

    for (rows, columns) in [(0, 80), (24, 0)] {
        let encoded = postcard::to_stdvec(&RawTerminalResize {
            session_id: session_id(),
            rows,
            columns,
            pixel_width: 0,
            pixel_height: 0,
        })
        .unwrap();
        assert!(matches!(
            decode::<TerminalResize>(&encoded),
            Err(CodecError::MalformedPayload(_))
        ));
    }
}

#[test]
fn generic_schema_record_limit_is_enforced_before_decode() {
    let oversized = vec![0; MAX_SCHEMA_RECORD_BYTES + 1];
    assert!(matches!(
        decode::<Vec<u8>>(&oversized),
        Err(CodecError::RecordTooLarge { .. })
    ));
    assert!(matches!(
        encode(&oversized),
        Err(CodecError::RecordTooLarge { .. })
    ));
}

#[test]
fn session_exit_code_is_accepted_only_for_exited() {
    let encoded = encode(&InvalidSessionExit {
        session_id: session_id(),
        outcome: SessionExitOutcome::Terminated,
        exit_code: Some(1),
    })
    .unwrap();
    assert!(matches!(
        decode::<SessionExit>(&encoded),
        Err(CodecError::MalformedPayload(_))
    ));
}

#[test]
fn record_encoding_has_stable_byte_order() {
    assert_eq!(encode(&ParticipantState::Intent).unwrap(), [0]);
}