lean_ctx_protocol/

edge_via.rs

// SPDX-License-Identifier: Apache-2.0

//! LeanCTX Edge↔Via protocol V1 core contracts.
//!
//! These metadata-first types intentionally contain no provider request,
//! header, cookie, credential, or arbitrary JSON-patch field. Edge validates
//! and executes typed plans; Via never receives authority to rewrite an opaque
//! provider request.

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::{self, Write as _};

use chrono::DateTime;
use serde::{
    Deserialize, Deserializer, Serialize,
    de::{Error as DeError, MapAccess, SeqAccess, Visitor},
};
use serde_json::{Map, Value};
use sha2::{Digest, Sha256};

use crate::{
    AgentId, CapabilityId, EventId, PlanId, ProjectId, ProtocolReference, SemanticVersion,
    SessionId, Sha256Digest, UtcTimestamp, ValidationError, WorkspaceId,
};

pub const VIA_PROTOCOL_V1: u16 = 1;
pub const MAX_VIA_CAUSALITY: usize = 32;
pub const MAX_VIA_CAPABILITIES: usize = 64;
pub const MAX_VIA_CONTENT_ITEMS: usize = 4_096;
pub const MAX_VIA_PLAN_SOURCES: usize = 64;
pub const MAX_VIA_FRAME_BYTES: usize = 65_536;
pub const MAX_VIA_MANIFEST_BYTES: usize = 262_144;
pub const MAX_VIA_PLAN_BYTES: usize = 65_536;
pub const MAX_VIA_WIRE_BYTES: usize = 262_656;
pub const MAX_VIA_OUTPUT_TOKENS: u64 = 1_000_000;
pub const MAX_VIA_PLAN_TTL_SECONDS: u64 = 300;
pub const MAX_VIA_ACK_EVENTS: u64 = 1_024;
pub const MAX_VIA_ACK_GRAPH_NODES: u64 = 4_096;
pub const MAX_VIA_ACK_GRAPH_EDGES: u64 = 8_192;
pub const MAX_VIA_STREAM_WINDOW: u16 = 32;
pub const MAX_VIA_BACKPRESSURE_RETRY_MS: u32 = 30_000;
pub const MAX_VIA_CONTENT_CHUNKS_PER_REQUEST: usize = 4;
pub const MAX_VIA_CONTENT_CHUNK_BYTES: usize = 16 * 1_024;
pub const MAX_VIA_CONTENT_REQUEST_BYTES: u64 = 64 * 1_024;
pub const MAX_VIA_CONTENT_REQUEST_TOKENS: u64 = 16_384;
pub const MAX_VIA_CONTENT_REQUEST_TTL_SECONDS: u64 = 30;
pub const MAX_VIA_INFERENCE_SEGMENTS: usize = 128;
pub const MAX_VIA_INFERENCE_LINEAGE: usize = 8;
pub const MAX_VIA_INFERENCE_MANIFEST_BYTES: usize = 49_152;
pub const MAX_VIA_INFERENCE_CANDIDATES: usize = 64;
pub const MAX_VIA_INFERENCE_CANDIDATE_SET_BYTES: usize = 65_536;
pub const MAX_VIA_PLAN_REQUEST_BYTES: usize = 65_536;

fn deserialize_via_protocol_version<'de, D>(deserializer: D) -> Result<u16, D::Error>
where
    D: Deserializer<'de>,
{
    let version = u16::deserialize(deserializer)?;
    if version == VIA_PROTOCOL_V1 {
        Ok(version)
    } else {
        Err(DeError::custom(format!(
            "unsupported Via protocol_version {version}; expected {VIA_PROTOCOL_V1}"
        )))
    }
}

fn validate_protocol_version(version: u16) -> Result<(), ValidationError> {
    if version == VIA_PROTOCOL_V1 {
        Ok(())
    } else {
        Err(ValidationError::new(format!(
            "unsupported Via protocol_version {version}; expected {VIA_PROTOCOL_V1}"
        )))
    }
}

fn sha256(bytes: &[u8]) -> Result<Sha256Digest, ValidationError> {
    let mut value = String::with_capacity(71);
    value.push_str("sha256:");
    for byte in Sha256::digest(bytes) {
        write!(&mut value, "{byte:02x}").expect("writing to String cannot fail");
    }
    Sha256Digest::new(value)
}

fn canonical_json_bytes<T: Serialize>(value: &T) -> Result<Vec<u8>, ValidationError> {
    let value = serde_json::to_value(value)
        .map_err(|error| ValidationError::new(format!("serialize canonical JSON: {error}")))?;
    serde_json::to_vec(&canonical_json_value(value))
        .map_err(|error| ValidationError::new(format!("encode canonical JSON: {error}")))
}

fn canonical_json_value(value: serde_json::Value) -> serde_json::Value {
    match value {
        serde_json::Value::Object(fields) => {
            let sorted = fields
                .into_iter()
                .map(|(key, value)| (key, canonical_json_value(value)))
                .collect::<BTreeMap<_, _>>();
            serde_json::Value::Object(sorted.into_iter().collect())
        }
        serde_json::Value::Array(values) => {
            serde_json::Value::Array(values.into_iter().map(canonical_json_value).collect())
        }
        scalar => scalar,
    }
}

fn validate_content_ttl(
    now: &UtcTimestamp,
    expires_at: &UtcTimestamp,
) -> Result<(), ValidationError> {
    let now = DateTime::parse_from_rfc3339(now.as_str())
        .map_err(|_| ValidationError::new("content request time is invalid"))?;
    let expires_at = DateTime::parse_from_rfc3339(expires_at.as_str())
        .map_err(|_| ValidationError::new("content request expiry is invalid"))?;
    let seconds = expires_at.signed_duration_since(now).num_seconds();
    if !(1..=MAX_VIA_CONTENT_REQUEST_TTL_SECONDS as i64).contains(&seconds) {
        return Err(ValidationError::new(
            "content request TTL must be active and at most 30 seconds",
        ));
    }
    Ok(())
}

fn validate_plan_ttl(now: &UtcTimestamp, expires_at: &UtcTimestamp) -> Result<(), ValidationError> {
    let now = DateTime::parse_from_rfc3339(now.as_str())
        .map_err(|_| ValidationError::new("plan time is invalid"))?;
    let expires_at = DateTime::parse_from_rfc3339(expires_at.as_str())
        .map_err(|_| ValidationError::new("plan expiry is invalid"))?;
    let seconds = expires_at.signed_duration_since(now).num_seconds();
    if !(1..=MAX_VIA_PLAN_TTL_SECONDS as i64).contains(&seconds) {
        return Err(ValidationError::new(
            "plan TTL must be active and at most 300 seconds",
        ));
    }
    Ok(())
}

fn validate_unique<T: Ord + Clone>(
    values: &[T],
    maximum: usize,
    field: &str,
) -> Result<(), ValidationError> {
    if values.len() > maximum {
        return Err(ValidationError::new(format!(
            "{field} exceeds the {maximum} item limit"
        )));
    }
    let unique = values.iter().cloned().collect::<BTreeSet<_>>();
    if unique.len() != values.len() {
        return Err(ValidationError::new(format!(
            "{field} must not contain duplicates"
        )));
    }
    Ok(())
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ViaClassificationV1 {
    Public,
    Normal,
    Sensitive,
    Secret,
    LocalOnly,
    EnterprisePrivate,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaScopeV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub event_or_request_id: ProtocolReference,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub causality_id: Option<ProtocolReference>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub user_id: Option<ProtocolReference>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub device_id: Option<ProtocolReference>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub project_id: Option<ProjectId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub workspace_id: Option<WorkspaceId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<SessionId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub conversation_id: Option<ProtocolReference>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub agent_id: Option<AgentId>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub team_id: Option<ProtocolReference>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub organization_id: Option<ProtocolReference>,
    pub classification: ViaClassificationV1,
}

impl ViaScopeV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)
    }

    pub fn digest(&self) -> Result<Sha256Digest, ValidationError> {
        self.validate()?;
        let bytes = serde_json::to_vec(self)
            .map_err(|error| ValidationError::new(format!("serialize Via scope: {error}")))?;
        sha256(&bytes)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ViaCapabilityV1 {
    EventStream,
    ContentManifest,
    ContentOnDemand,
    TypedPlans,
    MultiAgentIdentity,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaProtocolHelloV1 {
    pub minimum_version: u16,
    pub maximum_version: u16,
    pub capabilities: Vec<ViaCapabilityV1>,
}

impl ViaProtocolHelloV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        if self.minimum_version == 0 || self.minimum_version > self.maximum_version {
            return Err(ValidationError::new(
                "ViaProtocolHelloV1 version range must be nonzero and ordered",
            ));
        }
        validate_unique(
            &self.capabilities,
            MAX_VIA_CAPABILITIES,
            "ViaProtocolHelloV1 capabilities",
        )
    }

    pub fn negotiate(&self, peer: &Self) -> Result<u16, ValidationError> {
        self.validate()?;
        peer.validate()?;
        let minimum = self.minimum_version.max(peer.minimum_version);
        let maximum = self.maximum_version.min(peer.maximum_version);
        if minimum > maximum || !(minimum..=maximum).contains(&VIA_PROTOCOL_V1) {
            return Err(ValidationError::new(
                "Via protocol version ranges have no locally supported overlap",
            ));
        }
        Ok(VIA_PROTOCOL_V1)
    }

    pub fn shared_capabilities(
        &self,
        peer: &Self,
    ) -> Result<Vec<ViaCapabilityV1>, ValidationError> {
        self.negotiate(peer)?;
        let peer_capabilities = peer.capabilities.iter().copied().collect::<BTreeSet<_>>();
        let mut shared = self
            .capabilities
            .iter()
            .copied()
            .filter(|capability| peer_capabilities.contains(capability))
            .collect::<Vec<_>>();
        shared.sort_unstable();
        Ok(shared)
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ScopedContentIdV1 {
    pub scope_digest: Sha256Digest,
    pub content_digest: Sha256Digest,
}

impl ScopedContentIdV1 {
    /// Derive path-independent identity from exact bytes and canonical scope.
    pub fn from_scope_and_bytes(
        scope: &ViaScopeV1,
        content: &[u8],
    ) -> Result<Self, ValidationError> {
        Ok(Self {
            scope_digest: scope.digest()?,
            content_digest: sha256(content)?,
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ScopedReferenceV1 {
    pub scope_digest: Sha256Digest,
    pub reference: ProtocolReference,
}

impl ScopedReferenceV1 {
    pub fn new(scope: &ViaScopeV1, reference: ProtocolReference) -> Result<Self, ValidationError> {
        Ok(Self {
            scope_digest: scope.digest()?,
            reference,
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContentKindV1 {
    SourceFile,
    Message,
    SearchResult,
    ShellResult,
    McpResult,
    InferenceManifest,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContentAvailabilityV1 {
    EdgeOnly,
    TransientRemoteAllowed,
    DeniedRemote,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TokenCountV1 {
    pub family: CapabilityId,
    pub tokens: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferenceSegmentRoleV1 {
    System,
    Developer,
    User,
    Assistant,
    ToolCall,
    ToolResult,
    ToolSchema,
    RequestControl,
}

impl InferenceSegmentRoleV1 {
    const fn discriminant(self) -> u8 {
        match self {
            Self::System => 0,
            Self::Developer => 1,
            Self::User => 2,
            Self::Assistant => 3,
            Self::ToolCall => 4,
            Self::ToolResult => 5,
            Self::ToolSchema => 6,
            Self::RequestControl => 7,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferenceMutabilityV1 {
    Immutable,
    Conditional,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferenceDeliveryStateV1 {
    CacheMarked,
    Uncached,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferenceTokenizerFamilyV1 {
    Cl100kBaseApproxV1,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferenceCachePolicyV1 {
    AnthropicExplicitCacheControlV1,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InferenceSegmentV1 {
    pub segment_id: Sha256Digest,
    pub position: u16,
    pub content_or_representation_id: ScopedContentIdV1,
    pub role: InferenceSegmentRoleV1,
    pub mutability: InferenceMutabilityV1,
    pub guard_hash: Sha256Digest,
    pub token_count: u64,
    pub delivery_state: InferenceDeliveryStateV1,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub source_lineage: Vec<ScopedContentIdV1>,
}

impl InferenceSegmentV1 {
    #[allow(clippy::too_many_arguments)] // Mirrors the closed wire fields; no options bag or arbitrary JSON.
    pub fn from_canonical_bytes(
        scope: &ViaScopeV1,
        position: u16,
        role: InferenceSegmentRoleV1,
        mutability: InferenceMutabilityV1,
        canonical_bytes: &[u8],
        token_count: u64,
        delivery_state: InferenceDeliveryStateV1,
        source_lineage: Vec<ScopedContentIdV1>,
    ) -> Result<Self, ValidationError> {
        let content_or_representation_id =
            ScopedContentIdV1::from_scope_and_bytes(scope, canonical_bytes)?;
        let guard_hash = content_or_representation_id.content_digest.clone();
        let segment_id = inference_segment_id(scope, position, role, &guard_hash)?;
        let segment = Self {
            segment_id,
            position,
            content_or_representation_id,
            role,
            mutability,
            guard_hash,
            token_count,
            delivery_state,
            source_lineage,
        };
        segment.validate(scope, usize::from(position))?;
        Ok(segment)
    }

    fn validate(
        &self,
        scope: &ViaScopeV1,
        expected_position: usize,
    ) -> Result<(), ValidationError> {
        if usize::from(self.position) != expected_position {
            return Err(ValidationError::new(
                "InferenceSegmentV1 positions must be contiguous and ordered",
            ));
        }
        let scope_digest = scope.digest()?;
        if self.content_or_representation_id.scope_digest != scope_digest
            || self
                .source_lineage
                .iter()
                .any(|source| source.scope_digest != scope_digest)
        {
            return Err(ValidationError::new(
                "InferenceSegmentV1 content and lineage IDs must match manifest scope",
            ));
        }
        if self.guard_hash != self.content_or_representation_id.content_digest {
            return Err(ValidationError::new(
                "InferenceSegmentV1 guard_hash must match content identity",
            ));
        }
        if self.source_lineage.len() > MAX_VIA_INFERENCE_LINEAGE {
            return Err(ValidationError::new(format!(
                "InferenceSegmentV1 source_lineage exceeds the {MAX_VIA_INFERENCE_LINEAGE} item limit"
            )));
        }
        validate_unique(
            &self.source_lineage,
            MAX_VIA_INFERENCE_LINEAGE,
            "InferenceSegmentV1 source_lineage",
        )?;
        if self.segment_id
            != inference_segment_id(scope, self.position, self.role, &self.guard_hash)?
        {
            return Err(ValidationError::new(
                "InferenceSegmentV1 segment_id does not match its canonical binding",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InferenceAccountingV1 {
    pub total_tokens: u64,
    pub immutable_tokens: u64,
    pub conditional_tokens: u64,
    pub cache_marked_tokens: u64,
    pub uncached_tokens: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InferenceManifestV1 {
    pub scope: ViaScopeV1,
    pub manifest_id: ScopedContentIdV1,
    pub tokenizer_family: InferenceTokenizerFamilyV1,
    pub cache_policy: InferenceCachePolicyV1,
    pub segments: Vec<InferenceSegmentV1>,
    pub accounting: InferenceAccountingV1,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stable_prefix_digest: Option<Sha256Digest>,
}

#[derive(Serialize)]
struct InferenceManifestIdentityV1<'a> {
    accounting: &'a InferenceAccountingV1,
    cache_policy: InferenceCachePolicyV1,
    scope: &'a ViaScopeV1,
    segments: &'a [InferenceSegmentV1],
    stable_prefix_digest: &'a Option<Sha256Digest>,
    tokenizer_family: InferenceTokenizerFamilyV1,
}

impl InferenceManifestV1 {
    pub fn new(
        scope: ViaScopeV1,
        segments: Vec<InferenceSegmentV1>,
        stable_prefix_digest: Option<Sha256Digest>,
    ) -> Result<Self, ValidationError> {
        let accounting = inference_accounting(&segments)?;
        let placeholder = ScopedContentIdV1::from_scope_and_bytes(&scope, b"")?;
        let mut manifest = Self {
            scope,
            manifest_id: placeholder,
            tokenizer_family: InferenceTokenizerFamilyV1::Cl100kBaseApproxV1,
            cache_policy: InferenceCachePolicyV1::AnthropicExplicitCacheControlV1,
            segments,
            accounting,
            stable_prefix_digest,
        };
        manifest.manifest_id = manifest.expected_manifest_id()?;
        manifest.validate()?;
        Ok(manifest)
    }

    pub fn validate(&self) -> Result<(), ValidationError> {
        self.scope.validate()?;
        if self.segments.is_empty() || self.segments.len() > MAX_VIA_INFERENCE_SEGMENTS {
            return Err(ValidationError::new(format!(
                "InferenceManifestV1 segments must contain 1..={MAX_VIA_INFERENCE_SEGMENTS} items"
            )));
        }
        let mut segment_ids = BTreeSet::new();
        for (position, segment) in self.segments.iter().enumerate() {
            segment.validate(&self.scope, position)?;
            if !segment_ids.insert(segment.segment_id.clone()) {
                return Err(ValidationError::new(
                    "InferenceManifestV1 segment IDs must be unique",
                ));
            }
        }
        if self.accounting != inference_accounting(&self.segments)? {
            return Err(ValidationError::new(
                "InferenceManifestV1 accounting does not match segment tokens",
            ));
        }
        if self.manifest_id != self.expected_manifest_id()? {
            return Err(ValidationError::new(
                "InferenceManifestV1 manifest_id does not match canonical fields",
            ));
        }
        let bytes = serde_json::to_vec(self).map_err(|error| {
            ValidationError::new(format!("serialize inference manifest: {error}"))
        })?;
        if bytes.len() > MAX_VIA_INFERENCE_MANIFEST_BYTES {
            return Err(ValidationError::new(format!(
                "InferenceManifestV1 exceeds the {MAX_VIA_INFERENCE_MANIFEST_BYTES} byte limit"
            )));
        }
        Ok(())
    }

    fn expected_manifest_id(&self) -> Result<ScopedContentIdV1, ValidationError> {
        let body = InferenceManifestIdentityV1 {
            accounting: &self.accounting,
            cache_policy: self.cache_policy,
            scope: &self.scope,
            segments: &self.segments,
            stable_prefix_digest: &self.stable_prefix_digest,
            tokenizer_family: self.tokenizer_family,
        };
        let encoded = canonical_json_bytes(&body)?;
        let mut domain = b"leanctx.inference.manifest.v1\0".to_vec();
        domain.extend_from_slice(&encoded);
        ScopedContentIdV1::from_scope_and_bytes(&self.scope, &domain)
    }
}

fn inference_accounting(
    segments: &[InferenceSegmentV1],
) -> Result<InferenceAccountingV1, ValidationError> {
    let mut accounting = InferenceAccountingV1 {
        total_tokens: 0,
        immutable_tokens: 0,
        conditional_tokens: 0,
        cache_marked_tokens: 0,
        uncached_tokens: 0,
    };
    for segment in segments {
        accounting.total_tokens = accounting
            .total_tokens
            .checked_add(segment.token_count)
            .ok_or_else(|| ValidationError::new("InferenceManifestV1 total token overflow"))?;
        let mutability = match segment.mutability {
            InferenceMutabilityV1::Immutable => &mut accounting.immutable_tokens,
            InferenceMutabilityV1::Conditional => &mut accounting.conditional_tokens,
        };
        *mutability = mutability
            .checked_add(segment.token_count)
            .ok_or_else(|| ValidationError::new("InferenceManifestV1 mutability token overflow"))?;
        let delivery = match segment.delivery_state {
            InferenceDeliveryStateV1::CacheMarked => &mut accounting.cache_marked_tokens,
            InferenceDeliveryStateV1::Uncached => &mut accounting.uncached_tokens,
        };
        *delivery = delivery
            .checked_add(segment.token_count)
            .ok_or_else(|| ValidationError::new("InferenceManifestV1 delivery token overflow"))?;
    }
    Ok(accounting)
}

fn inference_segment_id(
    scope: &ViaScopeV1,
    position: u16,
    role: InferenceSegmentRoleV1,
    guard_hash: &Sha256Digest,
) -> Result<Sha256Digest, ValidationError> {
    let mut bytes = b"leanctx.inference.segment.v1\0".to_vec();
    bytes.extend_from_slice(scope.digest()?.as_str().as_bytes());
    bytes.extend_from_slice(&position.to_be_bytes());
    bytes.push(role.discriminant());
    bytes.extend_from_slice(guard_hash.as_str().as_bytes());
    sha256(&bytes)
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ChunkDescriptorV1 {
    pub chunk_id: Sha256Digest,
    pub start_byte: u64,
    pub end_byte: u64,
    pub byte_count: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub symbol: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub token_count: Option<u64>,
    pub operator_id: CapabilityId,
    pub operator_version: SemanticVersion,
}

impl ChunkDescriptorV1 {
    fn validate(&self, content_bytes: u64) -> Result<(), ValidationError> {
        if self.start_byte >= self.end_byte
            || self.end_byte > content_bytes
            || self.byte_count != self.end_byte - self.start_byte
        {
            return Err(ValidationError::new(
                "ChunkDescriptorV1 byte range and byte_count are inconsistent",
            ));
        }
        if self.symbol.as_ref().is_some_and(|value| {
            value.trim().is_empty() || value.len() > 1_024 || value.chars().any(char::is_control)
        }) {
            return Err(ValidationError::new(
                "ChunkDescriptorV1 symbol must be bounded printable text",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContentManifestV1 {
    pub scope: ViaScopeV1,
    pub content_id: ScopedContentIdV1,
    pub kind: ContentKindV1,
    pub byte_count: u64,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub token_counts: Vec<TokenCountV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_version: Option<Sha256Digest>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub language: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub structural_manifest: Option<Sha256Digest>,
    pub chunks: Vec<ChunkDescriptorV1>,
    pub availability: ContentAvailabilityV1,
}

impl ContentManifestV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        self.scope.validate()?;
        if self.content_id.scope_digest != self.scope.digest()? {
            return Err(ValidationError::new(
                "ContentManifestV1 content_id scope does not match manifest scope",
            ));
        }
        if self.chunks.len() > MAX_VIA_CONTENT_ITEMS {
            return Err(ValidationError::new(format!(
                "ContentManifestV1 chunks exceeds the {MAX_VIA_CONTENT_ITEMS} item limit"
            )));
        }
        if self.byte_count == 0 && !self.chunks.is_empty()
            || self.byte_count > 0 && self.chunks.is_empty()
        {
            return Err(ValidationError::new(
                "ContentManifestV1 empty state is inconsistent with chunks",
            ));
        }
        if matches!(
            self.scope.classification,
            ViaClassificationV1::Secret | ViaClassificationV1::LocalOnly
        ) && self.availability == ContentAvailabilityV1::TransientRemoteAllowed
        {
            return Err(ValidationError::new(
                "secret or local-only content cannot allow remote availability",
            ));
        }
        if self.language.as_ref().is_some_and(|value| {
            value.trim().is_empty() || value.len() > 128 || value.chars().any(char::is_control)
        }) {
            return Err(ValidationError::new(
                "ContentManifestV1 language must be bounded printable text",
            ));
        }
        let families = self
            .token_counts
            .iter()
            .map(|entry| entry.family.as_str().to_owned())
            .collect::<Vec<_>>();
        validate_unique(
            &families,
            MAX_VIA_CAPABILITIES,
            "ContentManifestV1 token families",
        )?;
        let mut previous_end = 0;
        let mut chunk_ids = BTreeSet::new();
        for chunk in &self.chunks {
            chunk.validate(self.byte_count)?;
            if chunk.start_byte != previous_end || !chunk_ids.insert(chunk.chunk_id.clone()) {
                return Err(ValidationError::new(
                    "ContentManifestV1 chunks must be contiguous, ordered, and unique",
                ));
            }
            previous_end = chunk.end_byte;
        }
        if previous_end != self.byte_count {
            return Err(ValidationError::new(
                "ContentManifestV1 chunks must cover the declared byte_count",
            ));
        }
        let bytes = serde_json::to_vec(self).map_err(|error| {
            ValidationError::new(format!("serialize content manifest: {error}"))
        })?;
        if bytes.len() > MAX_VIA_MANIFEST_BYTES {
            return Err(ValidationError::new(format!(
                "ContentManifestV1 exceeds the {MAX_VIA_MANIFEST_BYTES} byte limit"
            )));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NeedContentPurposeV1 {
    ContextPlanning,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NeedContentV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub request_id: ScopedReferenceV1,
    pub source_content_id: ScopedContentIdV1,
    pub chunk_ids: Vec<Sha256Digest>,
    pub purpose: NeedContentPurposeV1,
    pub max_bytes: u64,
    pub max_tokens: u64,
    pub expires_at: UtcTimestamp,
    pub policy_version: SemanticVersion,
}

impl NeedContentV1 {
    pub fn validate_at(&self, now: &UtcTimestamp) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        if self.request_id.scope_digest != self.source_content_id.scope_digest {
            return Err(ValidationError::new(
                "NeedContentV1 request and source scopes must match",
            ));
        }
        if self.chunk_ids.is_empty() {
            return Err(ValidationError::new(
                "NeedContentV1 must request at least one chunk",
            ));
        }
        validate_unique(
            &self.chunk_ids,
            MAX_VIA_CONTENT_CHUNKS_PER_REQUEST,
            "NeedContentV1 chunk IDs",
        )?;
        if self.max_bytes == 0
            || self.max_bytes > MAX_VIA_CONTENT_REQUEST_BYTES
            || self.max_tokens == 0
            || self.max_tokens > MAX_VIA_CONTENT_REQUEST_TOKENS
        {
            return Err(ValidationError::new(
                "NeedContentV1 budgets and expiry must be bounded and active",
            ));
        }
        validate_content_ttl(now, &self.expires_at)?;
        Ok(())
    }
}

#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContentChunkV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub request_id: ScopedReferenceV1,
    pub source_content_id: ScopedContentIdV1,
    pub chunk_id: Sha256Digest,
    pub start_byte: u64,
    pub end_byte: u64,
    pub byte_count: u64,
    pub source_version: Sha256Digest,
    pub content: String,
    pub payload_digest: Sha256Digest,
    pub expires_at: UtcTimestamp,
}

impl fmt::Debug for ContentChunkV1 {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ContentChunkV1")
            .field("request_id", &self.request_id)
            .field("source_content_id", &self.source_content_id)
            .field("chunk_id", &self.chunk_id)
            .field("byte_count", &self.byte_count)
            .field("content", &"[REDACTED]")
            .field("payload_digest", &self.payload_digest)
            .finish()
    }
}

impl ContentChunkV1 {
    pub fn validate_at(&self, now: &UtcTimestamp) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        if self.request_id.scope_digest != self.source_content_id.scope_digest {
            return Err(ValidationError::new(
                "ContentChunkV1 request and source scopes must match",
            ));
        }
        let content = self.content.as_bytes();
        if self.start_byte >= self.end_byte
            || self.end_byte.checked_sub(self.start_byte) != Some(self.byte_count)
            || self.byte_count as usize != content.len()
            || content.len() > MAX_VIA_CONTENT_CHUNK_BYTES
        {
            return Err(ValidationError::new(
                "ContentChunkV1 range, size, or expiry is invalid",
            ));
        }
        validate_content_ttl(now, &self.expires_at)?;
        let digest = sha256(content)?;
        if digest != self.chunk_id || digest != self.payload_digest {
            return Err(ValidationError::new(
                "ContentChunkV1 payload digest does not bind exact content bytes",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContentChunkAckV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub request_id: ScopedReferenceV1,
    pub source_content_id: ScopedContentIdV1,
    pub chunk_id: Sha256Digest,
}

impl ContentChunkAckV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        if self.request_id.scope_digest != self.source_content_id.scope_digest {
            return Err(ValidationError::new(
                "ContentChunkAckV1 request and source scopes must match",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ViaObservedEventKindV1 {
    SessionStarted,
    SessionEnded,
    AgentObserved,
    UserMessageObserved,
    AssistantMessageObserved,
    FileObserved,
    FileReadObserved,
    FileChanged,
    SearchExecuted,
    SearchResultObserved,
    ShellExecuted,
    ShellResultObserved,
    McpToolCalled,
    McpResultObserved,
    ProviderRequestPending,
    ContextDelivered,
    ContextReused,
    ContextInvalidated,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ObservationPayloadV1 {
    pub kind: ViaObservedEventKindV1,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub input_ids: Vec<ScopedContentIdV1>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub output_ids: Vec<ScopedContentIdV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub capability_id: Option<CapabilityId>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PlanEventStatusV1 {
    Applied,
    Rejected,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PlanEventPayloadV1 {
    pub plan_id: PlanId,
    pub status: PlanEventStatusV1,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "event_type", content = "payload", rename_all = "snake_case")]
pub enum ViaEventPayloadV1 {
    Observation(ObservationPayloadV1),
    ContentManifest(Box<ContentManifestV1>),
    InferenceManifest(Box<InferenceManifestV1>),
    Plan(PlanEventPayloadV1),
}

impl ViaEventPayloadV1 {
    fn validate(&self, scope: &ViaScopeV1) -> Result<(), ValidationError> {
        match self {
            Self::Observation(payload) => {
                validate_unique(
                    &payload.input_ids,
                    MAX_VIA_CONTENT_ITEMS,
                    "ObservationPayloadV1 input_ids",
                )?;
                validate_unique(
                    &payload.output_ids,
                    MAX_VIA_CONTENT_ITEMS,
                    "ObservationPayloadV1 output_ids",
                )?;
                let scope_digest = scope.digest()?;
                if payload
                    .input_ids
                    .iter()
                    .chain(&payload.output_ids)
                    .any(|id| id.scope_digest != scope_digest)
                {
                    return Err(ValidationError::new(
                        "ObservationPayloadV1 content IDs must match frame scope",
                    ));
                }
                Ok(())
            }
            Self::ContentManifest(manifest) => {
                if manifest.scope != *scope {
                    return Err(ValidationError::new(
                        "ContentManifestV1 scope must match frame scope",
                    ));
                }
                manifest.validate()
            }
            Self::InferenceManifest(manifest) => {
                if manifest.scope != *scope {
                    return Err(ValidationError::new(
                        "InferenceManifestV1 scope must match frame scope",
                    ));
                }
                manifest.validate()
            }
            Self::Plan(_) => Ok(()),
        }
    }

    pub fn digest(&self) -> Result<Sha256Digest, ValidationError> {
        let bytes = serde_json::to_vec(self).map_err(|error| {
            ValidationError::new(format!("serialize Via event payload: {error}"))
        })?;
        sha256(&bytes)
    }
}

#[derive(Deserialize)]
struct ViaEventFrameV1Wire {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    protocol_version: u16,
    stream_id: ScopedReferenceV1,
    seq: u64,
    event_id: EventId,
    #[serde(default)]
    causality: Vec<EventId>,
    payload_hash: Sha256Digest,
    scope: ViaScopeV1,
    #[serde(flatten)]
    payload: ViaEventPayloadV1,
    #[serde(flatten)]
    unknown_fields: BTreeMap<String, serde_json::Value>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ViaEventFrameV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub stream_id: ScopedReferenceV1,
    pub seq: u64,
    pub event_id: EventId,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub causality: Vec<EventId>,
    pub payload_hash: Sha256Digest,
    pub scope: ViaScopeV1,
    #[serde(flatten)]
    pub payload: ViaEventPayloadV1,
}

impl<'de> Deserialize<'de> for ViaEventFrameV1 {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let wire = ViaEventFrameV1Wire::deserialize(deserializer)?;
        if !wire.unknown_fields.is_empty() {
            let fields = wire
                .unknown_fields
                .keys()
                .map(|field| field.as_str())
                .collect::<Vec<_>>()
                .join(", ");
            return Err(DeError::custom(format!(
                "ViaEventFrameV1 contains unknown top-level fields: {fields}"
            )));
        }
        Ok(Self {
            protocol_version: wire.protocol_version,
            stream_id: wire.stream_id,
            seq: wire.seq,
            event_id: wire.event_id,
            causality: wire.causality,
            payload_hash: wire.payload_hash,
            scope: wire.scope,
            payload: wire.payload,
        })
    }
}

impl ViaEventFrameV1 {
    pub fn new(
        stream_id: ScopedReferenceV1,
        seq: u64,
        event_id: EventId,
        causality: Vec<EventId>,
        scope: ViaScopeV1,
        payload: ViaEventPayloadV1,
    ) -> Result<Self, ValidationError> {
        let payload_hash = payload.digest()?;
        let frame = Self {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id,
            seq,
            event_id,
            causality,
            payload_hash,
            scope,
            payload,
        };
        frame.validate()?;
        Ok(frame)
    }

    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        self.scope.validate()?;
        if self.stream_id.scope_digest != self.scope.digest()? {
            return Err(ValidationError::new(
                "ViaEventFrameV1 stream_id scope does not match frame scope",
            ));
        }
        validate_unique(
            &self.causality,
            MAX_VIA_CAUSALITY,
            "ViaEventFrameV1 causality",
        )?;
        if self.causality.contains(&self.event_id) {
            return Err(ValidationError::new(
                "ViaEventFrameV1 event_id cannot be its own cause",
            ));
        }
        self.payload.validate(&self.scope)?;
        if self.payload.digest()? != self.payload_hash {
            return Err(ValidationError::new(
                "ViaEventFrameV1 payload_hash does not match payload",
            ));
        }
        let bytes = serde_json::to_vec(self)
            .map_err(|error| ValidationError::new(format!("serialize Via event frame: {error}")))?;
        if bytes.len() > MAX_VIA_FRAME_BYTES {
            return Err(ValidationError::new(format!(
                "ViaEventFrameV1 exceeds the {MAX_VIA_FRAME_BYTES} byte limit"
            )));
        }
        Ok(())
    }

    /// Validate deterministic stream ordering against the immediately prior frame.
    pub fn validate_successor(&self, previous: &Self) -> Result<(), ValidationError> {
        self.validate()?;
        previous.validate()?;
        let expected_seq = previous.seq.checked_add(1).ok_or_else(|| {
            ValidationError::new("ViaEventFrameV1 predecessor seq cannot advance")
        })?;
        if self.stream_id != previous.stream_id || self.seq != expected_seq {
            return Err(ValidationError::new(
                "ViaEventFrameV1 successor must share stream_id and increment seq by one",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ViaEventAckStatusV1 {
    Stored,
    Duplicate,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaEventAckV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub stream_id: ScopedReferenceV1,
    pub acknowledged_seq: u64,
    pub acknowledged_event_id: EventId,
    pub next_seq: u64,
    pub status: ViaEventAckStatusV1,
    pub accepted_event_count: u64,
    pub graph_nodes: u64,
    pub graph_edges: u64,
}

impl ViaEventAckV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        if self.acknowledged_seq.checked_add(1) != Some(self.next_seq) {
            return Err(ValidationError::new(
                "ViaEventAckV1 next_seq must immediately follow acknowledged_seq",
            ));
        }
        if self.accepted_event_count != self.next_seq
            || self.accepted_event_count > MAX_VIA_ACK_EVENTS
            || self.graph_nodes > MAX_VIA_ACK_GRAPH_NODES
            || self.graph_edges > MAX_VIA_ACK_GRAPH_EDGES
        {
            return Err(ValidationError::new(
                "ViaEventAckV1 accepted_event_count must equal the stream frontier and counters must remain bounded",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaBackpressureV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub stream_id: ScopedReferenceV1,
    pub accepted_window: u16,
    pub retry_after_ms: u32,
}

impl ViaBackpressureV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        if self.accepted_window > MAX_VIA_STREAM_WINDOW
            || self.retry_after_ms == 0
            || self.retry_after_ms > MAX_VIA_BACKPRESSURE_RETRY_MS
        {
            return Err(ValidationError::new(
                "ViaBackpressureV1 window or retry interval exceeds its bound",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaStreamResumeV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub stream_id: ScopedReferenceV1,
}

impl ViaStreamResumeV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaStreamPositionV1 {
    pub seq: u64,
    pub event_id: EventId,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaStreamCursorV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub stream_id: ScopedReferenceV1,
    pub next_seq: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_accepted: Option<ViaStreamPositionV1>,
}

impl ViaStreamCursorV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        match &self.last_accepted {
            None if self.next_seq == 0 => Ok(()),
            Some(position) if position.seq.checked_add(1) == Some(self.next_seq) => Ok(()),
            _ => Err(ValidationError::new(
                "ViaStreamCursorV1 next_seq must bind the last accepted position",
            )),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaPlanHeaderV1 {
    pub plan_id: PlanId,
    pub caused_by: EventId,
    pub policy_id: CapabilityId,
    pub policy_version: SemanticVersion,
    pub expected_source_ids: Vec<ScopedContentIdV1>,
    pub immutable_guard_digest: Sha256Digest,
    pub operator_id: CapabilityId,
    pub operator_version: SemanticVersion,
    pub expires_at: UtcTimestamp,
    pub output_budget_tokens: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ViaPlanRequestV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub stream_id: ScopedReferenceV1,
    pub event_id: EventId,
    pub immutable_guard_digest: Sha256Digest,
    pub now: UtcTimestamp,
    pub expires_at: UtcTimestamp,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub inference_candidates: Option<InferenceCandidateSetV1>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InferenceSegmentTransitionV1 {
    pub position: u16,
    pub old_segment_id: Sha256Digest,
    pub old_representation_id: ScopedContentIdV1,
    pub old_guard_hash: Sha256Digest,
    pub new_segment_id: Sha256Digest,
    pub new_representation_id: ScopedContentIdV1,
    pub new_guard_hash: Sha256Digest,
    pub old_token_count: u64,
    pub new_token_count: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InferenceCandidateSetV1 {
    pub candidate_set_id: ScopedContentIdV1,
    pub manifest_id: ScopedContentIdV1,
    pub operator_id: CapabilityId,
    pub operator_version: SemanticVersion,
    pub transitions: Vec<InferenceSegmentTransitionV1>,
}

#[derive(Serialize)]
struct InferenceCandidateSetIdentityV1<'a> {
    manifest_id: &'a ScopedContentIdV1,
    operator_id: &'a CapabilityId,
    operator_version: &'a SemanticVersion,
    transitions: &'a [InferenceSegmentTransitionV1],
}

#[derive(Serialize)]
struct InferencePlanIdIdentityV1<'a> {
    scope_digest: &'a Sha256Digest,
    caused_by: &'a EventId,
    manifest_id: &'a ScopedContentIdV1,
    candidate_set_id: &'a ScopedContentIdV1,
    selected_transitions: &'a [InferenceSegmentTransitionV1],
    output_budget_tokens: u64,
}

impl InferenceCandidateSetV1 {
    pub fn validate_for_scope(&self, scope_digest: &Sha256Digest) -> Result<(), ValidationError> {
        if self.manifest_id.scope_digest != *scope_digest
            || self.candidate_set_id.scope_digest != *scope_digest
        {
            return Err(ValidationError::new(
                "InferenceCandidateSetV1 scope does not match request",
            ));
        }
        if self.operator_id.as_str() != "leanctx.inference.history.compact"
            || self.operator_version.as_str() != "1.0.0"
        {
            return Err(ValidationError::new(
                "InferenceCandidateSetV1 operator is unsupported",
            ));
        }
        if self.transitions.is_empty() || self.transitions.len() > MAX_VIA_INFERENCE_CANDIDATES {
            return Err(ValidationError::new(format!(
                "InferenceCandidateSetV1 transitions must be within 1..={MAX_VIA_INFERENCE_CANDIDATES}"
            )));
        }
        let mut prior_position = None;
        let mut old_segments = BTreeSet::new();
        let mut new_segments = BTreeSet::new();
        for transition in &self.transitions {
            if prior_position.is_some_and(|position| transition.position <= position) {
                return Err(ValidationError::new(
                    "InferenceCandidateSetV1 transitions must be position-sorted",
                ));
            }
            prior_position = Some(transition.position);
            if !old_segments.insert(&transition.old_segment_id)
                || !new_segments.insert(&transition.new_segment_id)
                || transition.old_representation_id.scope_digest != *scope_digest
                || transition.new_representation_id.scope_digest != *scope_digest
                || transition.old_guard_hash != transition.old_representation_id.content_digest
                || transition.new_guard_hash != transition.new_representation_id.content_digest
                || transition.old_segment_id == transition.new_segment_id
                || transition.old_representation_id == transition.new_representation_id
                || transition.new_token_count >= transition.old_token_count
            {
                return Err(ValidationError::new(
                    "InferenceCandidateSetV1 transition identity or reduction is invalid",
                ));
            }
        }
        let encoded = serde_json::to_vec(self).map_err(|error| {
            ValidationError::new(format!("serialize inference candidate set: {error}"))
        })?;
        if encoded.len() > MAX_VIA_INFERENCE_CANDIDATE_SET_BYTES {
            return Err(ValidationError::new(format!(
                "InferenceCandidateSetV1 exceeds the {MAX_VIA_INFERENCE_CANDIDATE_SET_BYTES} byte limit"
            )));
        }
        if self.candidate_set_id != self.expected_id(scope_digest)? {
            return Err(ValidationError::new(
                "InferenceCandidateSetV1 ID does not match canonical metadata",
            ));
        }
        Ok(())
    }

    pub fn expected_id(
        &self,
        scope_digest: &Sha256Digest,
    ) -> Result<ScopedContentIdV1, ValidationError> {
        let identity = InferenceCandidateSetIdentityV1 {
            manifest_id: &self.manifest_id,
            operator_id: &self.operator_id,
            operator_version: &self.operator_version,
            transitions: &self.transitions,
        };
        let mut bytes = b"leanctx.inference.candidate-set.v1\0".to_vec();
        bytes.extend_from_slice(&serde_json::to_vec(&identity).map_err(|error| {
            ValidationError::new(format!("serialize inference candidate identity: {error}"))
        })?);
        Ok(ScopedContentIdV1 {
            scope_digest: scope_digest.clone(),
            content_digest: sha256(&bytes)?,
        })
    }
}

pub fn inference_plan_id(
    scope_digest: &Sha256Digest,
    caused_by: &EventId,
    manifest_id: &ScopedContentIdV1,
    candidate_set_id: &ScopedContentIdV1,
    selected_transitions: &[InferenceSegmentTransitionV1],
    output_budget_tokens: u64,
) -> Result<PlanId, ValidationError> {
    let identity = InferencePlanIdIdentityV1 {
        scope_digest,
        caused_by,
        manifest_id,
        candidate_set_id,
        selected_transitions,
        output_budget_tokens,
    };
    let mut bytes = b"leanctx.inference.plan-id.v1\0".to_vec();
    bytes.extend_from_slice(&serde_json::to_vec(&identity).map_err(|error| {
        ValidationError::new(format!("serialize inference plan identity: {error}"))
    })?);
    let digest = sha256(&bytes)?;
    let hex = digest
        .as_str()
        .strip_prefix("sha256:")
        .ok_or_else(|| ValidationError::new("inference plan digest prefix is invalid"))?;
    PlanId::new(format!("inference-{hex}"))
}

/// Deterministically binds every immutable inference-manifest field that an
/// optimizer is forbidden to change.
pub fn inference_immutable_guard(
    manifest: &InferenceManifestV1,
) -> Result<Sha256Digest, ValidationError> {
    manifest.validate()?;
    let mut bytes = b"leanctx.inference.immutable-guard.v1\0".to_vec();
    push_guard_text(&mut bytes, manifest.scope.digest()?.as_str())?;
    bytes.push(match manifest.tokenizer_family {
        InferenceTokenizerFamilyV1::Cl100kBaseApproxV1 => 0,
    });
    bytes.push(match manifest.cache_policy {
        InferenceCachePolicyV1::AnthropicExplicitCacheControlV1 => 0,
    });
    let immutable = manifest
        .segments
        .iter()
        .filter(|segment| segment.mutability == InferenceMutabilityV1::Immutable)
        .collect::<Vec<_>>();
    bytes.extend_from_slice(
        &u16::try_from(immutable.len())
            .map_err(|_| ValidationError::new("too many immutable inference segments"))?
            .to_be_bytes(),
    );
    for segment in immutable {
        bytes.extend_from_slice(&segment.position.to_be_bytes());
        bytes.push(match segment.role {
            InferenceSegmentRoleV1::System => 0,
            InferenceSegmentRoleV1::Developer => 1,
            InferenceSegmentRoleV1::User => 2,
            InferenceSegmentRoleV1::Assistant => 3,
            InferenceSegmentRoleV1::ToolCall => 4,
            InferenceSegmentRoleV1::ToolResult => 5,
            InferenceSegmentRoleV1::ToolSchema => 6,
            InferenceSegmentRoleV1::RequestControl => 7,
        });
        push_guard_text(&mut bytes, segment.segment_id.as_str())?;
        push_guard_text(
            &mut bytes,
            segment.content_or_representation_id.scope_digest.as_str(),
        )?;
        push_guard_text(
            &mut bytes,
            segment.content_or_representation_id.content_digest.as_str(),
        )?;
        push_guard_text(&mut bytes, segment.guard_hash.as_str())?;
        bytes.extend_from_slice(&segment.token_count.to_be_bytes());
        bytes.push(match segment.delivery_state {
            InferenceDeliveryStateV1::CacheMarked => 0,
            InferenceDeliveryStateV1::Uncached => 1,
        });
        bytes.push(
            u8::try_from(segment.source_lineage.len())
                .map_err(|_| ValidationError::new("too much inference lineage"))?,
        );
        for source in &segment.source_lineage {
            push_guard_text(&mut bytes, source.scope_digest.as_str())?;
            push_guard_text(&mut bytes, source.content_digest.as_str())?;
        }
    }
    sha256(&bytes)
}

fn push_guard_text(bytes: &mut Vec<u8>, value: &str) -> Result<(), ValidationError> {
    bytes.extend_from_slice(
        &u32::try_from(value.len())
            .map_err(|_| ValidationError::new("inference guard field is too large"))?
            .to_be_bytes(),
    );
    bytes.extend_from_slice(value.as_bytes());
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InferencePlanV1 {
    pub manifest_id: ScopedContentIdV1,
    pub transitions: Vec<InferenceSegmentTransitionV1>,
}

impl InferencePlanV1 {
    fn validate(&self, scope: &ViaScopeV1) -> Result<(), ValidationError> {
        if self.transitions.is_empty() || self.transitions.len() > MAX_VIA_INFERENCE_SEGMENTS {
            return Err(ValidationError::new(format!(
                "InferencePlanV1 transitions must be within 1..={MAX_VIA_INFERENCE_SEGMENTS}"
            )));
        }
        let scope_digest = scope.digest()?;
        if self.manifest_id.scope_digest != scope_digest {
            return Err(ValidationError::new(
                "InferencePlanV1 manifest_id must match plan scope",
            ));
        }
        let mut prior_position = None;
        let mut old_segments = BTreeSet::new();
        let mut new_segments = BTreeSet::new();
        for transition in &self.transitions {
            if prior_position.is_some_and(|position| transition.position <= position) {
                return Err(ValidationError::new(
                    "InferencePlanV1 transitions must be strictly position-sorted",
                ));
            }
            prior_position = Some(transition.position);
            if !old_segments.insert(&transition.old_segment_id)
                || !new_segments.insert(&transition.new_segment_id)
            {
                return Err(ValidationError::new(
                    "InferencePlanV1 transition segment IDs must be unique",
                ));
            }
            if transition.old_representation_id.scope_digest != scope_digest
                || transition.new_representation_id.scope_digest != scope_digest
                || transition.old_guard_hash != transition.old_representation_id.content_digest
                || transition.new_guard_hash != transition.new_representation_id.content_digest
                || transition.old_segment_id == transition.new_segment_id
                || transition.old_representation_id == transition.new_representation_id
                || transition.new_token_count >= transition.old_token_count
            {
                return Err(ValidationError::new(
                    "InferencePlanV1 transition identity or token reduction is invalid",
                ));
            }
        }
        Ok(())
    }
}

impl ViaPlanRequestV1 {
    pub fn validate(&self) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        validate_plan_ttl(&self.now, &self.expires_at)?;
        if let Some(candidates) = &self.inference_candidates {
            candidates.validate_for_scope(&self.stream_id.scope_digest)?;
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "plan_type", rename_all = "snake_case")]
pub enum ViaPlanBodyV1 {
    Read { source_id: ScopedContentIdV1 },
    Search { index_id: ScopedContentIdV1 },
    Shell { output_id: ScopedContentIdV1 },
    Mcp { output_id: ScopedContentIdV1 },
    Reuse { source_ids: Vec<ScopedContentIdV1> },
    Inference { inference_plan: InferencePlanV1 },
}

impl ViaPlanBodyV1 {
    fn source_ids(&self) -> Vec<&ScopedContentIdV1> {
        match self {
            Self::Read { source_id } => vec![source_id],
            Self::Search { index_id } => vec![index_id],
            Self::Shell { output_id } | Self::Mcp { output_id } => vec![output_id],
            Self::Reuse { source_ids } => source_ids.iter().collect(),
            Self::Inference { inference_plan } => vec![&inference_plan.manifest_id],
        }
    }
}

#[derive(Deserialize)]
struct ViaOptimizationPlanV1Wire {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    protocol_version: u16,
    scope: ViaScopeV1,
    header: ViaPlanHeaderV1,
    #[serde(flatten)]
    body: ViaPlanBodyV1,
}

struct StrictPlanJsonValue(Value);

impl<'de> Deserialize<'de> for StrictPlanJsonValue {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        deserializer.deserialize_any(StrictPlanJsonVisitor)
    }
}

struct StrictPlanJsonVisitor;

impl<'de> Visitor<'de> for StrictPlanJsonVisitor {
    type Value = StrictPlanJsonValue;

    fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("a JSON plan value with unique object keys")
    }

    fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::Bool(value)))
    }

    fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::Number(value.into())))
    }

    fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::Number(value.into())))
    }

    fn visit_f64<E>(self, value: f64) -> Result<Self::Value, E>
    where
        E: serde::de::Error,
    {
        serde_json::Number::from_f64(value)
            .map(Value::Number)
            .map(StrictPlanJsonValue)
            .ok_or_else(|| E::custom("non-finite JSON number in Via plan"))
    }

    fn visit_str<E>(self, value: &str) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::String(value.to_owned())))
    }

    fn visit_string<E>(self, value: String) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::String(value)))
    }

    fn visit_none<E>(self) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::Null))
    }

    fn visit_unit<E>(self) -> Result<Self::Value, E> {
        Ok(StrictPlanJsonValue(Value::Null))
    }

    fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
    where
        A: SeqAccess<'de>,
    {
        let mut values = Vec::new();
        while let Some(value) = seq.next_element::<StrictPlanJsonValue>()? {
            values.push(value.0);
        }
        Ok(StrictPlanJsonValue(Value::Array(values)))
    }

    fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
    where
        A: MapAccess<'de>,
    {
        let mut object = Map::new();
        while let Some(key) = map.next_key::<String>()? {
            if object.contains_key(&key) {
                return Err(serde::de::Error::custom(format!(
                    "duplicate JSON object key {key:?}"
                )));
            }
            let value = map.next_value::<StrictPlanJsonValue>()?;
            object.insert(key, value.0);
        }
        Ok(StrictPlanJsonValue(Value::Object(object)))
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ViaOptimizationPlanV1 {
    #[serde(deserialize_with = "deserialize_via_protocol_version")]
    pub protocol_version: u16,
    pub scope: ViaScopeV1,
    pub header: ViaPlanHeaderV1,
    #[serde(flatten)]
    pub body: ViaPlanBodyV1,
}

impl<'de> Deserialize<'de> for ViaOptimizationPlanV1 {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = StrictPlanJsonValue::deserialize(deserializer)?.0;
        let object = value
            .as_object()
            .ok_or_else(|| DeError::custom("ViaOptimizationPlanV1 must be an object"))?;
        let plan_type = object
            .get("plan_type")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| DeError::custom("ViaOptimizationPlanV1 requires plan_type"))?;
        let variant_field = match plan_type {
            "read" => "source_id",
            "search" => "index_id",
            "shell" | "mcp" => "output_id",
            "reuse" => "source_ids",
            "inference" => "inference_plan",
            other => {
                return Err(DeError::custom(format!(
                    "ViaOptimizationPlanV1 has unsupported plan_type {other}"
                )));
            }
        };
        let unknown = object
            .keys()
            .filter(|field| {
                !matches!(
                    field.as_str(),
                    "protocol_version" | "scope" | "header" | "plan_type"
                ) && field.as_str() != variant_field
            })
            .map(String::as_str)
            .collect::<Vec<_>>();
        if !unknown.is_empty() {
            return Err(DeError::custom(format!(
                "ViaOptimizationPlanV1 contains unknown top-level fields: {}",
                unknown.join(", ")
            )));
        }
        let wire =
            serde_json::from_value::<ViaOptimizationPlanV1Wire>(value).map_err(DeError::custom)?;
        Ok(Self {
            protocol_version: wire.protocol_version,
            scope: wire.scope,
            header: wire.header,
            body: wire.body,
        })
    }
}

impl ViaOptimizationPlanV1 {
    pub fn validate_at(&self, now: &UtcTimestamp) -> Result<(), ValidationError> {
        validate_protocol_version(self.protocol_version)?;
        self.scope.validate()?;
        let scope_digest = self.scope.digest()?;
        if self.header.output_budget_tokens == 0
            || self.header.output_budget_tokens > MAX_VIA_OUTPUT_TOKENS
        {
            return Err(ValidationError::new(format!(
                "ViaPlanHeaderV1 output_budget_tokens must be within 1..={MAX_VIA_OUTPUT_TOKENS}"
            )));
        }
        validate_unique(
            &self.header.expected_source_ids,
            MAX_VIA_PLAN_SOURCES,
            "ViaPlanHeaderV1 expected_source_ids",
        )?;
        if self.header.expected_source_ids.is_empty() {
            return Err(ValidationError::new(
                "ViaPlanHeaderV1 expected_source_ids must not be empty",
            ));
        }
        if self
            .header
            .expected_source_ids
            .iter()
            .any(|source| source.scope_digest != scope_digest)
        {
            return Err(ValidationError::new(
                "ViaPlanHeaderV1 expected_source_ids must match plan scope",
            ));
        }
        validate_plan_ttl(now, &self.header.expires_at)?;
        let expected = self
            .header
            .expected_source_ids
            .iter()
            .collect::<BTreeSet<_>>();
        let body_sources = self
            .body
            .source_ids()
            .into_iter()
            .cloned()
            .collect::<Vec<_>>();
        validate_unique(
            &body_sources,
            MAX_VIA_PLAN_SOURCES,
            "Via plan body source IDs",
        )?;
        if body_sources.is_empty() || body_sources.iter().any(|source| !expected.contains(source)) {
            return Err(ValidationError::new(
                "Via plan body sources must be covered by expected_source_ids",
            ));
        }
        if let ViaPlanBodyV1::Inference { inference_plan } = &self.body {
            inference_plan.validate(&self.scope)?;
        }
        let bytes = serde_json::to_vec(self)
            .map_err(|error| ValidationError::new(format!("serialize Via plan: {error}")))?;
        if bytes.len() > MAX_VIA_PLAN_BYTES {
            return Err(ValidationError::new(format!(
                "ViaOptimizationPlanV1 exceeds the {MAX_VIA_PLAN_BYTES} byte limit"
            )));
        }
        Ok(())
    }

    /// Validate a plan against Edge-owned state immediately before execution.
    pub fn validate_for_execution(
        &self,
        now: &UtcTimestamp,
        current_event_id: &EventId,
        current_source_ids: &[ScopedContentIdV1],
        current_guard_digest: &Sha256Digest,
    ) -> Result<(), ValidationError> {
        self.validate_at(now)?;
        if &self.header.caused_by != current_event_id {
            return Err(ValidationError::new(
                "Via optimization plan causal event is stale",
            ));
        }
        validate_unique(
            current_source_ids,
            MAX_VIA_PLAN_SOURCES,
            "current source IDs",
        )?;
        let expected = self
            .header
            .expected_source_ids
            .iter()
            .collect::<BTreeSet<_>>();
        let current = current_source_ids.iter().collect::<BTreeSet<_>>();
        if current != expected {
            return Err(ValidationError::new(
                "Via optimization plan source state is stale",
            ));
        }
        if current_guard_digest != &self.header.immutable_guard_digest {
            return Err(ValidationError::new(
                "Via optimization plan immutable guard is stale",
            ));
        }
        Ok(())
    }
}

/// Closed V1 serialization boundary between Edge and Via.
///
/// Only these protocol-owned variants can enter the canonical wire encoder.
/// Provider requests, headers, credentials, and cookie stores have no variant
/// and cannot acquire one outside this crate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(
    tag = "message_type",
    content = "message",
    rename_all = "snake_case",
    deny_unknown_fields
)]
pub enum ViaWireMessageV1 {
    Hello(ViaProtocolHelloV1),
    Event(Box<ViaEventFrameV1>),
    EventAck(ViaEventAckV1),
    Backpressure(ViaBackpressureV1),
    StreamResume(ViaStreamResumeV1),
    StreamCursor(ViaStreamCursorV1),
    ContentManifest(Box<ContentManifestV1>),
    NeedContent(NeedContentV1),
    ContentChunk(Box<ContentChunkV1>),
    ContentChunkAck(ContentChunkAckV1),
    PlanRequest(ViaPlanRequestV1),
    OptimizationPlan(Box<ViaOptimizationPlanV1>),
}

impl ViaWireMessageV1 {
    pub fn validate_at(&self, now: &UtcTimestamp) -> Result<(), ValidationError> {
        match self {
            Self::Hello(hello) => hello.validate(),
            Self::Event(event) => event.validate(),
            Self::EventAck(ack) => ack.validate(),
            Self::Backpressure(backpressure) => backpressure.validate(),
            Self::StreamResume(resume) => resume.validate(),
            Self::StreamCursor(cursor) => cursor.validate(),
            Self::ContentManifest(manifest) => manifest.validate(),
            Self::NeedContent(request) => request.validate_at(now),
            Self::ContentChunk(chunk) => chunk.validate_at(now),
            Self::ContentChunkAck(ack) => ack.validate(),
            Self::PlanRequest(request) => request.validate(),
            Self::OptimizationPlan(plan) => plan.validate_at(now),
        }
    }

    pub fn encode_at(&self, now: &UtcTimestamp) -> Result<Vec<u8>, ValidationError> {
        self.validate_at(now)?;
        let bytes = serde_json::to_vec(self)
            .map_err(|error| ValidationError::new(format!("serialize Via V1 message: {error}")))?;
        if bytes.len() > MAX_VIA_WIRE_BYTES {
            return Err(ValidationError::new(format!(
                "ViaWireMessageV1 exceeds the {MAX_VIA_WIRE_BYTES} byte limit"
            )));
        }
        if matches!(self, Self::PlanRequest(_)) && bytes.len() > MAX_VIA_PLAN_REQUEST_BYTES {
            return Err(ValidationError::new(format!(
                "Via plan request exceeds the {MAX_VIA_PLAN_REQUEST_BYTES} byte limit"
            )));
        }
        Ok(bytes)
    }

    pub fn decode_at(bytes: &[u8], now: &UtcTimestamp) -> Result<Self, ValidationError> {
        if bytes.len() > MAX_VIA_WIRE_BYTES {
            return Err(ValidationError::new(format!(
                "ViaWireMessageV1 exceeds the {MAX_VIA_WIRE_BYTES} byte limit"
            )));
        }
        let message = serde_json::from_slice::<Self>(bytes)
            .map_err(|error| ValidationError::new(format!("decode Via V1 message: {error}")))?;
        if matches!(message, Self::PlanRequest(_)) && bytes.len() > MAX_VIA_PLAN_REQUEST_BYTES {
            return Err(ValidationError::new(format!(
                "Via plan request exceeds the {MAX_VIA_PLAN_REQUEST_BYTES} byte limit"
            )));
        }
        message.validate_at(now)?;
        Ok(message)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const A: &str = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
    const B: &str = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
    const C: &str = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
    const FORBIDDEN_AUTH_FIELDS: &[&str] = &[
        "authorization",
        "cookie",
        "api_key",
        "api-key",
        "x-api-key",
        "x-goog-api-key",
        "x-oauth-token",
        "x-xai-token-auth",
        "oauth_token",
        "access_token",
        "refresh_token",
        "session_token",
        "client_secret",
        "credentials",
        "provider_headers",
        "raw_provider_request",
    ];

    fn id<T>(value: &str) -> T
    where
        T: TryFrom<String>,
        <T as TryFrom<String>>::Error: std::fmt::Debug,
    {
        T::try_from(value.to_owned()).unwrap()
    }

    fn digest(value: &str) -> Sha256Digest {
        Sha256Digest::new(value).unwrap()
    }

    fn scope(classification: ViaClassificationV1) -> ViaScopeV1 {
        ViaScopeV1 {
            protocol_version: VIA_PROTOCOL_V1,
            event_or_request_id: id("request-1"),
            causality_id: Some(id("cause-1")),
            user_id: Some(id("user-1")),
            device_id: Some(id("device-1")),
            project_id: Some(id("project-1")),
            workspace_id: Some(id("workspace-1")),
            session_id: Some(id("session-1")),
            conversation_id: Some(id("conversation-1")),
            agent_id: Some(id("agent-1")),
            team_id: None,
            organization_id: None,
            classification,
        }
    }

    fn content_id(scope: &ViaScopeV1) -> ScopedContentIdV1 {
        ScopedContentIdV1 {
            scope_digest: scope.digest().unwrap(),
            content_digest: digest(B),
        }
    }

    fn stream_id(scope: &ViaScopeV1) -> ScopedReferenceV1 {
        ScopedReferenceV1::new(scope, id("stream-1")).unwrap()
    }

    fn valid_manifest(
        scope: ViaScopeV1,
        kind: ContentKindV1,
        availability: ContentAvailabilityV1,
    ) -> ContentManifestV1 {
        let content_id = content_id(&scope);
        ContentManifestV1 {
            scope,
            content_id,
            kind,
            byte_count: 4,
            token_counts: vec![],
            source_version: None,
            language: None,
            structural_manifest: None,
            chunks: vec![ChunkDescriptorV1 {
                chunk_id: digest(A),
                start_byte: 0,
                end_byte: 4,
                byte_count: 4,
                symbol: None,
                token_count: Some(1),
                operator_id: id("operator:chunk"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
            }],
            availability,
        }
    }

    fn valid_plan(scope: ViaScopeV1, body: ViaPlanBodyV1) -> ViaOptimizationPlanV1 {
        ViaOptimizationPlanV1 {
            protocol_version: VIA_PROTOCOL_V1,
            header: ViaPlanHeaderV1 {
                plan_id: id("plan-wire-1"),
                caused_by: id("event-wire-1"),
                policy_id: id("policy:wire"),
                policy_version: SemanticVersion::new("1.0.0").unwrap(),
                expected_source_ids: vec![content_id(&scope)],
                immutable_guard_digest: digest(A),
                operator_id: id("operator:wire"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
                expires_at: UtcTimestamp::new("2026-08-26T17:05:00Z").unwrap(),
                output_budget_tokens: 1_000,
            },
            scope,
            body,
        }
    }

    fn assert_observed_kind_is_covered(kind: ViaObservedEventKindV1) {
        match kind {
            ViaObservedEventKindV1::SessionStarted
            | ViaObservedEventKindV1::SessionEnded
            | ViaObservedEventKindV1::AgentObserved
            | ViaObservedEventKindV1::UserMessageObserved
            | ViaObservedEventKindV1::AssistantMessageObserved
            | ViaObservedEventKindV1::FileObserved
            | ViaObservedEventKindV1::FileReadObserved
            | ViaObservedEventKindV1::FileChanged
            | ViaObservedEventKindV1::SearchExecuted
            | ViaObservedEventKindV1::SearchResultObserved
            | ViaObservedEventKindV1::ShellExecuted
            | ViaObservedEventKindV1::ShellResultObserved
            | ViaObservedEventKindV1::McpToolCalled
            | ViaObservedEventKindV1::McpResultObserved
            | ViaObservedEventKindV1::ProviderRequestPending
            | ViaObservedEventKindV1::ContextDelivered
            | ViaObservedEventKindV1::ContextReused
            | ViaObservedEventKindV1::ContextInvalidated => {}
        }
    }

    fn assert_manifest_enums_are_covered(
        classification: ViaClassificationV1,
        kind: ContentKindV1,
        availability: ContentAvailabilityV1,
    ) {
        match classification {
            ViaClassificationV1::Public
            | ViaClassificationV1::Normal
            | ViaClassificationV1::Sensitive
            | ViaClassificationV1::Secret
            | ViaClassificationV1::LocalOnly
            | ViaClassificationV1::EnterprisePrivate => {}
        }
        match kind {
            ContentKindV1::SourceFile
            | ContentKindV1::Message
            | ContentKindV1::SearchResult
            | ContentKindV1::ShellResult
            | ContentKindV1::McpResult
            | ContentKindV1::InferenceManifest => {}
        }
        match availability {
            ContentAvailabilityV1::EdgeOnly
            | ContentAvailabilityV1::TransientRemoteAllowed
            | ContentAvailabilityV1::DeniedRemote => {}
        }
    }

    #[test]
    fn identity_scope_round_trips_with_reserved_agent_fields() {
        let value = scope(ViaClassificationV1::Normal);
        let json = serde_json::to_string(&value).unwrap();
        assert_eq!(serde_json::from_str::<ViaScopeV1>(&json).unwrap(), value);
        assert!(json.contains("conversation_id"));
        assert!(json.contains("agent_id"));
    }

    #[test]
    fn negotiation_selects_v1_and_rejects_no_overlap() {
        let hello = ViaProtocolHelloV1 {
            minimum_version: 1,
            maximum_version: 2,
            capabilities: vec![ViaCapabilityV1::EventStream],
        };
        assert_eq!(hello.negotiate(&hello).unwrap(), VIA_PROTOCOL_V1);
        let future = ViaProtocolHelloV1 {
            minimum_version: 2,
            maximum_version: 3,
            capabilities: vec![],
        };
        assert!(hello.negotiate(&future).is_err());
        let peer = ViaProtocolHelloV1 {
            minimum_version: 1,
            maximum_version: 1,
            capabilities: vec![ViaCapabilityV1::TypedPlans, ViaCapabilityV1::EventStream],
        };
        assert_eq!(
            hello.shared_capabilities(&peer).unwrap(),
            vec![ViaCapabilityV1::EventStream]
        );
    }

    #[test]
    fn manifest_is_metadata_only_bounded_and_classification_safe() {
        let scope = scope(ViaClassificationV1::Secret);
        let mut manifest = ContentManifestV1 {
            content_id: content_id(&scope),
            scope,
            kind: ContentKindV1::SourceFile,
            byte_count: 4,
            token_counts: vec![],
            source_version: None,
            language: Some("rust".into()),
            structural_manifest: None,
            chunks: vec![ChunkDescriptorV1 {
                chunk_id: digest(A),
                start_byte: 0,
                end_byte: 4,
                byte_count: 4,
                symbol: None,
                token_count: Some(1),
                operator_id: id("operator:chunk"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
            }],
            availability: ContentAvailabilityV1::TransientRemoteAllowed,
        };
        assert!(manifest.validate().is_err());
        manifest.availability = ContentAvailabilityV1::EdgeOnly;
        manifest.validate().unwrap();
        manifest.content_id.scope_digest = digest(A);
        assert!(manifest.validate().is_err());
        manifest.content_id.scope_digest = manifest.scope.digest().unwrap();
        let json = serde_json::to_string(&manifest).unwrap();
        assert!(!json.contains("bytes"));
    }

    #[test]
    fn event_hash_is_deterministic_and_detects_mutation() {
        let scope = scope(ViaClassificationV1::Normal);
        let payload = ViaEventPayloadV1::Observation(ObservationPayloadV1 {
            kind: ViaObservedEventKindV1::FileObserved,
            input_ids: vec![],
            output_ids: vec![content_id(&scope)],
            capability_id: None,
        });
        let mut frame = ViaEventFrameV1::new(
            stream_id(&scope),
            1,
            id("event-1"),
            vec![],
            scope,
            payload.clone(),
        )
        .unwrap();
        assert_eq!(frame.payload_hash, payload.digest().unwrap());
        assert_eq!(
            serde_json::to_vec(&frame).unwrap(),
            serde_json::to_vec(&frame).unwrap()
        );
        frame.payload = ViaEventPayloadV1::Plan(PlanEventPayloadV1 {
            plan_id: id("plan-1"),
            status: PlanEventStatusV1::Applied,
        });
        assert!(frame.validate().is_err());
    }

    #[test]
    fn event_decoder_rejects_unsupported_version() {
        let scope = scope(ViaClassificationV1::Normal);
        let payload = ViaEventPayloadV1::Observation(ObservationPayloadV1 {
            kind: ViaObservedEventKindV1::SessionStarted,
            input_ids: vec![],
            output_ids: vec![],
            capability_id: None,
        });
        let frame =
            ViaEventFrameV1::new(stream_id(&scope), 0, id("event-1"), vec![], scope, payload)
                .unwrap();
        let mut value = serde_json::to_value(frame).unwrap();
        value["protocol_version"] = serde_json::json!(2);
        assert!(serde_json::from_value::<ViaEventFrameV1>(value).is_err());
    }

    #[test]
    fn typed_plan_rejects_stale_or_unexpected_sources() {
        let scope = scope(ViaClassificationV1::Normal);
        let expected = content_id(&scope);
        let mut plan = ViaOptimizationPlanV1 {
            protocol_version: VIA_PROTOCOL_V1,
            scope,
            header: ViaPlanHeaderV1 {
                plan_id: id("plan-1"),
                caused_by: id("event-1"),
                policy_id: id("policy:read"),
                policy_version: SemanticVersion::new("1.0.0").unwrap(),
                expected_source_ids: vec![expected.clone()],
                immutable_guard_digest: digest(A),
                operator_id: id("operator:read"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
                expires_at: UtcTimestamp::new("2026-08-26T17:05:00Z").unwrap(),
                output_budget_tokens: 1_000,
            },
            body: ViaPlanBodyV1::Read {
                source_id: expected.clone(),
            },
        };
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        plan.validate_at(&now).unwrap();
        plan.validate_for_execution(
            &now,
            &id("event-1"),
            std::slice::from_ref(&expected),
            &digest(A),
        )
        .unwrap();
        assert!(
            plan.validate_for_execution(
                &now,
                &id("event-1"),
                std::slice::from_ref(&expected),
                &digest(B),
            )
            .is_err()
        );
        assert!(
            plan.validate_for_execution(&now, &id("event-1"), &[], &digest(A))
                .is_err()
        );
        assert!(
            plan.validate_for_execution(
                &now,
                &id("event-other"),
                std::slice::from_ref(&expected),
                &digest(A),
            )
            .is_err()
        );
        let expired = UtcTimestamp::new("2026-08-26T17:05:00Z").unwrap();
        assert!(plan.validate_at(&expired).is_err());
        plan.body = ViaPlanBodyV1::Reuse {
            source_ids: vec![expected.clone(), expected],
        };
        assert!(plan.validate_at(&now).is_err());
        plan.body = ViaPlanBodyV1::Read {
            source_id: ScopedContentIdV1 {
                scope_digest: digest(B),
                content_digest: digest(A),
            },
        };
        assert!(plan.validate_at(&now).is_err());
    }

    #[test]
    fn scoped_content_identity_depends_on_scope_and_bytes_not_path() {
        let primary_scope = scope(ViaClassificationV1::Normal);
        let other = scope(ViaClassificationV1::Sensitive);
        let first = ScopedContentIdV1::from_scope_and_bytes(&primary_scope, b"same bytes").unwrap();
        let second =
            ScopedContentIdV1::from_scope_and_bytes(&primary_scope, b"same bytes").unwrap();
        let other_scope = ScopedContentIdV1::from_scope_and_bytes(&other, b"same bytes").unwrap();
        assert_eq!(first, second);
        assert_ne!(first, other_scope);
    }

    #[test]
    fn manifest_rejects_gaps_and_incomplete_coverage() {
        let chunk = |chunk_id, start_byte, end_byte| ChunkDescriptorV1 {
            chunk_id: digest(chunk_id),
            start_byte,
            end_byte,
            byte_count: end_byte - start_byte,
            symbol: None,
            token_count: None,
            operator_id: id("operator:chunk"),
            operator_version: SemanticVersion::new("1.0.0").unwrap(),
        };
        let scope = scope(ViaClassificationV1::Normal);
        let mut manifest = ContentManifestV1 {
            content_id: content_id(&scope),
            scope,
            kind: ContentKindV1::SourceFile,
            byte_count: 8,
            token_counts: vec![],
            source_version: None,
            language: None,
            structural_manifest: None,
            chunks: vec![chunk(A, 0, 4), chunk(B, 5, 8)],
            availability: ContentAvailabilityV1::EdgeOnly,
        };
        assert!(manifest.validate().is_err());
        manifest.chunks = vec![chunk(A, 0, 4)];
        assert!(manifest.validate().is_err());
        manifest.chunks = vec![chunk(A, 0, 4), chunk(B, 4, 8)];
        manifest.validate().unwrap();

        manifest.byte_count = 300;
        manifest.chunks = (0_u64..300)
            .map(|offset| ChunkDescriptorV1 {
                chunk_id: sha256(&offset.to_be_bytes()).unwrap(),
                start_byte: offset,
                end_byte: offset + 1,
                byte_count: 1,
                symbol: Some("x".repeat(1_024)),
                token_count: None,
                operator_id: id("operator:chunk"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
            })
            .collect();
        assert!(manifest.validate().is_err());
    }

    #[test]
    fn event_rejects_duplicate_self_causality_and_invalid_successor() {
        let scope = scope(ViaClassificationV1::Normal);
        let payload = || {
            ViaEventPayloadV1::Observation(ObservationPayloadV1 {
                kind: ViaObservedEventKindV1::SessionStarted,
                input_ids: vec![],
                output_ids: vec![],
                capability_id: None,
            })
        };
        let first = ViaEventFrameV1::new(
            stream_id(&scope),
            7,
            id("event-1"),
            vec![],
            scope.clone(),
            payload(),
        )
        .unwrap();
        let next = ViaEventFrameV1::new(
            stream_id(&scope),
            8,
            id("event-2"),
            vec![id("event-1")],
            scope.clone(),
            payload(),
        )
        .unwrap();
        next.validate_successor(&first).unwrap();

        let terminal = ViaEventFrameV1::new(
            stream_id(&scope),
            u64::MAX,
            id("event-terminal"),
            vec![],
            scope.clone(),
            payload(),
        )
        .unwrap();
        let terminal_replay = ViaEventFrameV1::new(
            stream_id(&scope),
            u64::MAX,
            id("event-replay"),
            vec![],
            scope,
            payload(),
        )
        .unwrap();
        assert!(terminal_replay.validate_successor(&terminal).is_err());

        let mut invalid = next.clone();
        invalid.stream_id.scope_digest = digest(A);
        assert!(invalid.validate().is_err());
        invalid = next.clone();
        invalid.seq = 9;
        assert!(invalid.validate_successor(&first).is_err());
        invalid = next.clone();
        invalid.causality = vec![id("event-1"), id("event-1")];
        assert!(invalid.validate().is_err());
        invalid.causality = vec![id("event-2")];
        assert!(invalid.validate().is_err());
    }

    #[test]
    fn stream_ack_and_resume_cursor_are_exactly_bound() {
        let scope = scope(ViaClassificationV1::Normal);
        let stream_id = stream_id(&scope);
        let event_id: EventId = id("event-ack-7");
        let ack = ViaEventAckV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id.clone(),
            acknowledged_seq: 7,
            acknowledged_event_id: event_id.clone(),
            next_seq: 8,
            status: ViaEventAckStatusV1::Stored,
            accepted_event_count: 8,
            graph_nodes: 2,
            graph_edges: 1,
        };
        ack.validate().unwrap();
        let mut invalid_ack = ack.clone();
        invalid_ack.next_seq = 9;
        assert!(invalid_ack.validate().is_err());
        invalid_ack = ack.clone();
        invalid_ack.accepted_event_count = 0;
        assert!(invalid_ack.validate().is_err());
        invalid_ack = ack.clone();
        invalid_ack.accepted_event_count = 9;
        assert!(invalid_ack.validate().is_err());
        invalid_ack = ack.clone();
        invalid_ack.graph_edges = MAX_VIA_ACK_GRAPH_EDGES + 1;
        assert!(invalid_ack.validate().is_err());

        let backpressure = ViaBackpressureV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id.clone(),
            accepted_window: MAX_VIA_STREAM_WINDOW,
            retry_after_ms: MAX_VIA_BACKPRESSURE_RETRY_MS,
        };
        backpressure.validate().unwrap();
        let mut invalid_backpressure = backpressure.clone();
        invalid_backpressure.accepted_window = MAX_VIA_STREAM_WINDOW + 1;
        assert!(invalid_backpressure.validate().is_err());
        invalid_backpressure = backpressure;
        invalid_backpressure.retry_after_ms = 0;
        assert!(invalid_backpressure.validate().is_err());

        ViaStreamResumeV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id.clone(),
        }
        .validate()
        .unwrap();
        ViaStreamCursorV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id.clone(),
            next_seq: 0,
            last_accepted: None,
        }
        .validate()
        .unwrap();
        let cursor = ViaStreamCursorV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id,
            next_seq: 8,
            last_accepted: Some(ViaStreamPositionV1 { seq: 7, event_id }),
        };
        cursor.validate().unwrap();
        let mut invalid_cursor = cursor;
        invalid_cursor.next_seq = 7;
        assert!(invalid_cursor.validate().is_err());
    }

    #[test]
    fn content_on_demand_is_request_bound_bounded_and_debug_redacted() {
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        let expires_at = UtcTimestamp::new("2026-08-26T17:00:30Z").unwrap();
        let scope = scope(ViaClassificationV1::Normal);
        let source_content_id = ScopedContentIdV1::from_scope_and_bytes(&scope, b"hello").unwrap();
        let request_id = ScopedReferenceV1::new(&scope, id("content-request-1")).unwrap();
        let chunk_id = sha256(b"hello").unwrap();
        let request = NeedContentV1 {
            protocol_version: VIA_PROTOCOL_V1,
            request_id: request_id.clone(),
            source_content_id: source_content_id.clone(),
            chunk_ids: vec![chunk_id.clone()],
            purpose: NeedContentPurposeV1::ContextPlanning,
            max_bytes: 5,
            max_tokens: 5,
            expires_at: expires_at.clone(),
            policy_version: SemanticVersion::new("1.0.0").unwrap(),
        };
        request.validate_at(&now).unwrap();
        let mut invalid_request = request.clone();
        invalid_request.chunk_ids = vec![chunk_id.clone(); 2];
        assert!(invalid_request.validate_at(&now).is_err());
        invalid_request = request.clone();
        invalid_request.max_bytes = MAX_VIA_CONTENT_REQUEST_BYTES + 1;
        assert!(invalid_request.validate_at(&now).is_err());
        invalid_request = request.clone();
        invalid_request.expires_at = UtcTimestamp::new("2026-08-26T17:00:31Z").unwrap();
        assert!(invalid_request.validate_at(&now).is_err());
        assert!(request.validate_at(&expires_at).is_err());

        let chunk = ContentChunkV1 {
            protocol_version: VIA_PROTOCOL_V1,
            request_id: request_id.clone(),
            source_content_id: source_content_id.clone(),
            chunk_id: chunk_id.clone(),
            start_byte: 0,
            end_byte: 5,
            byte_count: 5,
            source_version: sha256(b"hello").unwrap(),
            content: "hello".to_owned(),
            payload_digest: chunk_id.clone(),
            expires_at,
        };
        chunk.validate_at(&now).unwrap();
        assert!(!format!("{chunk:?}").contains("hello"));
        let mut over_ttl = chunk.clone();
        over_ttl.expires_at = UtcTimestamp::new("2026-08-26T17:00:31Z").unwrap();
        assert!(over_ttl.validate_at(&now).is_err());
        let mut altered = chunk.clone();
        altered.content = "jello".to_owned();
        assert!(altered.validate_at(&now).is_err());

        let ack = ContentChunkAckV1 {
            protocol_version: VIA_PROTOCOL_V1,
            request_id,
            source_content_id,
            chunk_id,
        };
        ack.validate().unwrap();
        for message in [
            ViaWireMessageV1::NeedContent(request),
            ViaWireMessageV1::ContentChunk(Box::new(chunk)),
            ViaWireMessageV1::ContentChunkAck(ack),
        ] {
            let encoded = message.encode_at(&now).unwrap();
            assert_eq!(
                ViaWireMessageV1::decode_at(&encoded, &now).unwrap(),
                message
            );
        }
    }

    #[test]
    fn negotiation_and_plan_bounds_fail_closed() {
        let hello = ViaProtocolHelloV1 {
            minimum_version: 1,
            maximum_version: 1,
            capabilities: vec![ViaCapabilityV1::EventStream, ViaCapabilityV1::EventStream],
        };
        assert!(hello.validate().is_err());

        let scope = scope(ViaClassificationV1::Normal);
        let expected = content_id(&scope);
        let mut plan = ViaOptimizationPlanV1 {
            protocol_version: VIA_PROTOCOL_V1,
            scope,
            header: ViaPlanHeaderV1 {
                plan_id: id("plan-1"),
                caused_by: id("event-1"),
                policy_id: id("policy:read"),
                policy_version: SemanticVersion::new("1.0.0").unwrap(),
                expected_source_ids: vec![expected.clone()],
                immutable_guard_digest: digest(A),
                operator_id: id("operator:read"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
                expires_at: UtcTimestamp::new("2026-08-26T17:05:00Z").unwrap(),
                output_budget_tokens: 0,
            },
            body: ViaPlanBodyV1::Read {
                source_id: expected,
            },
        };
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        assert!(plan.validate_at(&now).is_err());
        plan.header.output_budget_tokens = MAX_VIA_OUTPUT_TOKENS + 1;
        assert!(plan.validate_at(&now).is_err());
        plan.header.output_budget_tokens = 1;
        plan.header.expires_at = UtcTimestamp::new("2026-08-26T17:05:01Z").unwrap();
        assert!(plan.validate_at(&now).is_err());
    }

    #[test]
    fn wire_gate_covers_all_v1_variants_and_drops_or_rejects_auth_injection() {
        const CANARIES: &[&str] = &[
            "via-auth-canary-never-wire",
            "via-cookie-canary-never-wire",
            "via-oauth-canary-never-wire",
            "via-refresh-canary-never-wire",
            "via-client-secret-canary-never-wire",
        ];
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        let normal_scope = scope(ViaClassificationV1::Normal);
        let source = content_id(&normal_scope);
        let mut messages = vec![ViaWireMessageV1::Hello(ViaProtocolHelloV1 {
            minimum_version: VIA_PROTOCOL_V1,
            maximum_version: VIA_PROTOCOL_V1,
            capabilities: vec![
                ViaCapabilityV1::EventStream,
                ViaCapabilityV1::ContentManifest,
                ViaCapabilityV1::ContentOnDemand,
                ViaCapabilityV1::TypedPlans,
                ViaCapabilityV1::MultiAgentIdentity,
            ],
        })];
        let stream = stream_id(&normal_scope);
        messages.extend([
            ViaWireMessageV1::EventAck(ViaEventAckV1 {
                protocol_version: VIA_PROTOCOL_V1,
                stream_id: stream.clone(),
                acknowledged_seq: 0,
                acknowledged_event_id: id("event-wire-ack"),
                next_seq: 1,
                status: ViaEventAckStatusV1::Stored,
                accepted_event_count: 1,
                graph_nodes: 0,
                graph_edges: 0,
            }),
            ViaWireMessageV1::Backpressure(ViaBackpressureV1 {
                protocol_version: VIA_PROTOCOL_V1,
                stream_id: stream.clone(),
                accepted_window: 0,
                retry_after_ms: 100,
            }),
            ViaWireMessageV1::StreamResume(ViaStreamResumeV1 {
                protocol_version: VIA_PROTOCOL_V1,
                stream_id: stream.clone(),
            }),
            ViaWireMessageV1::StreamCursor(ViaStreamCursorV1 {
                protocol_version: VIA_PROTOCOL_V1,
                stream_id: stream,
                next_seq: 0,
                last_accepted: None,
            }),
        ]);

        for kind in [
            ViaObservedEventKindV1::SessionStarted,
            ViaObservedEventKindV1::SessionEnded,
            ViaObservedEventKindV1::AgentObserved,
            ViaObservedEventKindV1::UserMessageObserved,
            ViaObservedEventKindV1::AssistantMessageObserved,
            ViaObservedEventKindV1::FileObserved,
            ViaObservedEventKindV1::FileReadObserved,
            ViaObservedEventKindV1::FileChanged,
            ViaObservedEventKindV1::SearchExecuted,
            ViaObservedEventKindV1::SearchResultObserved,
            ViaObservedEventKindV1::ShellExecuted,
            ViaObservedEventKindV1::ShellResultObserved,
            ViaObservedEventKindV1::McpToolCalled,
            ViaObservedEventKindV1::McpResultObserved,
            ViaObservedEventKindV1::ProviderRequestPending,
            ViaObservedEventKindV1::ContextDelivered,
            ViaObservedEventKindV1::ContextReused,
            ViaObservedEventKindV1::ContextInvalidated,
        ] {
            assert_observed_kind_is_covered(kind);
            let payload = ViaEventPayloadV1::Observation(ObservationPayloadV1 {
                kind,
                input_ids: vec![source.clone()],
                output_ids: vec![],
                capability_id: None,
            });
            messages.push(ViaWireMessageV1::Event(Box::new(
                ViaEventFrameV1::new(
                    stream_id(&normal_scope),
                    1,
                    id("event-wire-1"),
                    vec![],
                    normal_scope.clone(),
                    payload,
                )
                .unwrap(),
            )));
        }

        for status in [PlanEventStatusV1::Applied, PlanEventStatusV1::Rejected] {
            let payload = ViaEventPayloadV1::Plan(PlanEventPayloadV1 {
                plan_id: id("plan-wire-1"),
                status,
            });
            messages.push(ViaWireMessageV1::Event(Box::new(
                ViaEventFrameV1::new(
                    stream_id(&normal_scope),
                    1,
                    id("event-wire-1"),
                    vec![],
                    normal_scope.clone(),
                    payload,
                )
                .unwrap(),
            )));
        }

        for (classification, kind) in [
            (ViaClassificationV1::Public, ContentKindV1::SourceFile),
            (ViaClassificationV1::Normal, ContentKindV1::Message),
            (ViaClassificationV1::Sensitive, ContentKindV1::SearchResult),
            (ViaClassificationV1::Secret, ContentKindV1::ShellResult),
            (ViaClassificationV1::LocalOnly, ContentKindV1::McpResult),
            (
                ViaClassificationV1::EnterprisePrivate,
                ContentKindV1::InferenceManifest,
            ),
        ] {
            assert_manifest_enums_are_covered(
                classification,
                kind,
                ContentAvailabilityV1::EdgeOnly,
            );
            messages.push(ViaWireMessageV1::ContentManifest(Box::new(valid_manifest(
                scope(classification),
                kind,
                ContentAvailabilityV1::EdgeOnly,
            ))));
        }
        messages.push(ViaWireMessageV1::ContentManifest(Box::new(valid_manifest(
            normal_scope.clone(),
            ContentKindV1::SourceFile,
            ContentAvailabilityV1::TransientRemoteAllowed,
        ))));
        messages.push(ViaWireMessageV1::ContentManifest(Box::new(valid_manifest(
            normal_scope.clone(),
            ContentKindV1::SourceFile,
            ContentAvailabilityV1::DeniedRemote,
        ))));

        for body in [
            ViaPlanBodyV1::Read {
                source_id: source.clone(),
            },
            ViaPlanBodyV1::Search {
                index_id: source.clone(),
            },
            ViaPlanBodyV1::Shell {
                output_id: source.clone(),
            },
            ViaPlanBodyV1::Mcp {
                output_id: source.clone(),
            },
            ViaPlanBodyV1::Reuse {
                source_ids: vec![source.clone()],
            },
            ViaPlanBodyV1::Inference {
                inference_plan: InferencePlanV1 {
                    manifest_id: source.clone(),
                    transitions: vec![InferenceSegmentTransitionV1 {
                        position: 0,
                        old_segment_id: digest(A),
                        old_representation_id: source.clone(),
                        old_guard_hash: source.content_digest.clone(),
                        new_segment_id: digest(
                            "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
                        ),
                        new_representation_id: ScopedContentIdV1::from_scope_and_bytes(
                            &normal_scope,
                            b"new",
                        )
                        .unwrap(),
                        new_guard_hash: ScopedContentIdV1::from_scope_and_bytes(
                            &normal_scope,
                            b"new",
                        )
                        .unwrap()
                        .content_digest,
                        old_token_count: 2,
                        new_token_count: 1,
                    }],
                },
            },
        ] {
            messages.push(ViaWireMessageV1::OptimizationPlan(Box::new(valid_plan(
                normal_scope.clone(),
                body,
            ))));
        }
        messages.push(ViaWireMessageV1::PlanRequest(ViaPlanRequestV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id(&normal_scope),
            event_id: id("event-wire-1"),
            immutable_guard_digest: digest(A),
            now: now.clone(),
            expires_at: UtcTimestamp::new("2026-08-26T17:00:30Z").unwrap(),
            inference_candidates: None,
        }));

        for message in &messages {
            let encoded = message.encode_at(&now).unwrap();
            assert_eq!(
                ViaWireMessageV1::decode_at(&encoded, &now).unwrap(),
                *message
            );
            let wire = String::from_utf8(encoded).unwrap();
            for canary in CANARIES {
                assert!(!wire.contains(canary));
            }
            for field in FORBIDDEN_AUTH_FIELDS {
                assert!(!wire.contains(&format!("\"{field}\"")));
            }
        }

        for (index, field) in FORBIDDEN_AUTH_FIELDS.iter().enumerate() {
            let canary = format!("via-{index}-credential-canary-never-wire");
            let mut injected = serde_json::to_value(&messages[0]).unwrap();
            injected[*field] = serde_json::json!(&canary);
            let injected = serde_json::to_vec(&injected).unwrap();
            assert!(ViaWireMessageV1::decode_at(&injected, &now).is_err());

            let mut nested = serde_json::to_value(&messages[0]).unwrap();
            nested["message"][*field] = serde_json::json!(&canary);
            let nested = serde_json::to_vec(&nested).unwrap();
            assert!(ViaWireMessageV1::decode_at(&nested, &now).is_err());
        }
    }

    #[test]
    fn representative_wire_values_have_no_provider_authority_fields() {
        fn assert_safe_keys(value: &serde_json::Value) {
            match value {
                serde_json::Value::Object(fields) => {
                    for (key, value) in fields {
                        assert!(!FORBIDDEN_AUTH_FIELDS.contains(&key.as_str()));
                        assert_safe_keys(value);
                    }
                }
                serde_json::Value::Array(values) => {
                    values.iter().for_each(assert_safe_keys);
                }
                _ => {}
            }
        }

        let scope = scope(ViaClassificationV1::Sensitive);
        let payload = ViaEventPayloadV1::Observation(ObservationPayloadV1 {
            kind: ViaObservedEventKindV1::ProviderRequestPending,
            input_ids: vec![content_id(&scope)],
            output_ids: vec![],
            capability_id: Some(id("provider-call")),
        });
        let frame =
            ViaEventFrameV1::new(stream_id(&scope), 1, id("event-1"), vec![], scope, payload)
                .unwrap();
        assert_safe_keys(&serde_json::to_value(frame).unwrap());
    }

    #[test]
    fn event_and_plan_decoders_reject_unknown_authority_fields() {
        let scope = scope(ViaClassificationV1::Normal);
        let payload = ViaEventPayloadV1::Observation(ObservationPayloadV1 {
            kind: ViaObservedEventKindV1::SessionStarted,
            input_ids: vec![],
            output_ids: vec![],
            capability_id: None,
        });
        let frame = ViaEventFrameV1::new(
            stream_id(&scope),
            1,
            id("event-1"),
            vec![],
            scope.clone(),
            payload,
        )
        .unwrap();
        let frame_value = serde_json::to_value(&frame).unwrap();
        assert!(serde_json::from_value::<ViaEventFrameV1>(frame_value.clone()).is_ok());

        let expected = content_id(&scope);
        let plan = ViaOptimizationPlanV1 {
            protocol_version: VIA_PROTOCOL_V1,
            scope,
            header: ViaPlanHeaderV1 {
                plan_id: id("plan-1"),
                caused_by: id("event-1"),
                policy_id: id("policy:read"),
                policy_version: SemanticVersion::new("1.0.0").unwrap(),
                expected_source_ids: vec![expected.clone()],
                immutable_guard_digest: digest(A),
                operator_id: id("operator:read"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
                expires_at: UtcTimestamp::new("2026-08-26T17:05:00Z").unwrap(),
                output_budget_tokens: 1_000,
            },
            body: ViaPlanBodyV1::Read {
                source_id: expected,
            },
        };
        let plan_value = serde_json::to_value(&plan).unwrap();
        let decoded_plan = serde_json::from_value::<ViaOptimizationPlanV1>(plan_value.clone());
        assert!(decoded_plan.is_ok(), "{decoded_plan:?}");

        for field in FORBIDDEN_AUTH_FIELDS {
            let mut frame_with_unknown = frame_value.clone();
            frame_with_unknown[*field] = serde_json::json!("sensitive");
            assert!(
                serde_json::from_value::<ViaEventFrameV1>(frame_with_unknown).is_err(),
                "ViaEventFrameV1 must reject top-level {field}"
            );

            let mut plan_with_unknown = plan_value.clone();
            plan_with_unknown[*field] = serde_json::json!("sensitive");
            assert!(
                serde_json::from_value::<ViaOptimizationPlanV1>(plan_with_unknown).is_err(),
                "ViaOptimizationPlanV1 must reject top-level {field}"
            );
        }
    }

    #[test]
    fn inference_manifest_is_deterministic_metadata_only_and_strictly_bound() {
        const PROMPT_CANARY: &[u8] = b"provider-key-and-raw-prompt-canary";
        let scope = scope(ViaClassificationV1::Sensitive);
        let segment = |position, role, tokens| {
            InferenceSegmentV1::from_canonical_bytes(
                &scope,
                position,
                role,
                InferenceMutabilityV1::Immutable,
                PROMPT_CANARY,
                tokens,
                InferenceDeliveryStateV1::Uncached,
                Vec::new(),
            )
            .unwrap()
        };
        let first = InferenceManifestV1::new(
            scope.clone(),
            vec![
                segment(0, InferenceSegmentRoleV1::System, 4),
                segment(1, InferenceSegmentRoleV1::User, 7),
            ],
            None,
        )
        .unwrap();
        let second = InferenceManifestV1::new(
            scope.clone(),
            vec![
                segment(0, InferenceSegmentRoleV1::System, 4),
                segment(1, InferenceSegmentRoleV1::User, 7),
            ],
            None,
        )
        .unwrap();
        assert_eq!(first, second);
        assert_eq!(first.accounting.total_tokens, 11);
        let encoded = serde_json::to_vec(&first).unwrap();
        assert!(
            !encoded
                .windows(PROMPT_CANARY.len())
                .any(|window| window == PROMPT_CANARY)
        );

        let mut tampered = first.clone();
        tampered.segments[1].position = 0;
        assert!(tampered.validate().is_err());
        let mut tampered = first.clone();
        tampered.segments[0].guard_hash = digest(A);
        assert!(tampered.validate().is_err());
        let mut tampered = first.clone();
        tampered.accounting.total_tokens += 1;
        assert!(tampered.validate().is_err());
        let mut unknown = serde_json::to_value(first).unwrap();
        unknown["provider_api_key"] = serde_json::json!("credential");
        assert!(serde_json::from_value::<InferenceManifestV1>(unknown).is_err());
    }

    #[test]
    fn inference_manifest_bounds_and_overflow_fail_closed() {
        let scope = scope(ViaClassificationV1::Normal);
        assert!(InferenceManifestV1::new(scope.clone(), Vec::new(), None).is_err());
        let lineage = (0..=MAX_VIA_INFERENCE_LINEAGE)
            .map(|index| {
                ScopedContentIdV1::from_scope_and_bytes(&scope, index.to_string().as_bytes())
                    .unwrap()
            })
            .collect();
        assert!(
            InferenceSegmentV1::from_canonical_bytes(
                &scope,
                0,
                InferenceSegmentRoleV1::User,
                InferenceMutabilityV1::Immutable,
                b"bounded",
                1,
                InferenceDeliveryStateV1::Uncached,
                lineage,
            )
            .is_err()
        );
        let maximum = InferenceSegmentV1::from_canonical_bytes(
            &scope,
            0,
            InferenceSegmentRoleV1::System,
            InferenceMutabilityV1::Immutable,
            b"maximum",
            u64::MAX,
            InferenceDeliveryStateV1::Uncached,
            Vec::new(),
        )
        .unwrap();
        let overflow = InferenceSegmentV1::from_canonical_bytes(
            &scope,
            1,
            InferenceSegmentRoleV1::User,
            InferenceMutabilityV1::Immutable,
            b"overflow",
            1,
            InferenceDeliveryStateV1::Uncached,
            Vec::new(),
        )
        .unwrap();
        assert!(InferenceManifestV1::new(scope, vec![maximum, overflow], None).is_err());
    }

    fn inference_plan_fixture() -> ViaOptimizationPlanV1 {
        let scope = ViaScopeV1 {
            protocol_version: VIA_PROTOCOL_V1,
            event_or_request_id: id("request-inference-plan"),
            causality_id: None,
            user_id: None,
            device_id: None,
            project_id: None,
            workspace_id: None,
            session_id: None,
            conversation_id: None,
            agent_id: None,
            team_id: None,
            organization_id: None,
            classification: ViaClassificationV1::Normal,
        };
        let scope_digest = scope.digest().unwrap();
        let scoped = |content_digest| ScopedContentIdV1 {
            scope_digest: scope_digest.clone(),
            content_digest,
        };
        let manifest_id = scoped(digest(B));
        ViaOptimizationPlanV1 {
            protocol_version: VIA_PROTOCOL_V1,
            scope,
            header: ViaPlanHeaderV1 {
                plan_id: id("plan-inference-1"),
                caused_by: id("event-inference-pending"),
                policy_id: id("via.inference-planning"),
                policy_version: SemanticVersion::new("1.0.0").unwrap(),
                expected_source_ids: vec![manifest_id.clone()],
                immutable_guard_digest: digest(A),
                operator_id: id("leanctx.inference.history.compact"),
                operator_version: SemanticVersion::new("1.0.0").unwrap(),
                expires_at: UtcTimestamp::new("2026-08-26T17:00:30Z").unwrap(),
                output_budget_tokens: 8,
            },
            body: ViaPlanBodyV1::Inference {
                inference_plan: InferencePlanV1 {
                    manifest_id: manifest_id.clone(),
                    transitions: vec![InferenceSegmentTransitionV1 {
                        position: 3,
                        old_segment_id: digest(A),
                        old_representation_id: manifest_id,
                        old_guard_hash: digest(B),
                        new_segment_id: digest(
                            "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
                        ),
                        new_representation_id: scoped(digest(
                            "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
                        )),
                        new_guard_hash: digest(
                            "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
                        ),
                        old_token_count: 10,
                        new_token_count: 5,
                    }],
                },
            },
        }
    }

    #[test]
    #[allow(clippy::useless_concat)]
    fn inference_plan_wire_is_canonical_metadata_only_and_strict() {
        const FIXTURE: &str = concat!(
            r#"{"protocol_version":1,"scope":{"protocol_version":1,"event_or_request_id":"request-inference-plan","classification":"normal"},"header":{"plan_id":"plan-inference-1","caused_by":"event-inference-pending","policy_id":"via.inference-planning","policy_version":"1.0.0","expected_source_ids":[{"scope_digest":"sha256:4c6e93595932ddbe4418c020f30fe5fd321c4567ac17f5703644125647ccdcd0","content_digest":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}],"immutable_guard_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","operator_id":"leanctx.inference.history.compact","operator_version":"1.0.0","expires_at":"2026-08-26T17:00:30Z","output_budget_tokens":8},"plan_type":"inference","inference_plan":{"manifest_id":{"scope_digest":"sha256:4c6e93595932ddbe4418c020f30fe5fd321c4567ac17f5703644125647ccdcd0","content_digest":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},"transitions":[{"position":3,"old_segment_id":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","old_representation_id":{"scope_digest":"sha256:4c6e93595932ddbe4418c020f30fe5fd321c4567ac17f5703644125647ccdcd0","content_digest":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},"old_guard_hash":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","new_segment_id":"sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","new_representation_id":{"scope_digest":"sha256:4c6e93595932ddbe4418c020f30fe5fd321c4567ac17f5703644125647ccdcd0","content_digest":"sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"},"new_guard_hash":"sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd","old_token_count":10,"new_token_count":5}]}}"#,
        );
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        let plan = inference_plan_fixture();
        plan.validate_at(&now).unwrap();
        assert_eq!(serde_json::to_string(&plan).unwrap(), FIXTURE);
        assert_eq!(
            serde_json::from_str::<ViaOptimizationPlanV1>(FIXTURE).unwrap(),
            plan
        );
        for canary in ["provider-key", "raw-prompt", "authorization", "cookie"] {
            assert!(!FIXTURE.contains(canary));
        }
    }

    #[test]
    fn inference_plan_bounds_and_hostile_shapes_fail_closed() {
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        let valid = inference_plan_fixture();

        let mutate = |mut plan: ViaOptimizationPlanV1, change: fn(&mut InferencePlanV1)| {
            let ViaPlanBodyV1::Inference { inference_plan } = &mut plan.body else {
                unreachable!()
            };
            change(inference_plan);
            assert!(plan.validate_at(&now).is_err());
        };
        mutate(valid.clone(), |body| body.transitions.clear());
        mutate(valid.clone(), |body| {
            body.transitions[0].new_token_count = body.transitions[0].old_token_count;
        });
        mutate(valid.clone(), |body| {
            body.transitions.push(body.transitions[0].clone());
        });
        mutate(valid.clone(), |body| {
            body.transitions[0].new_guard_hash = digest(A);
        });

        let mut unknown = serde_json::to_value(valid).unwrap();
        unknown["inference_plan"]["raw_provider_request"] =
            serde_json::json!("provider-key-and-prompt-canary");
        assert!(serde_json::from_value::<ViaOptimizationPlanV1>(unknown).is_err());
        let encoded = serde_json::to_string(&inference_plan_fixture()).unwrap();
        let duplicate = encoded.replacen(
            "\"plan_type\":\"inference\"",
            "\"plan_type\":\"inference\",\"plan_type\":\"inference\"",
            1,
        );
        assert!(serde_json::from_str::<ViaOptimizationPlanV1>(&duplicate).is_err());
    }

    fn inference_candidate_set_fixture(scope: &ViaScopeV1) -> InferenceCandidateSetV1 {
        let scope_digest = scope.digest().unwrap();
        let scoped = |content_digest| ScopedContentIdV1 {
            scope_digest: scope_digest.clone(),
            content_digest,
        };
        let mut candidates = InferenceCandidateSetV1 {
            candidate_set_id: scoped(digest(A)),
            manifest_id: scoped(digest(B)),
            operator_id: id("leanctx.inference.history.compact"),
            operator_version: SemanticVersion::new("1.0.0").unwrap(),
            transitions: vec![InferenceSegmentTransitionV1 {
                position: 4,
                old_segment_id: digest(A),
                old_representation_id: scoped(digest(B)),
                old_guard_hash: digest(B),
                new_segment_id: digest(C),
                new_representation_id: scoped(digest(C)),
                new_guard_hash: digest(C),
                old_token_count: 10,
                new_token_count: 5,
            }],
        };
        candidates.candidate_set_id = candidates.expected_id(&scope_digest).unwrap();
        candidates
    }

    #[test]
    fn inference_candidate_set_wire_is_exact_metadata_only_and_strict() {
        let scope = scope(ViaClassificationV1::Normal);
        let candidates = inference_candidate_set_fixture(&scope);
        candidates
            .validate_for_scope(&scope.digest().unwrap())
            .unwrap();
        assert_eq!(
            candidates.candidate_set_id.content_digest.as_str(),
            "sha256:06a8f76a1d6118e7508364a6b5f72bebcde7a851d07a23c49e2ef0108f35fda9"
        );
        let value = serde_json::to_value(&candidates).unwrap();
        assert_eq!(
            value.as_object().unwrap().keys().collect::<Vec<_>>(),
            vec![
                "candidate_set_id",
                "manifest_id",
                "operator_id",
                "operator_version",
                "transitions"
            ]
        );
        let request = ViaPlanRequestV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id(&scope),
            event_id: id("event-candidates"),
            immutable_guard_digest: digest(A),
            now: UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap(),
            expires_at: UtcTimestamp::new("2026-08-26T17:00:30Z").unwrap(),
            inference_candidates: Some(candidates.clone()),
        };
        request.validate().unwrap();
        let encoded = serde_json::to_string(&request).unwrap();
        const REQUEST_FIXTURE: &str = r#"{"protocol_version":1,"stream_id":{"scope_digest":"sha256:261c81d042ed5b69e3ff4d24e632714d700d39ca9cd6e1295565c12a15d7124d","reference":"stream-1"},"event_id":"event-candidates","immutable_guard_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","now":"2026-08-26T17:00:00Z","expires_at":"2026-08-26T17:00:30Z","inference_candidates":{"candidate_set_id":{"scope_digest":"sha256:261c81d042ed5b69e3ff4d24e632714d700d39ca9cd6e1295565c12a15d7124d","content_digest":"sha256:06a8f76a1d6118e7508364a6b5f72bebcde7a851d07a23c49e2ef0108f35fda9"},"manifest_id":{"scope_digest":"sha256:261c81d042ed5b69e3ff4d24e632714d700d39ca9cd6e1295565c12a15d7124d","content_digest":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},"operator_id":"leanctx.inference.history.compact","operator_version":"1.0.0","transitions":[{"position":4,"old_segment_id":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","old_representation_id":{"scope_digest":"sha256:261c81d042ed5b69e3ff4d24e632714d700d39ca9cd6e1295565c12a15d7124d","content_digest":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},"old_guard_hash":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","new_segment_id":"sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","new_representation_id":{"scope_digest":"sha256:261c81d042ed5b69e3ff4d24e632714d700d39ca9cd6e1295565c12a15d7124d","content_digest":"sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"},"new_guard_hash":"sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","old_token_count":10,"new_token_count":5}]}}"#;
        assert_eq!(encoded, REQUEST_FIXTURE);
        assert_eq!(
            serde_json::from_str::<ViaPlanRequestV1>(&encoded).unwrap(),
            request
        );
        for canary in ["provider-key", "raw-prompt", "authorization", "cookie"] {
            assert!(!encoded.contains(canary));
        }
        let mut unknown = serde_json::to_value(&candidates).unwrap();
        unknown["raw_prompt"] = serde_json::json!("prompt-canary");
        assert!(serde_json::from_value::<InferenceCandidateSetV1>(unknown).is_err());
        let duplicate = serde_json::to_string(&candidates).unwrap().replacen(
            "\"operator_id\":\"leanctx.inference.history.compact\"",
            "\"operator_id\":\"leanctx.inference.history.compact\",\"operator_id\":\"leanctx.inference.history.compact\"",
            1,
        );
        assert!(serde_json::from_str::<InferenceCandidateSetV1>(&duplicate).is_err());
    }

    #[test]
    fn inference_candidate_set_limits_hash_and_scope_reject_hostile() {
        let scope = scope(ViaClassificationV1::Normal);
        let valid = inference_candidate_set_fixture(&scope);
        let scope_digest = scope.digest().unwrap();
        let mut wrong_scope = valid.clone();
        wrong_scope.manifest_id.scope_digest = digest(A);
        assert!(wrong_scope.validate_for_scope(&scope_digest).is_err());
        let mut wrong_request_scope = ViaPlanRequestV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id(&scope),
            event_id: id("event-candidates"),
            immutable_guard_digest: digest(A),
            now: UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap(),
            expires_at: UtcTimestamp::new("2026-08-26T17:00:30Z").unwrap(),
            inference_candidates: Some(valid.clone()),
        };
        wrong_request_scope.stream_id.scope_digest = digest(A);
        assert!(wrong_request_scope.validate().is_err());

        let mut wrong_operator = valid.clone();
        wrong_operator.operator_id = id("operator:unknown");
        assert!(wrong_operator.validate_for_scope(&scope_digest).is_err());

        let mut wrong_hash = valid.clone();
        wrong_hash.transitions[0].new_guard_hash = digest(A);
        assert!(wrong_hash.validate_for_scope(&scope_digest).is_err());

        let mut unsorted = valid.clone();
        unsorted.transitions.push(valid.transitions[0].clone());
        assert!(unsorted.validate_for_scope(&scope_digest).is_err());

        let mut maximum = valid.clone();
        maximum.transitions = (0..MAX_VIA_INFERENCE_CANDIDATES)
            .map(|position| {
                let mut transition = maximum.transitions[0].clone();
                transition.position = position as u16;
                transition.old_segment_id = digest(&format!("sha256:{position:064x}"));
                transition.new_segment_id = digest(&format!("sha256:{:064x}", position + 1));
                transition
            })
            .collect();
        maximum.candidate_set_id = maximum.expected_id(&scope_digest).unwrap();
        maximum.validate_for_scope(&scope_digest).unwrap();
        let now = UtcTimestamp::new("2026-08-26T17:00:00Z").unwrap();
        let request = ViaPlanRequestV1 {
            protocol_version: VIA_PROTOCOL_V1,
            stream_id: stream_id(&scope),
            event_id: id("event-candidates-maximum"),
            immutable_guard_digest: digest(A),
            now: now.clone(),
            expires_at: UtcTimestamp::new("2026-08-26T17:00:30Z").unwrap(),
            inference_candidates: Some(maximum),
        };
        let encoded = ViaWireMessageV1::PlanRequest(request)
            .encode_at(&now)
            .unwrap();
        assert!(encoded.len() <= MAX_VIA_PLAN_REQUEST_BYTES);
        let mut over_byte_cap = encoded;
        over_byte_cap.resize(MAX_VIA_PLAN_REQUEST_BYTES + 1, b' ');
        assert!(ViaWireMessageV1::decode_at(&over_byte_cap, &now).is_err());

        let mut oversized = valid;
        oversized.transitions = (0..MAX_VIA_INFERENCE_CANDIDATES + 1)
            .map(|position| {
                let mut transition = oversized.transitions[0].clone();
                transition.position = position as u16;
                transition.old_segment_id = digest(&format!("sha256:{position:064x}"));
                transition.new_segment_id = digest(&format!("sha256:{:064x}", position + 1));
                transition
            })
            .collect();
        assert!(oversized.validate_for_scope(&scope_digest).is_err());
    }
}