redevplugin_ipc/

lib.rs

use base64::Engine as _;
use ed25519_dalek::{Signature, Verifier, VerifyingKey};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet};
use std::sync::OnceLock;

mod contract_set_gen;

pub use contract_set_gen::CONTRACT_SET_SHA256;

pub const RUST_IPC_VERSION: &str = "rust-ipc-v6";
pub const WASM_ABI_VERSION: &str = "redevplugin-wasm-worker-v2";
pub const RUNTIME_LEASE_SIGNATURE_SCHEMA_VERSION: &str = "redevplugin.runtime_execution_lease.v1";

#[cfg(test)]
fn contract_fixture(id: redevplugin_contracts::ContractId) -> &'static str {
    std::str::from_utf8(redevplugin_contracts::get(id).bytes())
        .expect("generated contracts are valid UTF-8")
}
pub const RUNTIME_LEASE_TOKEN_KIND: &str = "runtime_execution_lease";
pub const RUNTIME_LEASE_SIGNATURE_ALGORITHM: &str = "ed25519";
pub const WORKER_INVOCATION_TARGET_SCHEMA_VERSION: &str = "redevplugin.worker_invocation_target.v1";
pub const MAX_RUNTIME_LEASE_MEMORY_BYTES: u64 = 256 * 1024 * 1024;
pub const MAX_JSON_SAFE_INTEGER: u64 = (1_u64 << 53) - 1;
pub const MIN_RUNTIME_WORKER_COUNT: usize = 1;
pub const MAX_RUNTIME_WORKER_COUNT: usize = 64;
pub const MIN_RUNTIME_QUEUE_CAPACITY: usize = 1;
pub const MAX_RUNTIME_QUEUE_CAPACITY: usize = 64;
pub const MIN_RUNTIME_PER_PLUGIN_CONCURRENCY: usize = 1;
pub const MAX_RUNTIME_PER_PLUGIN_CONCURRENCY: usize = 64;
pub const MIN_RUNTIME_MODULE_CACHE_ENTRIES: usize = 1;
pub const MAX_RUNTIME_MODULE_CACHE_ENTRIES: usize = 1024;
pub const MIN_RUNTIME_MODULE_CACHE_SOURCE_BYTES: usize = 1;
pub const MAX_RUNTIME_MODULE_CACHE_SOURCE_BYTES: usize = 128 * 1024 * 1024;
pub const FRAME_TYPE_HELLO: &str = "hello";
pub const FRAME_TYPE_HELLO_ACK: &str = "hello_ack";
pub const FRAME_TYPE_HEARTBEAT: &str = "heartbeat";
pub const FRAME_TYPE_INVOKE_WORKER: &str = "invoke_worker";
pub const FRAME_TYPE_INVOKE_WORKER_RESULT: &str = "invoke_worker_result";
pub const FRAME_TYPE_CANCEL_INVOKE: &str = "cancel_invoke";
pub const FRAME_TYPE_CANCEL_INVOKE_ACK: &str = "cancel_invoke_ack";
pub const FRAME_TYPE_COMPILE_FLIGHT_REGISTER: &str = "compile_flight_register";
pub const FRAME_TYPE_COMPILE_FLIGHT_COMPLETE: &str = "compile_flight_complete";
pub const FRAME_TYPE_OPEN_HANDLE: &str = "open_handle";
pub const FRAME_TYPE_VALIDATE_HANDLE_GRANT: &str = "validate_handle_grant";
pub const FRAME_TYPE_STORAGE_FILE: &str = "storage_file";
pub const FRAME_TYPE_STORAGE_KV: &str = "storage_kv";
pub const FRAME_TYPE_STORAGE_SQLITE: &str = "storage_sqlite";
pub const FRAME_TYPE_NETWORK_GRANT: &str = "network_grant";
pub const FRAME_TYPE_NETWORK_EXECUTE: &str = "network_execute";
pub const FRAME_TYPE_REVOKE_EPOCH: &str = "revoke_epoch";
pub const FRAME_TYPE_REVOKE_EPOCH_ACK: &str = "revoke_epoch_ack";
pub const FRAME_TYPE_SESSION_REVOKE: &str = "session_revoke";
pub const FRAME_TYPE_SESSION_REVOKE_ACK: &str = "session_revoke_ack";
pub const ERR_ARTIFACT_HANDLE_FAILED: &str = "ARTIFACT_HANDLE_FAILED";
pub const ERR_HANDLE_GRANT_VALIDATION_FAILED: &str = "HANDLE_GRANT_VALIDATION_FAILED";
pub const ERR_STORAGE_FILE_FAILED: &str = "STORAGE_FILE_FAILED";
pub const ERR_STORAGE_KV_FAILED: &str = "STORAGE_KV_FAILED";
pub const ERR_STORAGE_SQLITE_FAILED: &str = "STORAGE_SQLITE_FAILED";
pub const ERR_NETWORK_GRANT_FAILED: &str = "NETWORK_GRANT_FAILED";
pub const ERR_NETWORK_EXECUTE_FAILED: &str = "NETWORK_EXECUTE_FAILED";
pub const ERR_NETWORK_STREAM_STORE_UNAVAILABLE: &str = "NETWORK_STREAM_STORE_UNAVAILABLE";
pub const ERR_NETWORK_STREAM_FAILED: &str = "NETWORK_STREAM_FAILED";
pub const ERR_NETWORK_STREAM_BACKPRESSURE: &str = "NETWORK_STREAM_BACKPRESSURE";
pub const ERR_NETWORK_STREAM_INVALID: &str = "NETWORK_STREAM_INVALID";
pub const ERR_NETWORK_STREAM_NOT_FOUND: &str = "NETWORK_STREAM_NOT_FOUND";
pub const ERR_NETWORK_STREAM_CLOSED: &str = "NETWORK_STREAM_CLOSED";
pub const ERR_WORKER_INVOCATION_INVALID: &str = "WORKER_INVOCATION_INVALID";
pub const ERR_RUNTIME_CAPABILITY_REVOKED: &str = "RUNTIME_CAPABILITY_REVOKED";
pub const ERR_RUNTIME_CONTROL_CHANNEL_STALE: &str = "RUNTIME_CONTROL_CHANNEL_STALE";
pub const ERR_RUNTIME_LEASE_INVALID: &str = "RUNTIME_LEASE_INVALID";
pub const ERR_RUNTIME_LEASE_SIGNATURE_INVALID: &str = "RUNTIME_LEASE_SIGNATURE_INVALID";
pub const ERR_LEASE_REPLAYED: &str = "PLUGIN_LEASE_REPLAYED";
pub const ERR_WASM_WORKER_INVALID: &str = "WASM_WORKER_INVALID";
pub const ERR_WASM_WORKER_FAILED: &str = "WASM_WORKER_FAILED";
pub const ERR_WASM_HOSTCALL_FAILED: &str = "WASM_HOSTCALL_FAILED";
pub const ERR_RUNTIME_CAPACITY_EXCEEDED: &str = "RUNTIME_CAPACITY_EXCEEDED";
pub const ERR_RUNTIME_INVOCATION_CANCELED: &str = "RUNTIME_INVOCATION_CANCELED";
pub const ERR_SESSION_REVOKED: &str = "PLUGIN_SESSION_REVOKED";
pub const ERR_SESSION_REVOKE_SEQUENCE_STALE: &str = "SESSION_REVOKE_SEQUENCE_STALE";
pub const ERR_SESSION_REVOKE_DRAIN_TIMEOUT: &str = "SESSION_REVOKE_DRAIN_TIMEOUT";
pub const ERR_UNSUPPORTED_FRAME: &str = "UNSUPPORTED_FRAME";
pub const ERROR_ORIGIN_RUNTIME: &str = "runtime";
pub const ERROR_ORIGIN_HOSTCALL: &str = "hostcall";
pub const ERROR_ORIGIN_PLUGIN: &str = "plugin";

#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum IpcError {
    DecodeFailed { context: &'static str },
    EncodeFailed { context: &'static str },
    MissingField { field: &'static str },
    InvalidField { field: &'static str },
    ProtocolViolation { message: &'static str },
    CapacityOverflow { capacity: &'static str },
    RemoteFailure { code: String },
    InvalidResponseResultJson,
    EmptyResponseErrorCode,
    EmptyResponseErrorMessage,
}

pub type IpcResult<T> = Result<T, IpcError>;

impl std::fmt::Display for IpcError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::DecodeFailed { context } => write!(formatter, "failed to decode {context}"),
            Self::EncodeFailed { context } => write!(formatter, "failed to encode {context}"),
            Self::MissingField { field } => write!(formatter, "missing {field}"),
            Self::InvalidField { field } => write!(formatter, "invalid {field}"),
            Self::ProtocolViolation { message } => formatter.write_str(message),
            Self::CapacityOverflow { capacity } => write!(formatter, "{capacity} overflows usize"),
            Self::RemoteFailure { code } => {
                write!(formatter, "hostcall response failed with code {code}")
            }
            Self::InvalidResponseResultJson => {
                formatter.write_str("runtime response result must be valid JSON")
            }
            Self::EmptyResponseErrorCode => {
                formatter.write_str("runtime response code is required")
            }
            Self::EmptyResponseErrorMessage => {
                formatter.write_str("runtime response message is required")
            }
        }
    }
}

impl std::error::Error for IpcError {}

fn decode_failed(context: &'static str) -> IpcError {
    IpcError::DecodeFailed { context }
}

fn encode_failed(context: &'static str) -> IpcError {
    IpcError::EncodeFailed { context }
}

fn missing_field(field: &'static str) -> IpcError {
    IpcError::MissingField { field }
}

fn invalid_field(field: &'static str) -> IpcError {
    IpcError::InvalidField { field }
}

fn protocol_violation(message: &'static str) -> IpcError {
    IpcError::ProtocolViolation { message }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FrameType {
    Hello,
    HelloAck,
    Heartbeat,
    InvokeWorker,
    InvokeWorkerResult,
    CompileFlightRegister,
    CompileFlightComplete,
    OpenHandle,
    ValidateHandleGrant,
    StorageFile,
    StorageKV,
    StorageSQLite,
    NetworkGrant,
    NetworkExecute,
    RevokeEpoch,
    RevokeEpochAck,
    SessionRevoke,
    SessionRevokeAck,
    Diagnostic,
}

#[cfg(test)]
mod property_gates {
    use super::*;
    use proptest::prelude::*;

    proptest! {
        #[test]
        fn ipc_frame_parser_is_total(input in any::<String>()) {
            let parsed = std::panic::catch_unwind(|| {
                let _ = decode_runtime_input_frame(&input);
                let _ = parse_frame_identity(&input);
                let _ = parse_hello_frame(&input);
                let _ = validate_hello_frame(&input);
                let _ = parse_worker_invocation(&input);
                let _ = parse_worker_invocation_context(&input);
                let _ = parse_worker_invocation_identity(&input);
                let _ = parse_worker_lease_replay_key(&input);
                let _ = parse_worker_response_v2(&input);
                let _ = parse_heartbeat_request(&input);
                let _ = parse_revoke_epoch_request(&input);
                let _ = parse_session_revoke_request(&input);
                let _ = parse_cancel_invoke(&input);
                let _ = parse_runtime_lease_public_keys(&input);
                let _ = bind_parent_request_id(&input, "parent_request");
            });
            prop_assert!(parsed.is_ok());
        }

        #[test]
        fn response_frame_builders_are_total(
            frame_type in any::<String>(),
            request_id in any::<String>(),
            runtime_generation_id in any::<String>(),
            result_json in any::<String>(),
            code in any::<String>(),
            message in any::<String>(),
        ) {
            let success = std::panic::catch_unwind(|| {
                success_response_frame(
                    &frame_type,
                    &request_id,
                    &runtime_generation_id,
                    &result_json,
                )
            });
            prop_assert!(success.is_ok());
            let error = std::panic::catch_unwind(|| ResponseError::runtime(&code, &message));
            prop_assert!(error.is_ok());
            if let Ok(Ok(error)) = error {
                let frame = std::panic::catch_unwind(|| {
                    error_response_frame(
                        &frame_type,
                        &request_id,
                        &runtime_generation_id,
                        error,
                    )
                });
                prop_assert!(frame.is_ok());
            }
        }

        #[test]
        fn session_revoke_ack_builder_is_total(
            request_id in any::<String>(),
            runtime_generation_id in any::<String>(),
            sequence in any::<u64>(),
            queued_invocations in any::<u64>(),
            running_invocations in any::<u64>(),
            storage_hostcalls in any::<u64>(),
            active_network_requests in any::<u64>(),
            sockets in any::<u64>(),
            network_streams in any::<u64>(),
        ) {
            let built = std::panic::catch_unwind(|| {
                session_revoke_ack_frame(
                    &request_id,
                    &runtime_generation_id,
                    sequence,
                    SessionRevokeState::Complete,
                    SessionRevokeAckCounts {
                        queued_invocations,
                        running_invocations,
                        storage_hostcalls,
                        active_network_requests,
                        sockets,
                        network_streams,
                    },
                )
            });
            prop_assert!(built.is_ok());
        }

        #[test]
        fn network_frame_builders_are_total(
            request_id in any::<String>(),
            runtime_generation_id in any::<String>(),
            scope_kind in any::<String>(),
            owner_env_hash in any::<String>(),
            owner_user_hash in any::<String>(),
            query_json in any::<String>(),
            headers_json in any::<String>(),
        ) {
            let resource_scope = NetworkResourceScope {
                kind: scope_kind,
                owner_env_hash,
                owner_user_hash,
            };
            let grant = NetworkGrantRequest {
                plugin_instance_id: "plugini_1".to_string(),
                active_fingerprint: "sha256:active".to_string(),
                resource_scope: resource_scope.clone(),
                runtime_instance_id: "runtime_1".to_string(),
                runtime_generation_id: runtime_generation_id.clone(),
                runtime_shard_id: "runtime_shard_1".to_string(),
                policy_revision: 1,
                management_revision: 1,
                revoke_epoch: 1,
                connector_id: "api".to_string(),
                transport: "http".to_string(),
                destination: "https://api.example.com".to_string(),
                ttl_ms: 1,
            };
            let grant_result = std::panic::catch_unwind(|| {
                network_grant_frame(&request_id, &runtime_generation_id, &grant)
            });
            prop_assert!(grant_result.is_ok());

            let execute = NetworkExecuteRequest {
                plugin_id: "com.example.worker".to_string(),
                plugin_instance_id: "plugini_1".to_string(),
                active_fingerprint: "sha256:active".to_string(),
                resource_scope,
                runtime_instance_id: "runtime_1".to_string(),
                runtime_generation_id: runtime_generation_id.clone(),
                runtime_shard_id: "runtime_shard_1".to_string(),
                policy_revision: 1,
                management_revision: 1,
                revoke_epoch: 1,
                connector_id: "api".to_string(),
                transport: "http".to_string(),
                destination: "https://api.example.com".to_string(),
                ttl_ms: 1,
                operation: "http".to_string(),
                method: "GET".to_string(),
                path: "/".to_string(),
                query_json,
                headers_json,
                message_type: String::new(),
                body_base64: String::new(),
                payload_base64: String::new(),
                max_request_bytes: 1,
                max_response_bytes: 1,
                max_chunk_bytes: 1,
                max_buffered_bytes: 1,
                timeout_ms: 1,
                stream_id: String::new(),
                stream_method: String::new(),
                stream_effect: String::new(),
                stream_execution: String::new(),
                surface_instance_id: String::new(),
                owner_session_hash: String::new(),
                owner_user_hash: String::new(),
                owner_env_hash: String::new(),
                session_channel_id_hash: String::new(),
                bridge_channel_id: String::new(),
                content_type: String::new(),
            };
            let execute_result = std::panic::catch_unwind(|| {
                network_execute_frame(&request_id, &runtime_generation_id, &execute)
            });
            prop_assert!(execute_result.is_ok());
        }

        #[test]
        fn runtime_limits_keep_derived_capacities_bounded(
            worker_count in MIN_RUNTIME_WORKER_COUNT..=MAX_RUNTIME_WORKER_COUNT,
            queue_capacity in MIN_RUNTIME_QUEUE_CAPACITY..=MAX_RUNTIME_QUEUE_CAPACITY,
            per_plugin_concurrency in MIN_RUNTIME_PER_PLUGIN_CONCURRENCY..=MAX_RUNTIME_PER_PLUGIN_CONCURRENCY,
            module_cache_entries in MIN_RUNTIME_MODULE_CACHE_ENTRIES..=MAX_RUNTIME_MODULE_CACHE_ENTRIES,
            module_cache_source_bytes in MIN_RUNTIME_MODULE_CACHE_SOURCE_BYTES..=MAX_RUNTIME_MODULE_CACHE_SOURCE_BYTES,
        ) {
            let limits = RuntimeLimits {
                worker_count,
                queue_capacity,
                per_plugin_concurrency,
                module_cache_entries,
                module_cache_source_bytes,
            };
            match limits.validate() {
                Ok(validated) => {
                    prop_assert!(per_plugin_concurrency <= worker_count);
                    prop_assert_eq!(validated.hostcall_active_route_capacity(), worker_count);
                    prop_assert_eq!(
                        validated.hostcall_canceled_route_capacity().unwrap(),
                        worker_count + queue_capacity,
                    );
                    prop_assert_eq!(validated.compile_flight_route_capacity(), worker_count);
                }
                Err(_) => prop_assert!(per_plugin_concurrency > worker_count),
            }
        }

        #[test]
        fn lease_signature_payload_is_stable_for_valid_fields(
            lease_id in "[a-z][a-z0-9_]{0,24}",
            token_id in "[a-z][a-z0-9_]{0,24}",
            nonce in prop::collection::vec(any::<u8>(), 16..=32),
            method in "worker\\.[a-z][a-z0-9_]{0,16}",
        ) {
            let nonce = nonce.into_iter().map(|byte| format!("{byte:02x}")).collect::<String>();
            let fixture: serde_json::Value = serde_json::from_str(include_str!(
                "../testdata/runtime-lease-signature-v1.json"
            ))
            .unwrap();
            let mut lease = fixture.get("lease").cloned().unwrap();
            lease["lease_id"] = serde_json::Value::String(lease_id);
            lease["token_id"] = serde_json::Value::String(token_id);
            lease["lease_nonce"] = serde_json::Value::String(nonce);
            lease["method"] = serde_json::Value::String(method.clone());
            let typed: WorkerLeasePayload = serde_json::from_value(lease).unwrap();
            let first = runtime_lease_signature_payload_json(&typed, method.as_str()).unwrap();
            let second = runtime_lease_signature_payload_json(&typed, method.as_str()).unwrap();
            prop_assert_eq!(&first, &second);
            let parsed: serde_json::Value = serde_json::from_str(&first).unwrap();
            prop_assert!(parsed.is_object());
            prop_assert!(parsed.get("signature").is_none());
        }
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RawIPCFrame {
    ipc_version: String,
    frame_type: String,
    request_id: String,
    parent_request_id: Option<String>,
    runtime_generation_id: Option<String>,
    payload: Box<serde_json::value::RawValue>,
}

#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct RuntimeLimits {
    pub worker_count: usize,
    pub queue_capacity: usize,
    pub per_plugin_concurrency: usize,
    pub module_cache_entries: usize,
    pub module_cache_source_bytes: usize,
}

#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct ProcessContainmentEvidence {
    pub schema_version: String,
    pub profile: String,
    pub seccomp_policy_sha256: String,
    pub no_new_privs: bool,
    pub seccomp_tsync: bool,
    pub process_creation_denied: bool,
    pub reexec_denied: bool,
    pub active: bool,
}

impl ProcessContainmentEvidence {
    pub fn validate(&self) -> IpcResult<()> {
        if self.schema_version != "redevplugin.process_containment.v1"
            || self.profile != "linux-runtime-v1"
            || self.seccomp_policy_sha256.len() != 64
            || !self
                .seccomp_policy_sha256
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
            || !self.no_new_privs
            || !self.seccomp_tsync
            || !self.process_creation_denied
            || !self.reexec_denied
            || !self.active
        {
            return Err(invalid_field("process_containment"));
        }
        Ok(())
    }
}

impl RuntimeLimits {
    pub fn validate(self) -> IpcResult<Self> {
        if self.worker_count < MIN_RUNTIME_WORKER_COUNT
            || self.queue_capacity < MIN_RUNTIME_QUEUE_CAPACITY
            || self.per_plugin_concurrency < MIN_RUNTIME_PER_PLUGIN_CONCURRENCY
            || self.module_cache_entries < MIN_RUNTIME_MODULE_CACHE_ENTRIES
            || self.module_cache_source_bytes < MIN_RUNTIME_MODULE_CACHE_SOURCE_BYTES
        {
            return Err(protocol_violation(
                "runtime limits are below platform minimums",
            ));
        }
        if self.worker_count > MAX_RUNTIME_WORKER_COUNT
            || self.queue_capacity > MAX_RUNTIME_QUEUE_CAPACITY
            || self.per_plugin_concurrency > MAX_RUNTIME_PER_PLUGIN_CONCURRENCY
            || self.module_cache_entries > MAX_RUNTIME_MODULE_CACHE_ENTRIES
            || self.module_cache_source_bytes > MAX_RUNTIME_MODULE_CACHE_SOURCE_BYTES
        {
            return Err(protocol_violation(
                "runtime limits exceed platform maximums",
            ));
        }
        if self.per_plugin_concurrency > self.worker_count {
            return Err(protocol_violation(
                "runtime per_plugin_concurrency exceeds worker_count",
            ));
        }
        self.hostcall_canceled_route_capacity()?;
        Ok(self)
    }

    pub fn hostcall_active_route_capacity(self) -> usize {
        self.worker_count
    }

    pub fn hostcall_canceled_route_capacity(self) -> IpcResult<usize> {
        self.worker_count
            .checked_add(self.queue_capacity)
            .ok_or(IpcError::CapacityOverflow {
                capacity: "runtime hostcall canceled route capacity",
            })
    }

    pub fn compile_flight_route_capacity(self) -> usize {
        self.worker_count
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct HelloPayload {
    target: String,
    host_process_id: u64,
    host_ipc_version: String,
    host_wasm_abi: String,
    contract_set_sha256: String,
    started_unix_nano: u64,
    channel_nonce: String,
    runtime_lease_public_keys: Vec<RuntimeLeasePublicKeyPayload>,
    limits: RuntimeLimits,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeTarget {
    LinuxAmd64,
    LinuxArm64,
}

impl RuntimeTarget {
    pub fn parse(value: &str) -> IpcResult<Self> {
        match value {
            "linux/amd64" => Ok(Self::LinuxAmd64),
            "linux/arm64" => Ok(Self::LinuxArm64),
            _ => Err(protocol_violation("unsupported runtime target")),
        }
    }

    pub fn as_str(&self) -> &str {
        match self {
            Self::LinuxAmd64 => "linux/amd64",
            Self::LinuxArm64 => "linux/arm64",
        }
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RuntimeLeasePublicKeyPayload {
    algorithm: String,
    key_id: String,
    public_key_base64: String,
}

#[derive(Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct WorkerFramePayload {
    lease: WorkerLeasePayload,
    method: String,
    invocation: WorkerInvocationPayload,
}

#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct WorkerLeasePayload {
    lease_id: Option<String>,
    token_id: Option<String>,
    lease_nonce: Option<String>,
    plugin_id: Option<String>,
    plugin_version: Option<String>,
    active_fingerprint: Option<String>,
    surface_instance_id: Option<String>,
    owner_session_hash: Option<String>,
    owner_user_hash: Option<String>,
    owner_env_hash: Option<String>,
    session_channel_id_hash: Option<String>,
    bridge_channel_id: Option<String>,
    runtime_generation_id: Option<String>,
    plugin_instance_id: Option<String>,
    method: Option<String>,
    effect: Option<String>,
    execution: Option<String>,
    operation_id: Option<String>,
    stream_id: Option<String>,
    audit_correlation_id: Option<String>,
    target_descriptor_hashes: Option<Vec<String>>,
    limits: Option<WorkerLeaseLimitsPayload>,
    policy_revision: Option<u64>,
    management_revision: Option<u64>,
    revoke_epoch: Option<u64>,
    runtime_shard_id: Option<String>,
    runtime_instance_id: Option<String>,
    ipc_channel_id: Option<String>,
    connection_nonce: Option<String>,
    key_id: Option<String>,
    signature: Option<String>,
    issued_at_unix_ms: Option<i64>,
    expires_at_unix_ms: Option<i64>,
}

#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct WorkerLeaseLimitsPayload {
    timeout_ms: Option<i64>,
    memory_bytes: Option<u64>,
    max_payload_bytes: Option<i64>,
    max_stream_bytes_per_sec: Option<i64>,
}

#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct WorkerInvocationPayload {
    plugin_id: Option<String>,
    plugin_instance_id: Option<String>,
    active_fingerprint: Option<String>,
    runtime_instance_id: Option<String>,
    runtime_generation_id: Option<String>,
    package_hash: Option<String>,
    worker_id: Option<String>,
    worker_mode: Option<String>,
    worker_scope: Option<String>,
    artifact: Option<String>,
    artifact_sha256: Option<String>,
    abi: Option<String>,
    method: Option<String>,
    effect: Option<String>,
    execution: Option<String>,
    surface_instance_id: Option<String>,
    owner_session_hash: Option<String>,
    owner_user_hash: Option<String>,
    owner_env_hash: Option<String>,
    session_channel_id_hash: Option<String>,
    bridge_channel_id: Option<String>,
    operation_id: Option<String>,
    stream_id: Option<String>,
    audit_correlation_id: Option<String>,
    policy_revision: Option<u64>,
    management_revision: Option<u64>,
    revoke_epoch: Option<u64>,
    params_sha256: Option<String>,
    params: Option<serde_json::Map<String, serde_json::Value>>,
    storage_handle_grants: Option<HashMap<String, String>>,
    broker_access: Option<WorkerBrokerAccessPayload>,
    broker_access_sha256: Option<String>,
}

#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct WorkerBrokerAccessPayload {
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    storage: Vec<WorkerStorageBrokerAccessPayload>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    network: Vec<WorkerNetworkBrokerAccessPayload>,
}

#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct WorkerStorageBrokerAccessPayload {
    store_id: String,
    scope: String,
    operations: Vec<String>,
}

#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct WorkerNetworkBrokerAccessPayload {
    connector_id: String,
    transport: String,
    scope: String,
    operations: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    http_methods: Vec<String>,
}

struct ClosedWorkerFrame {
    request_id: String,
    runtime_generation_id: String,
    method: String,
    lease: WorkerLeasePayload,
    invocation: WorkerInvocationPayload,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FrameIdentity {
    pub frame_type: String,
    pub request_id: String,
    pub parent_request_id: Option<String>,
    pub runtime_generation_id: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HelloFrame {
    pub request_id: String,
    pub runtime_generation_id: String,
    pub target: RuntimeTarget,
    pub contract_set_sha256: String,
    pub channel_nonce: String,
    pub runtime_lease_public_keys: Vec<RuntimeLeasePublicKey>,
    pub limits: RuntimeLimits,
}

pub struct ParsedWorkerInvocation {
    request_id: String,
    runtime_generation_id: String,
    method: String,
    lease: WorkerLeasePayload,
    invocation: WorkerInvocationPayload,
    params_json: Option<String>,
    broker_access_json: Option<String>,
    context: OnceLock<IpcResult<WorkerInvocationContext>>,
    identity: OnceLock<IpcResult<WorkerInvocationIdentity>>,
    target_hash: OnceLock<IpcResult<String>>,
}

pub struct WorkerInvocationInput {
    pub identity: FrameIdentity,
    pub invocation: IpcResult<ParsedWorkerInvocation>,
}

pub struct CancelInvocationInput {
    pub identity: FrameIdentity,
    pub invocation_request_id: String,
}

pub struct RuntimeHostcallResponseInput {
    pub identity: FrameIdentity,
    pub raw_frame: String,
}

pub enum RuntimeInputFrame {
    InvokeWorker(Box<WorkerInvocationInput>),
    CancelInvoke(CancelInvocationInput),
    HostcallResponse(RuntimeHostcallResponseInput),
    Unsupported(FrameIdentity),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkerInvocationContext {
    pub plugin_id: String,
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub method: String,
    pub effect: String,
    pub execution: String,
    pub surface_instance_id: String,
    pub owner_session_hash: String,
    pub owner_user_hash: String,
    pub owner_env_hash: String,
    pub session_channel_id_hash: String,
    pub bridge_channel_id: String,
    pub operation_id: String,
    pub stream_id: String,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
    pub storage_handle_grants: HashMap<String, String>,
    pub broker_access_json: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeartbeatRequest {
    pub sent_unix_nano: u64,
    pub max_staleness_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RevokeEpochRequest {
    pub resource_scope: NetworkResourceScope,
    pub plugin_instance_id: String,
    pub revoke_epoch: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct SessionScope {
    pub owner_session_hash: String,
    pub owner_user_hash: String,
    pub owner_env_hash: String,
    pub session_channel_id_hash: String,
}

impl SessionScope {
    pub fn new(
        owner_session_hash: impl Into<String>,
        owner_user_hash: impl Into<String>,
        owner_env_hash: impl Into<String>,
        session_channel_id_hash: impl Into<String>,
    ) -> IpcResult<Self> {
        let scope = Self {
            owner_session_hash: owner_session_hash.into(),
            owner_user_hash: owner_user_hash.into(),
            owner_env_hash: owner_env_hash.into(),
            session_channel_id_hash: session_channel_id_hash.into(),
        };
        scope.validate()?;
        Ok(scope)
    }

    fn validate(&self) -> IpcResult<()> {
        for (value, field) in [
            (&self.owner_session_hash, "owner_session_hash"),
            (&self.owner_user_hash, "owner_user_hash"),
            (&self.owner_env_hash, "owner_env_hash"),
            (&self.session_channel_id_hash, "session_channel_id_hash"),
        ] {
            if value.is_empty() || value.trim() != value {
                return Err(invalid_field(field));
            }
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionRevokeRequest {
    pub request_id: String,
    pub runtime_generation_id: String,
    pub session_revoke_sequence: u64,
    pub owner_session_hash: String,
    pub owner_user_hash: String,
    pub owner_env_hash: String,
    pub session_channel_id_hash: String,
}

impl SessionRevokeRequest {
    pub fn session_scope(&self) -> SessionScope {
        SessionScope {
            owner_session_hash: self.owner_session_hash.clone(),
            owner_user_hash: self.owner_user_hash.clone(),
            owner_env_hash: self.owner_env_hash.clone(),
            session_channel_id_hash: self.session_channel_id_hash.clone(),
        }
    }
}

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

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SessionRevokeAckCounts {
    pub queued_invocations: u64,
    pub running_invocations: u64,
    pub storage_hostcalls: u64,
    pub active_network_requests: u64,
    pub sockets: u64,
    pub network_streams: u64,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct HeartbeatRequestPayload {
    sent_unix_nano: u64,
    max_staleness_ms: u64,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RevokeEpochRequestPayload {
    resource_scope: NetworkResourceScope,
    plugin_instance_id: String,
    revoke_epoch: u64,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SessionRevokeRequestPayload {
    session_revoke_sequence: u64,
    owner_session_hash: String,
    owner_user_hash: String,
    owner_env_hash: String,
    session_channel_id_hash: String,
}

fn parse_raw_frame(input: &str) -> IpcResult<RawIPCFrame> {
    serde_json::from_str(input).map_err(|_| decode_failed("IPC frame"))
}

fn parse_hello_payload(frame: &RawIPCFrame) -> IpcResult<HelloPayload> {
    serde_json::from_str(frame.payload.get()).map_err(|_| decode_failed("hello payload"))
}

fn parse_closed_worker_frame(
    identity: &FrameIdentity,
    payload: &serde_json::value::RawValue,
) -> IpcResult<ClosedWorkerFrame> {
    if identity.parent_request_id.is_some() {
        return Err(protocol_violation(
            "invoke_worker must not have parent_request_id",
        ));
    }
    let payload: WorkerFramePayload =
        serde_json::from_str(payload.get()).map_err(|_| decode_failed("worker frame payload"))?;
    if payload.method.trim().is_empty() {
        return Err(invalid_field("worker frame method"));
    }
    if payload
        .invocation
        .method
        .as_deref()
        .is_some_and(|method| method.trim() != payload.method.trim())
    {
        return Err(protocol_violation(
            "worker invocation method does not match the frame envelope",
        ));
    }
    Ok(ClosedWorkerFrame {
        request_id: identity.request_id.clone(),
        runtime_generation_id: identity.runtime_generation_id.clone(),
        method: payload.method,
        lease: payload.lease,
        invocation: payload.invocation,
    })
}

fn parsed_worker_invocation(
    identity: &FrameIdentity,
    payload: &serde_json::value::RawValue,
) -> IpcResult<ParsedWorkerInvocation> {
    let parsed = parse_closed_worker_frame(identity, payload)?;
    let params_json = parsed
        .invocation
        .params
        .as_ref()
        .map(encode_worker_canonical_json)
        .transpose()
        .map_err(|_| encode_failed("parsed worker params"))?;
    let broker_access_json = parsed
        .invocation
        .broker_access
        .as_ref()
        .map(encode_worker_canonical_json)
        .transpose()
        .map_err(|_| encode_failed("parsed worker broker access"))?;
    Ok(ParsedWorkerInvocation {
        request_id: parsed.request_id,
        runtime_generation_id: parsed.runtime_generation_id,
        method: parsed.method,
        lease: parsed.lease,
        invocation: parsed.invocation,
        params_json,
        broker_access_json,
        context: OnceLock::new(),
        identity: OnceLock::new(),
        target_hash: OnceLock::new(),
    })
}

pub fn parse_worker_invocation(input: &str) -> IpcResult<ParsedWorkerInvocation> {
    match decode_runtime_input_frame(input)? {
        RuntimeInputFrame::InvokeWorker(worker) => worker.invocation,
        _ => Err(protocol_violation("expected invoke_worker frame")),
    }
}

fn encode_worker_canonical_json<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
    let encoded = serde_json::to_string(value)?;
    if !encoded.contains(['\u{2028}', '\u{2029}']) {
        return Ok(encoded);
    }
    let mut canonical = String::with_capacity(encoded.len());
    for character in encoded.chars() {
        match character {
            '\u{2028}' => canonical.push_str("\\u2028"),
            '\u{2029}' => canonical.push_str("\\u2029"),
            _ => canonical.push(character),
        }
    }
    Ok(canonical)
}

fn required_string(value: &Option<String>, field: &'static str) -> IpcResult<String> {
    value
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .ok_or_else(|| missing_field(field))
}

impl ParsedWorkerInvocation {
    pub fn request_id(&self) -> &str {
        &self.request_id
    }

    pub fn runtime_generation_id(&self) -> &str {
        &self.runtime_generation_id
    }

    pub fn plugin_instance_id(&self) -> IpcResult<&str> {
        self.invocation
            .plugin_instance_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| missing_field("plugin_instance_id"))
    }

    /// Returns the exact session scope when the invocation is session-bound.
    /// Background invocations may omit both session-specific hashes. A partial
    /// session identity is rejected instead of being treated as unscoped.
    pub fn session_scope(&self) -> IpcResult<Option<SessionScope>> {
        let invocation = &self.invocation;
        let session_present = invocation
            .owner_session_hash
            .as_deref()
            .is_some_and(|value| !value.is_empty())
            || invocation
                .session_channel_id_hash
                .as_deref()
                .is_some_and(|value| !value.is_empty());
        if !session_present {
            return Ok(None);
        }
        SessionScope::new(
            required_string(&invocation.owner_session_hash, "owner_session_hash")?,
            required_string(&invocation.owner_user_hash, "owner_user_hash")?,
            required_string(&invocation.owner_env_hash, "owner_env_hash")?,
            required_string(
                &invocation.session_channel_id_hash,
                "session_channel_id_hash",
            )?,
        )
        .map(Some)
    }

    pub fn context(&self) -> IpcResult<WorkerInvocationContext> {
        self.context.get_or_init(|| self.build_context()).clone()
    }

    fn build_context(&self) -> IpcResult<WorkerInvocationContext> {
        let invocation = &self.invocation;
        Ok(WorkerInvocationContext {
            plugin_id: required_string(&invocation.plugin_id, "plugin_id")?,
            plugin_instance_id: required_string(
                &invocation.plugin_instance_id,
                "plugin_instance_id",
            )?,
            active_fingerprint: required_string(
                &invocation.active_fingerprint,
                "active_fingerprint",
            )?,
            runtime_instance_id: required_string(
                &invocation.runtime_instance_id,
                "runtime_instance_id",
            )?,
            runtime_generation_id: required_string(
                &invocation.runtime_generation_id,
                "runtime_generation_id",
            )?,
            runtime_shard_id: required_string(&self.lease.runtime_shard_id, "runtime_shard_id")?,
            method: required_string(&invocation.method, "method")?,
            effect: invocation.effect.clone().unwrap_or_default(),
            execution: invocation.execution.clone().unwrap_or_default(),
            surface_instance_id: invocation.surface_instance_id.clone().unwrap_or_default(),
            owner_session_hash: invocation.owner_session_hash.clone().unwrap_or_default(),
            owner_user_hash: invocation.owner_user_hash.clone().unwrap_or_default(),
            owner_env_hash: invocation.owner_env_hash.clone().unwrap_or_default(),
            session_channel_id_hash: invocation
                .session_channel_id_hash
                .clone()
                .unwrap_or_default(),
            bridge_channel_id: invocation.bridge_channel_id.clone().unwrap_or_default(),
            operation_id: invocation.operation_id.clone().unwrap_or_default(),
            stream_id: invocation.stream_id.clone().unwrap_or_default(),
            policy_revision: required_safe_u64(self.lease.policy_revision, "policy_revision")?,
            management_revision: required_safe_u64(
                self.lease.management_revision,
                "management_revision",
            )?,
            revoke_epoch: required_positive_u64(self.lease.revoke_epoch, "revoke_epoch")?,
            storage_handle_grants: invocation.storage_handle_grants.clone().unwrap_or_default(),
            broker_access_json: self
                .broker_access_json
                .clone()
                .unwrap_or_else(|| "{}".to_string()),
        })
    }

    pub fn identity(&self) -> IpcResult<WorkerInvocationIdentity> {
        self.identity.get_or_init(|| self.build_identity()).clone()
    }

    pub fn validate_worker_contract(&self) -> IpcResult<()> {
        if self.invocation.abi.as_deref() != Some(WASM_ABI_VERSION) {
            return Err(protocol_violation("worker invocation ABI is unsupported"));
        }
        if self.invocation.worker_mode.as_deref() != Some("job") {
            return Err(protocol_violation("worker invocation mode is unsupported"));
        }
        required_string(&self.invocation.worker_scope, "worker_scope")?;
        Ok(())
    }

    fn build_identity(&self) -> IpcResult<WorkerInvocationIdentity> {
        let invocation = &self.invocation;
        let package_hash = invocation
            .package_hash
            .clone()
            .ok_or_else(|| missing_field("package_hash"))?;
        if !is_sha256_ref(&package_hash) {
            return Err(invalid_field("package_hash"));
        }
        let artifact = invocation
            .artifact
            .clone()
            .ok_or_else(|| missing_field("artifact"))?;
        if !is_worker_artifact_path(&artifact) {
            return Err(invalid_field("artifact"));
        }
        let artifact_sha256 = invocation
            .artifact_sha256
            .clone()
            .ok_or_else(|| missing_field("artifact_sha256"))?;
        if !is_sha256_ref(&artifact_sha256) {
            return Err(invalid_field("artifact_sha256"));
        }
        let worker_id = invocation
            .worker_id
            .clone()
            .ok_or_else(|| missing_field("worker_id"))?;
        if worker_id.trim().is_empty() {
            return Err(invalid_field("worker_id"));
        }
        let method = invocation
            .method
            .clone()
            .ok_or_else(|| missing_field("method"))?;
        if method.trim().is_empty() {
            return Err(invalid_field("method"));
        }
        Ok(WorkerInvocationIdentity {
            package_hash,
            artifact,
            artifact_sha256,
            worker_id,
            method,
        })
    }

    pub fn worker_request_json_v2(&self) -> IpcResult<String> {
        let method = required_string(&self.invocation.method, "worker invocation method")?;
        let params = self
            .params_json
            .as_ref()
            .ok_or_else(|| missing_field("worker invocation params"))?;
        Ok(format!(
            "{{\"schema_version\":\"redevplugin.worker_request.v2\",\"method\":\"{}\",\"params\":{}}}",
            escape_json_string(&method),
            params
        ))
    }

    pub fn memory_limit_bytes(&self) -> IpcResult<usize> {
        let memory_bytes = self
            .lease
            .limits
            .as_ref()
            .and_then(|limits| limits.memory_bytes)
            .filter(|value| *value > 0)
            .ok_or_else(|| invalid_field("runtime lease memory_bytes limit"))?;
        if memory_bytes > MAX_RUNTIME_LEASE_MEMORY_BYTES {
            return Err(protocol_violation(
                "runtime lease memory_bytes limit exceeds platform maximum",
            ));
        }
        usize::try_from(memory_bytes)
            .map_err(|_| protocol_violation("runtime lease memory_bytes limit exceeds runtime"))
    }

    pub fn replay_key(&self) -> IpcResult<WorkerLeaseReplayKey> {
        let lease_id = self
            .lease
            .lease_id
            .clone()
            .ok_or_else(|| missing_field("lease_id"))?;
        if lease_id.trim().is_empty() {
            return Err(invalid_field("lease_id"));
        }
        let lease_nonce = self
            .lease
            .lease_nonce
            .clone()
            .ok_or_else(|| missing_field("lease_nonce"))?;
        if lease_nonce.trim().is_empty() {
            return Err(invalid_field("lease_nonce"));
        }
        let expires_at_unix_ms = self
            .lease
            .expires_at_unix_ms
            .filter(|value| *value > 0)
            .ok_or_else(|| invalid_field("expires_at_unix_ms"))?;
        Ok(WorkerLeaseReplayKey {
            lease_id,
            lease_nonce,
            expires_at_unix_ms,
        })
    }

    pub fn storage_handle_grant(&self, store_id: &str) -> IpcResult<String> {
        let grants = self
            .invocation
            .storage_handle_grants
            .as_ref()
            .ok_or_else(|| missing_field("worker invocation storage_handle_grants"))?;
        grants
            .get(store_id)
            .map(String::as_str)
            .filter(|value| !value.trim().is_empty())
            .map(str::to_string)
            .ok_or_else(|| invalid_field("worker invocation storage grant"))
    }

    pub fn validate_storage_broker_access(&self, store_id: &str, operation: &str) -> IpcResult<()> {
        let effect = required_string(&self.invocation.effect, "effect")?;
        if effect == "read" && !matches!(operation, "read" | "list" | "get" | "query") {
            return Err(protocol_violation(
                "worker method with read effect cannot perform the storage mutation",
            ));
        }
        if !self
            .invocation
            .broker_access
            .as_ref()
            .is_some_and(|access| {
                access.storage.iter().any(|entry| {
                    entry.store_id == store_id
                        && entry.operations.iter().any(|value| value == operation)
                })
            })
        {
            return Err(protocol_violation(
                "worker method is not allowed to perform the storage operation",
            ));
        }
        Ok(())
    }

    pub fn storage_broker_scope(&self, store_id: &str) -> IpcResult<String> {
        let scope = self
            .invocation
            .broker_access
            .as_ref()
            .and_then(|access| {
                access
                    .storage
                    .iter()
                    .find(|entry| entry.store_id == store_id)
            })
            .map(|entry| entry.scope.as_str())
            .filter(|scope| matches!(*scope, "user" | "environment"))
            .ok_or_else(|| invalid_field("worker invocation storage scope"))?;
        Ok(scope.to_string())
    }

    pub fn validate_network_broker_access(
        &self,
        connector_id: &str,
        transport: &str,
        operation: &str,
        http_method: &str,
    ) -> IpcResult<()> {
        let allowed = self
            .invocation
            .broker_access
            .as_ref()
            .is_some_and(|access| {
                access.network.iter().any(|entry| {
                    entry.connector_id == connector_id
                        && entry.transport == transport
                        && entry.operations.iter().any(|value| value == operation)
                        && (transport != "http"
                            || entry.http_methods.iter().any(|value| value == http_method))
                })
            });
        if !allowed {
            return Err(protocol_violation(
                "worker method is not allowed to perform the network operation",
            ));
        }
        Ok(())
    }

    pub fn network_broker_scope(&self, connector_id: &str, transport: &str) -> IpcResult<String> {
        let scope = self
            .invocation
            .broker_access
            .as_ref()
            .and_then(|access| {
                access.network.iter().find(|entry| {
                    entry.connector_id == connector_id && entry.transport == transport
                })
            })
            .map(|entry| entry.scope.trim())
            .filter(|scope| matches!(*scope, "user" | "environment"))
            .ok_or_else(|| invalid_field("worker invocation network scope"))?;
        Ok(scope.to_string())
    }
}

pub fn parse_worker_invocation_context(input: &str) -> IpcResult<WorkerInvocationContext> {
    parse_worker_invocation(input)?.context()
}

pub fn parse_heartbeat_request(input: &str) -> IpcResult<HeartbeatRequest> {
    let frame = parse_raw_frame(input)?;
    if frame.frame_type != FRAME_TYPE_HEARTBEAT {
        return Err(protocol_violation("expected heartbeat frame"));
    }
    let payload: HeartbeatRequestPayload = serde_json::from_str(frame.payload.get())
        .map_err(|_| decode_failed("heartbeat payload"))?;
    Ok(HeartbeatRequest {
        sent_unix_nano: payload.sent_unix_nano,
        max_staleness_ms: payload.max_staleness_ms,
    })
}

pub fn parse_revoke_epoch_request(input: &str) -> IpcResult<RevokeEpochRequest> {
    let frame = parse_raw_frame(input)?;
    if frame.frame_type != FRAME_TYPE_REVOKE_EPOCH {
        return Err(protocol_violation("expected revoke_epoch frame"));
    }
    let payload: RevokeEpochRequestPayload = serde_json::from_str(frame.payload.get())
        .map_err(|_| decode_failed("revoke_epoch payload"))?;
    if payload.plugin_instance_id.trim().is_empty() {
        return Err(invalid_field("plugin_instance_id"));
    }
    if !payload.resource_scope.valid() || payload.resource_scope.kind != "environment" {
        return Err(invalid_field("revoke resource_scope"));
    }
    validate_revoke_epoch(payload.revoke_epoch)?;
    Ok(RevokeEpochRequest {
        resource_scope: payload.resource_scope,
        plugin_instance_id: payload.plugin_instance_id,
        revoke_epoch: payload.revoke_epoch,
    })
}

pub fn parse_session_revoke_request(input: &str) -> IpcResult<SessionRevokeRequest> {
    let frame = parse_raw_frame(input)?;
    let identity = validated_frame_identity(&frame)?;
    if identity.frame_type != FRAME_TYPE_SESSION_REVOKE {
        return Err(protocol_violation("expected session_revoke frame"));
    }
    if identity.parent_request_id.is_some() {
        return Err(protocol_violation(
            "session_revoke must not have parent_request_id",
        ));
    }
    let payload: SessionRevokeRequestPayload = serde_json::from_str(frame.payload.get())
        .map_err(|_| decode_failed("session_revoke payload"))?;
    if payload.session_revoke_sequence == 0
        || payload.session_revoke_sequence > MAX_JSON_SAFE_INTEGER
    {
        return Err(invalid_field("session_revoke_sequence"));
    }
    let scope = SessionScope::new(
        payload.owner_session_hash,
        payload.owner_user_hash,
        payload.owner_env_hash,
        payload.session_channel_id_hash,
    )?;
    Ok(SessionRevokeRequest {
        request_id: identity.request_id,
        runtime_generation_id: identity.runtime_generation_id,
        session_revoke_sequence: payload.session_revoke_sequence,
        owner_session_hash: scope.owner_session_hash,
        owner_user_hash: scope.owner_user_hash,
        owner_env_hash: scope.owner_env_hash,
        session_channel_id_hash: scope.session_channel_id_hash,
    })
}

pub fn escape_json_string(input: &str) -> String {
    let mut out = String::with_capacity(input.len());
    for ch in input.chars() {
        match ch {
            '"' => out.push_str("\\\""),
            '\\' => out.push_str("\\\\"),
            '\n' => out.push_str("\\n"),
            '\r' => out.push_str("\\r"),
            '\t' => out.push_str("\\t"),
            c if c.is_control() => out.push_str(&format!("\\u{:04x}", c as u32)),
            other => out.push(other),
        }
    }
    out
}

pub fn bind_parent_request_id(frame: &str, parent_request_id: &str) -> IpcResult<String> {
    if parent_request_id.trim().is_empty() {
        return Err(invalid_field("parent_request_id"));
    }
    let mut value: serde_json::Value =
        serde_json::from_str(frame).map_err(|_| decode_failed("outbound IPC frame"))?;
    let object = value
        .as_object_mut()
        .ok_or_else(|| protocol_violation("outbound IPC frame must be an object"))?;
    object.insert(
        "parent_request_id".to_string(),
        serde_json::Value::String(parent_request_id.to_string()),
    );
    serde_json::to_string(&value).map_err(|_| encode_failed("outbound IPC frame"))
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeLeasePublicKey {
    pub key_id: String,
    pub public_key: [u8; 32],
}

fn parse_runtime_lease_public_key_payloads(
    keys: Vec<RuntimeLeasePublicKeyPayload>,
) -> IpcResult<Vec<RuntimeLeasePublicKey>> {
    let mut seen = HashSet::new();
    let mut parsed = Vec::with_capacity(keys.len());
    if keys.is_empty() {
        return Err(invalid_field("runtime_lease_public_keys"));
    }
    for key in keys {
        let key_id = key.key_id.trim().to_string();
        if key_id.is_empty() {
            return Err(invalid_field("runtime lease public key key_id"));
        }
        if !seen.insert(key_id.clone()) {
            return Err(protocol_violation(
                "runtime lease public key key_id is duplicated",
            ));
        }
        if key.algorithm != RUNTIME_LEASE_SIGNATURE_ALGORITHM {
            return Err(protocol_violation(
                "runtime lease public key algorithm is unsupported",
            ));
        }
        let decoded = base64::engine::general_purpose::STANDARD
            .decode(key.public_key_base64.as_bytes())
            .map_err(|_| invalid_field("runtime lease public key base64"))?;
        let public_key: [u8; 32] = decoded
            .try_into()
            .map_err(|_| invalid_field("runtime lease public key length"))?;
        parsed.push(RuntimeLeasePublicKey { key_id, public_key });
    }
    Ok(parsed)
}

pub fn parse_runtime_lease_public_keys(input: &str) -> IpcResult<Vec<RuntimeLeasePublicKey>> {
    let frame = parse_raw_frame(input)?;
    let payload = parse_hello_payload(&frame)?;
    parse_runtime_lease_public_key_payloads(payload.runtime_lease_public_keys)
}

pub fn verify_worker_runtime_lease_signature(
    input: &str,
    public_keys: &[RuntimeLeasePublicKey],
) -> IpcResult<()> {
    parse_worker_invocation(input)?.verify_runtime_lease_signature(public_keys)
}

impl ParsedWorkerInvocation {
    pub fn verify_runtime_lease_signature(
        &self,
        public_keys: &[RuntimeLeasePublicKey],
    ) -> IpcResult<()> {
        if public_keys.is_empty() {
            return Err(missing_field("runtime lease public keys"));
        }
        let key_id = required_string(&self.lease.key_id, "key_id")?;
        let public_key = public_keys
            .iter()
            .find(|key| key.key_id == key_id)
            .ok_or_else(|| invalid_field("runtime lease signing key"))?;
        let verifying_key = VerifyingKey::from_bytes(&public_key.public_key)
            .map_err(|_| invalid_field("runtime lease public key"))?;
        let payload = runtime_lease_signature_payload_json(&self.lease, &self.method)?;
        let signature =
            decode_runtime_lease_signature(&required_string(&self.lease.signature, "signature")?)?;
        verifying_key
            .verify(payload.as_bytes(), &signature)
            .map_err(|_| invalid_field("runtime lease signature"))
    }
}

pub fn validate_worker_runtime_lease(input: &str, now_unix_ms: i64) -> IpcResult<()> {
    parse_worker_invocation(input)?.validate_runtime_lease(now_unix_ms)
}

impl ParsedWorkerInvocation {
    pub fn validate_runtime_lease(&self, now_unix_ms: i64) -> IpcResult<()> {
        let lease = &self.lease;
        let invocation = &self.invocation;
        let expires_at_unix_ms = positive_i64(lease.expires_at_unix_ms, "expires_at_unix_ms")?;
        if expires_at_unix_ms <= now_unix_ms {
            return Err(protocol_violation("runtime execution lease is expired"));
        }
        validate_runtime_lease_string_binding(&lease.method, &invocation.method, "method", true)?;
        if required_string(&lease.method, "method")? != self.method {
            return Err(protocol_violation(
                "runtime lease method does not match the invocation envelope",
            ));
        }
        for (lease_value, invocation_value, field) in [
            (&lease.plugin_id, &invocation.plugin_id, "plugin_id"),
            (
                &lease.plugin_instance_id,
                &invocation.plugin_instance_id,
                "plugin_instance_id",
            ),
            (
                &lease.active_fingerprint,
                &invocation.active_fingerprint,
                "active_fingerprint",
            ),
            (
                &lease.runtime_instance_id,
                &invocation.runtime_instance_id,
                "runtime_instance_id",
            ),
            (
                &lease.runtime_generation_id,
                &invocation.runtime_generation_id,
                "runtime_generation_id",
            ),
            (&lease.effect, &invocation.effect, "effect"),
            (&lease.execution, &invocation.execution, "execution"),
            (
                &lease.audit_correlation_id,
                &invocation.audit_correlation_id,
                "audit_correlation_id",
            ),
        ] {
            validate_runtime_lease_string_binding(lease_value, invocation_value, field, true)?;
        }
        for (lease_value, invocation_value, field) in [
            (
                &lease.surface_instance_id,
                &invocation.surface_instance_id,
                "surface_instance_id",
            ),
            (
                &lease.owner_session_hash,
                &invocation.owner_session_hash,
                "owner_session_hash",
            ),
            (
                &lease.owner_user_hash,
                &invocation.owner_user_hash,
                "owner_user_hash",
            ),
            (
                &lease.owner_env_hash,
                &invocation.owner_env_hash,
                "owner_env_hash",
            ),
            (
                &lease.session_channel_id_hash,
                &invocation.session_channel_id_hash,
                "session_channel_id_hash",
            ),
            (
                &lease.bridge_channel_id,
                &invocation.bridge_channel_id,
                "bridge_channel_id",
            ),
            (
                &lease.operation_id,
                &invocation.operation_id,
                "operation_id",
            ),
            (&lease.stream_id, &invocation.stream_id, "stream_id"),
        ] {
            validate_runtime_lease_string_binding(lease_value, invocation_value, field, false)?;
        }
        if required_string(&lease.runtime_generation_id, "runtime_generation_id")?
            != self.runtime_generation_id
        {
            return Err(protocol_violation(
                "runtime lease runtime_generation_id does not match the invocation frame",
            ));
        }
        validate_runtime_execution_handles(
            &lease.execution,
            &lease.operation_id,
            &lease.stream_id,
        )?;
        validate_runtime_execution_handles(
            &invocation.execution,
            &invocation.operation_id,
            &invocation.stream_id,
        )?;
        let invocation_target_hash = self.target_hash()?;
        let target_hashes = lease
            .target_descriptor_hashes
            .as_ref()
            .ok_or_else(|| missing_field("runtime lease target_descriptor_hashes"))?;
        if target_hashes
            .iter()
            .filter(|value| value.as_str() == invocation_target_hash.as_str())
            .count()
            != 1
        {
            return Err(protocol_violation(
                "runtime lease does not bind the worker invocation target",
            ));
        }
        Ok(())
    }
}

pub fn worker_invocation_target_hash(input: &str) -> IpcResult<String> {
    parse_worker_invocation(input)?.target_hash()
}

impl ParsedWorkerInvocation {
    pub fn target_hash(&self) -> IpcResult<String> {
        self.target_hash
            .get_or_init(|| self.build_target_hash())
            .clone()
    }

    fn build_target_hash(&self) -> IpcResult<String> {
        let invocation = &self.invocation;
        let params = self
            .params_json
            .as_ref()
            .ok_or_else(|| missing_field("worker invocation params"))?;
        let broker_access = self
            .broker_access_json
            .as_ref()
            .ok_or_else(|| missing_field("worker invocation broker_access"))?;
        let params_hash = format!(
            "sha256:{}",
            lowercase_hex(&Sha256::digest(params.as_bytes()))
        );
        if required_string(&invocation.params_sha256, "params_sha256")? != params_hash {
            return Err(protocol_violation(
                "worker invocation params_sha256 does not match params",
            ));
        }
        let broker_access_hash = format!(
            "sha256:{}",
            lowercase_hex(&Sha256::digest(broker_access.as_bytes()))
        );
        if self.invocation.broker_access_sha256.as_deref() != Some(broker_access_hash.as_str()) {
            return Err(protocol_violation(
                "worker invocation broker_access_sha256 does not match broker_access",
            ));
        }
        let fields = [
            WORKER_INVOCATION_TARGET_SCHEMA_VERSION.to_string(),
            required_string(&invocation.plugin_id, "plugin_id")?,
            required_string(&invocation.plugin_instance_id, "plugin_instance_id")?,
            required_string(&invocation.active_fingerprint, "active_fingerprint")?,
            required_string(&invocation.runtime_instance_id, "runtime_instance_id")?,
            required_string(&invocation.runtime_generation_id, "runtime_generation_id")?,
            required_string(&invocation.package_hash, "package_hash")?,
            required_string(&invocation.worker_id, "worker_id")?,
            required_string(&invocation.worker_mode, "worker_mode")?,
            required_string(&invocation.worker_scope, "worker_scope")?,
            required_string(&invocation.artifact, "artifact")?,
            required_string(&invocation.artifact_sha256, "artifact_sha256")?,
            required_string(&invocation.abi, "abi")?,
            required_string(&invocation.method, "method")?,
            required_string(&invocation.effect, "effect")?,
            required_string(&invocation.execution, "execution")?,
            optional_string(&invocation.surface_instance_id),
            optional_string(&invocation.owner_session_hash),
            optional_string(&invocation.owner_user_hash),
            optional_string(&invocation.owner_env_hash),
            optional_string(&invocation.session_channel_id_hash),
            optional_string(&invocation.bridge_channel_id),
            optional_string(&invocation.operation_id),
            optional_string(&invocation.stream_id),
            required_string(&invocation.audit_correlation_id, "audit_correlation_id")?,
            params_hash,
            broker_access_hash,
        ];
        let mut canonical = Vec::new();
        for field in fields {
            let length = u32::try_from(field.len()).map_err(|_| {
                protocol_violation("worker invocation target field exceeds uint32 length")
            })?;
            canonical.extend_from_slice(&length.to_be_bytes());
            canonical.extend_from_slice(field.as_bytes());
        }
        Ok(format!(
            "invocation:sha256:{}",
            lowercase_hex(&Sha256::digest(canonical))
        ))
    }
}

fn lowercase_hex(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut encoded = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        encoded.push(HEX[(byte >> 4) as usize] as char);
        encoded.push(HEX[(byte & 0x0f) as usize] as char);
    }
    encoded
}

fn validate_runtime_lease_string_binding(
    lease: &Option<String>,
    invocation: &Option<String>,
    field: &'static str,
    required: bool,
) -> IpcResult<()> {
    let lease_value = optional_string_ref(lease);
    let invocation_value = optional_string_ref(invocation);
    if required && (lease_value.is_none() || invocation_value.is_none()) {
        return Err(IpcError::MissingField { field });
    }
    if lease_value != invocation_value {
        return Err(IpcError::InvalidField { field });
    }
    Ok(())
}

fn validate_runtime_execution_handles(
    execution: &Option<String>,
    operation_id: &Option<String>,
    stream_id: &Option<String>,
) -> IpcResult<()> {
    let execution = required_string(execution, "execution")?;
    let operation_id = optional_string_ref(operation_id).unwrap_or_default();
    let stream_id = optional_string_ref(stream_id).unwrap_or_default();
    match execution.as_str() {
        "sync" if operation_id.is_empty() && stream_id.is_empty() => Ok(()),
        "operation" if !operation_id.is_empty() && stream_id.is_empty() => Ok(()),
        "subscription" if !operation_id.is_empty() && !stream_id.is_empty() => Ok(()),
        _ => Err(invalid_field("runtime lease execution handles")),
    }
}

fn decode_runtime_lease_signature(input: &str) -> IpcResult<Signature> {
    let raw = input.trim();
    let prefix = format!("{RUNTIME_LEASE_SIGNATURE_ALGORITHM}:");
    let encoded = raw
        .strip_prefix(prefix.as_str())
        .ok_or_else(|| protocol_violation("runtime lease signature algorithm is unsupported"))?;
    let decoded = base64::engine::general_purpose::STANDARD
        .decode(encoded.as_bytes())
        .map_err(|_| invalid_field("runtime lease signature base64"))?;
    Signature::from_slice(&decoded).map_err(|_| invalid_field("runtime lease signature length"))
}

fn runtime_lease_signature_payload_json(
    lease: &WorkerLeasePayload,
    method: &str,
) -> IpcResult<String> {
    if let Some(lease_method) = optional_string_ref(&lease.method) {
        if lease_method != method.trim() {
            return Err(protocol_violation("runtime lease method mismatch"));
        }
    }
    let lease_id = required_string(&lease.lease_id, "lease_id")?;
    let token_id = required_string(&lease.token_id, "token_id")?;
    let expires_at_unix_ms = positive_i64(lease.expires_at_unix_ms, "expires_at_unix_ms")?;
    let issued_at_unix_ms = positive_i64(lease.issued_at_unix_ms, "issued_at_unix_ms")?;
    let mut out = String::new();
    out.push('{');
    append_json_string_field(
        &mut out,
        "schema_version",
        RUNTIME_LEASE_SIGNATURE_SCHEMA_VERSION,
        false,
    );
    append_json_string_field(&mut out, "token_kind", RUNTIME_LEASE_TOKEN_KIND, true);
    append_json_string_field(&mut out, "lease_id", &lease_id, true);
    append_json_string_field(&mut out, "token_id", &token_id, true);
    let lease_nonce = required_string(&lease.lease_nonce, "lease_nonce")?;
    if lease_nonce.len() < 16 {
        return Err(invalid_field("runtime lease lease_nonce"));
    }
    append_json_string_field(&mut out, "lease_nonce", &lease_nonce, true);
    append_json_string_field(
        &mut out,
        "plugin_instance_id",
        &required_string(&lease.plugin_instance_id, "plugin_instance_id")?,
        true,
    );
    append_json_string_field(
        &mut out,
        "plugin_id",
        &required_string(&lease.plugin_id, "plugin_id")?,
        true,
    );
    append_json_string_field(
        &mut out,
        "plugin_version",
        &required_string(&lease.plugin_version, "plugin_version")?,
        true,
    );
    append_json_string_field(
        &mut out,
        "active_fingerprint",
        &required_string(&lease.active_fingerprint, "active_fingerprint")?,
        true,
    );
    append_json_i64_field(&mut out, "issued_at_unix_ms", issued_at_unix_ms);
    append_json_string_field(&mut out, "method", method.trim(), true);
    let effect = required_string(&lease.effect, "effect")?;
    if !matches!(
        effect.as_str(),
        "read" | "write" | "execute" | "delete" | "admin"
    ) {
        return Err(invalid_field("runtime lease effect"));
    }
    append_json_string_field(&mut out, "effect", &effect, true);
    append_json_string_field(
        &mut out,
        "execution",
        &required_string(&lease.execution, "execution")?,
        true,
    );
    validate_runtime_execution_handles(&lease.execution, &lease.operation_id, &lease.stream_id)?;
    let operation_id = optional_string(&lease.operation_id);
    let stream_id = optional_string(&lease.stream_id);
    append_json_optional_string_field(&mut out, "operation_id", Some(&operation_id));
    append_json_optional_string_field(&mut out, "stream_id", Some(&stream_id));
    append_json_string_field(
        &mut out,
        "audit_correlation_id",
        &required_string(&lease.audit_correlation_id, "audit_correlation_id")?,
        true,
    );
    append_json_optional_string_field(
        &mut out,
        "surface_instance_id",
        optional_string_ref(&lease.surface_instance_id),
    );
    append_json_optional_string_field(
        &mut out,
        "owner_session_hash",
        optional_string_ref(&lease.owner_session_hash),
    );
    append_json_optional_string_field(
        &mut out,
        "owner_user_hash",
        optional_string_ref(&lease.owner_user_hash),
    );
    append_json_string_field(
        &mut out,
        "owner_env_hash",
        &required_string(&lease.owner_env_hash, "owner_env_hash")?,
        true,
    );
    append_json_optional_string_field(
        &mut out,
        "session_channel_id_hash",
        optional_string_ref(&lease.session_channel_id_hash),
    );
    append_json_optional_string_field(
        &mut out,
        "bridge_channel_id",
        optional_string_ref(&lease.bridge_channel_id),
    );
    let target_hashes = lease
        .target_descriptor_hashes
        .as_ref()
        .filter(|hashes| !hashes.is_empty())
        .ok_or_else(|| missing_field("runtime lease target_descriptor_hashes"))?;
    let mut seen_target_hashes = HashSet::new();
    out.push_str(",\"target_descriptor_hashes\":[");
    for (index, hash) in target_hashes.iter().enumerate() {
        let hash = hash.trim();
        if hash.is_empty() {
            return Err(invalid_field("target_descriptor_hashes item"));
        }
        if !seen_target_hashes.insert(hash) {
            return Err(protocol_violation(
                "target_descriptor_hashes item is duplicated",
            ));
        }
        if index > 0 {
            out.push(',');
        }
        out.push('"');
        out.push_str(&escape_json_string(hash));
        out.push('"');
    }
    out.push(']');
    append_runtime_lease_limits_field(
        &mut out,
        lease
            .limits
            .as_ref()
            .ok_or_else(|| missing_field("runtime lease limits"))?,
    )?;
    append_json_u64_field(
        &mut out,
        "policy_revision",
        required_safe_u64(lease.policy_revision, "policy_revision")?,
    );
    append_json_u64_field(
        &mut out,
        "management_revision",
        required_safe_u64(lease.management_revision, "management_revision")?,
    );
    append_json_u64_field(
        &mut out,
        "revoke_epoch",
        required_positive_u64(lease.revoke_epoch, "revoke_epoch")?,
    );
    append_json_i64_field(&mut out, "expires_at_unix_ms", expires_at_unix_ms);
    append_json_string_field(
        &mut out,
        "runtime_shard_id",
        &required_string(&lease.runtime_shard_id, "runtime_shard_id")?,
        true,
    );
    append_json_string_field(
        &mut out,
        "runtime_instance_id",
        &required_string(&lease.runtime_instance_id, "runtime_instance_id")?,
        true,
    );
    append_json_string_field(
        &mut out,
        "runtime_generation_id",
        &required_string(&lease.runtime_generation_id, "runtime_generation_id")?,
        true,
    );
    append_json_string_field(
        &mut out,
        "ipc_channel_id",
        &required_string(&lease.ipc_channel_id, "ipc_channel_id")?,
        true,
    );
    let connection_nonce = required_string(&lease.connection_nonce, "connection_nonce")?;
    if connection_nonce.len() < 16 {
        return Err(invalid_field("runtime lease connection_nonce"));
    }
    append_json_string_field(&mut out, "connection_nonce", &connection_nonce, true);
    append_json_string_field(
        &mut out,
        "key_id",
        &required_string(&lease.key_id, "key_id")?,
        true,
    );
    out.push('}');
    Ok(out)
}

fn optional_string_ref(value: &Option<String>) -> Option<&str> {
    value
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
}

fn optional_string(value: &Option<String>) -> String {
    optional_string_ref(value).unwrap_or_default().to_string()
}

fn positive_i64(value: Option<i64>, field: &'static str) -> IpcResult<i64> {
    value.ok_or_else(|| missing_field(field)).and_then(|value| {
        if value > 0 && value as u64 <= MAX_JSON_SAFE_INTEGER {
            Ok(value)
        } else {
            Err(invalid_field(field))
        }
    })
}

fn nonnegative_i64(value: Option<i64>, field: &'static str) -> IpcResult<i64> {
    value.ok_or_else(|| missing_field(field)).and_then(|value| {
        if value >= 0 && value as u64 <= MAX_JSON_SAFE_INTEGER {
            Ok(value)
        } else {
            Err(invalid_field(field))
        }
    })
}

fn required_u64(value: Option<u64>, field: &'static str) -> IpcResult<u64> {
    value.ok_or_else(|| missing_field(field))
}

fn required_safe_u64(value: Option<u64>, field: &'static str) -> IpcResult<u64> {
    validate_safe_u64(required_u64(value, field)?, field)
}

fn required_positive_u64(value: Option<u64>, field: &'static str) -> IpcResult<u64> {
    validate_positive_u64(required_u64(value, field)?, field)
}

fn validate_positive_u64(value: u64, field: &'static str) -> IpcResult<u64> {
    if value == 0 {
        return Err(invalid_field(field));
    }
    validate_safe_u64(value, field)
}

fn validate_safe_u64(value: u64, field: &'static str) -> IpcResult<u64> {
    if value > MAX_JSON_SAFE_INTEGER {
        return Err(invalid_field(field));
    }
    Ok(value)
}

fn validate_revoke_epoch(revoke_epoch: u64) -> IpcResult<()> {
    validate_positive_u64(revoke_epoch, "revoke_epoch").map(|_| ())
}

fn validate_revision_binding(
    policy_revision: u64,
    management_revision: u64,
    revoke_epoch: u64,
) -> IpcResult<()> {
    validate_safe_u64(policy_revision, "policy_revision")?;
    validate_safe_u64(management_revision, "management_revision")?;
    validate_revoke_epoch(revoke_epoch)
}

fn append_json_string_field(out: &mut String, key: &str, value: &str, comma: bool) {
    if comma {
        out.push(',');
    }
    out.push('"');
    out.push_str(key);
    out.push_str("\":\"");
    out.push_str(&escape_json_string(value));
    out.push('"');
}

fn append_json_optional_string_field(out: &mut String, key: &str, value: Option<&str>) {
    let Some(value) = value else {
        return;
    };
    let value = value.trim();
    if value.is_empty() {
        return;
    }
    append_json_string_field(out, key, value, true);
}

fn append_json_u64_field(out: &mut String, key: &str, value: u64) {
    out.push_str(",\"");
    out.push_str(key);
    out.push_str("\":");
    out.push_str(value.to_string().as_str());
}

fn append_json_i64_field(out: &mut String, key: &str, value: i64) {
    out.push_str(",\"");
    out.push_str(key);
    out.push_str("\":");
    out.push_str(value.to_string().as_str());
}

fn append_runtime_lease_limits_field(
    out: &mut String,
    limits: &WorkerLeaseLimitsPayload,
) -> IpcResult<()> {
    let timeout_ms = nonnegative_i64(limits.timeout_ms, "timeout_ms")?;
    let memory_bytes = limits
        .memory_bytes
        .filter(|value| *value > 0)
        .ok_or_else(|| invalid_field("memory_bytes"))?;
    if memory_bytes > MAX_RUNTIME_LEASE_MEMORY_BYTES {
        return Err(protocol_violation(
            "memory_bytes exceeds runtime lease limit",
        ));
    }
    let max_payload_bytes = nonnegative_i64(limits.max_payload_bytes, "max_payload_bytes")?;
    let max_stream_bytes_per_sec =
        nonnegative_i64(limits.max_stream_bytes_per_sec, "max_stream_bytes_per_sec")?;
    out.push_str(&format!(
        ",\"limits\":{{\"timeout_ms\":{timeout_ms},\"memory_bytes\":{memory_bytes},\"max_payload_bytes\":{max_payload_bytes},\"max_stream_bytes_per_sec\":{max_stream_bytes_per_sec}}}"
    ));
    Ok(())
}

#[derive(Debug, Clone, Copy)]
pub struct HelloAckFrameRequest<'a> {
    pub request_id: &'a str,
    pub runtime_generation_id: &'a str,
    pub channel_nonce: &'a str,
    pub runtime_version: &'a str,
    pub actual_target: &'a RuntimeTarget,
    pub wasm_abi_version: &'a str,
    pub limits: RuntimeLimits,
    pub process_containment: &'a ProcessContainmentEvidence,
}

pub fn hello_ack_frame(request: HelloAckFrameRequest<'_>) -> IpcResult<String> {
    let limits = request.limits.validate()?;
    let limits = serde_json::to_string(&limits).map_err(|_| encode_failed("runtime limits"))?;
    request.process_containment.validate()?;
    let process_containment = serde_json::to_string(request.process_containment)
        .map_err(|_| encode_failed("process containment evidence"))?;
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"runtime_version\":\"{}\",\"actual_target\":\"{}\",\"rust_ipc_version\":\"{}\",\"wasm_abi_version\":\"{}\",\"contract_set_sha256\":\"{}\",\"channel_nonce\":\"{}\",\"limits\":{},\"process_containment\":{}}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_HELLO_ACK,
        escape_json_string(request.request_id),
        escape_json_string(request.runtime_generation_id),
        escape_json_string(request.runtime_version),
        escape_json_string(request.actual_target.as_str()),
        RUST_IPC_VERSION,
        escape_json_string(request.wasm_abi_version),
        CONTRACT_SET_SHA256,
        escape_json_string(request.channel_nonce),
        limits,
        process_containment
    ))
}

pub fn success_response_frame(
    frame_type: &str,
    request_id: &str,
    runtime_generation_id: &str,
    result_json: &str,
) -> IpcResult<String> {
    serde_json::from_str::<serde_json::Value>(result_json)
        .map_err(|_| IpcError::InvalidResponseResultJson)?;
    let payload = format!("{{\"ok\":true,\"result\":{result_json}}}");
    Ok(render_response_frame(
        frame_type,
        request_id,
        runtime_generation_id,
        &payload,
    ))
}

pub fn session_revoke_ack_frame(
    request_id: &str,
    runtime_generation_id: &str,
    session_revoke_sequence: u64,
    state: SessionRevokeState,
    counts: SessionRevokeAckCounts,
) -> IpcResult<String> {
    if request_id.is_empty() || request_id.trim() != request_id {
        return Err(invalid_field("request_id"));
    }
    if runtime_generation_id.is_empty() || runtime_generation_id.trim() != runtime_generation_id {
        return Err(invalid_field("runtime_generation_id"));
    }
    if session_revoke_sequence == 0 || session_revoke_sequence > MAX_JSON_SAFE_INTEGER {
        return Err(invalid_field("session_revoke_sequence"));
    }
    for (count, field) in [
        (counts.queued_invocations, "queued_invocations"),
        (counts.running_invocations, "running_invocations"),
        (counts.storage_hostcalls, "storage_hostcalls"),
        (counts.active_network_requests, "active_network_requests"),
        (counts.sockets, "sockets"),
        (counts.network_streams, "network_streams"),
    ] {
        if count > MAX_JSON_SAFE_INTEGER {
            return Err(invalid_field(field));
        }
    }
    let result = serde_json::json!({
        "session_revoke_sequence": session_revoke_sequence,
        "state": state,
        "counts": counts,
    });
    success_response_frame(
        FRAME_TYPE_SESSION_REVOKE_ACK,
        request_id,
        runtime_generation_id,
        &result.to_string(),
    )
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ResponseError<'a> {
    code: &'a str,
    message: &'a str,
    origin: &'static str,
}

impl<'a> ResponseError<'a> {
    pub fn runtime(code: &'a str, message: &'a str) -> IpcResult<Self> {
        Self::new(code, message, ERROR_ORIGIN_RUNTIME)
    }

    pub fn hostcall(code: &'a str, message: &'a str) -> IpcResult<Self> {
        Self::new(code, message, ERROR_ORIGIN_HOSTCALL)
    }

    pub fn plugin(code: &'a str, message: &'a str) -> IpcResult<Self> {
        Self::new(code, message, ERROR_ORIGIN_PLUGIN)
    }

    fn new(code: &'a str, message: &'a str, origin: &'static str) -> IpcResult<Self> {
        if code.trim().is_empty() {
            return Err(IpcError::EmptyResponseErrorCode);
        }
        if message.trim().is_empty() {
            return Err(IpcError::EmptyResponseErrorMessage);
        }
        Ok(Self {
            code,
            message,
            origin,
        })
    }
}

pub fn error_response_frame(
    frame_type: &str,
    request_id: &str,
    runtime_generation_id: &str,
    error: ResponseError<'_>,
) -> IpcResult<String> {
    let payload = render_error_payload(error);
    Ok(render_response_frame(
        frame_type,
        request_id,
        runtime_generation_id,
        &payload,
    ))
}

fn render_error_payload(error: ResponseError<'_>) -> String {
    format!(
        "{{\"ok\":false,\"code\":\"{}\",\"message\":\"{}\",\"error_origin\":\"{}\"}}",
        escape_json_string(error.code),
        escape_json_string(error.message),
        error.origin,
    )
}

fn render_response_frame(
    frame_type: &str,
    request_id: &str,
    runtime_generation_id: &str,
    payload: &str,
) -> String {
    format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{}}}",
        RUST_IPC_VERSION,
        escape_json_string(frame_type),
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        payload,
    )
}

pub fn revoke_epoch_ack_result_json(
    resource_scope: &NetworkResourceScope,
    plugin_instance_id: &str,
    revoke_epoch: u64,
    closed_socket_count: u64,
    closed_stream_count: u64,
    closed_storage_handle_count: u64,
) -> IpcResult<String> {
    if !resource_scope.valid() || resource_scope.kind != "environment" {
        return Err(invalid_field("revoke resource scope"));
    }
    validate_revoke_epoch(revoke_epoch)?;
    let resource_scope = serde_json::to_string(resource_scope)
        .map_err(|_| encode_failed("revoke resource scope"))?;
    Ok(format!(
        "{{\"resource_scope\":{},\"plugin_instance_id\":\"{}\",\"revoke_epoch\":{},\"closed_socket_count\":{},\"closed_stream_count\":{},\"closed_storage_handle_count\":{}}}",
        resource_scope,
        escape_json_string(plugin_instance_id),
        revoke_epoch,
        closed_socket_count,
        closed_stream_count,
        closed_storage_handle_count
    ))
}

pub fn heartbeat_ack_result_json(
    runtime_generation_id: &str,
    runtime_unix_nano: u64,
    max_staleness_ms: u64,
    host_sent_unix_nano: u64,
    status: RuntimeHeartbeatStatus,
) -> IpcResult<String> {
    let limits = status.limits.validate()?;
    Ok(serde_json::json!({
        "runtime_generation_id": runtime_generation_id,
        "runtime_unix_nano": runtime_unix_nano,
        "max_staleness_ms": max_staleness_ms,
        "host_sent_unix_nano": host_sent_unix_nano,
        "active_invocations": status.active_invocations,
        "queued_invocations": status.queued_invocations,
        "limits": limits,
        "module_cache": status.module_cache,
    })
    .to_string())
}

#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
pub struct ModuleCacheMetrics {
    pub hits: u64,
    pub misses: u64,
    pub compiles: u64,
    pub entries: usize,
    pub source_bytes: usize,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RuntimeHeartbeatStatus {
    pub active_invocations: usize,
    pub queued_invocations: usize,
    pub limits: RuntimeLimits,
    pub module_cache: ModuleCacheMetrics,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CancelInvokePayload {
    invocation_request_id: String,
}

pub fn parse_cancel_invoke(input: &str) -> IpcResult<String> {
    let frame = parse_raw_frame(input)?;
    if frame.ipc_version != RUST_IPC_VERSION {
        return Err(protocol_violation("unsupported ipc_version"));
    }
    if frame.frame_type != FRAME_TYPE_CANCEL_INVOKE {
        return Err(protocol_violation("expected cancel_invoke frame"));
    }
    let payload: CancelInvokePayload = serde_json::from_str(frame.payload.get())
        .map_err(|_| decode_failed("cancel_invoke payload"))?;
    let request_id = payload.invocation_request_id.trim();
    if request_id.is_empty() {
        return Err(invalid_field("invocation_request_id"));
    }
    Ok(request_id.to_string())
}

pub fn cancel_invoke_ack_frame(
    request_id: &str,
    runtime_generation_id: &str,
    invocation_request_id: &str,
    disposition: &str,
) -> IpcResult<String> {
    let result = serde_json::json!({
        "invocation_request_id": invocation_request_id,
        "disposition": disposition,
    });
    success_response_frame(
        FRAME_TYPE_CANCEL_INVOKE_ACK,
        request_id,
        runtime_generation_id,
        &result.to_string(),
    )
}

enum HostcallResponsePayload<T> {
    Success(T),
    Failure(HostcallFailureResponsePayload),
}

#[derive(Deserialize)]
struct BooleanResponseDiscriminator {
    ok: bool,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct HostcallFailureResponsePayload {
    ok: bool,
    code: String,
    message: String,
    error_origin: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct OpenHandleSuccessResponsePayload {
    ok: bool,
    package_hash: String,
    artifact: String,
    sha256: String,
    content_base64: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct HandleGrantSuccessResponsePayload {
    ok: bool,
    handle_grant_id: String,
    handle_id: String,
    method: String,
    runtime_generation_id: String,
    resource_scope: NetworkResourceScope,
    max_bytes_per_second: Option<u64>,
    max_total_bytes: Option<u64>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageUsageResponsePayload {
    plugin_instance_id: String,
    store_id: String,
    usage_bytes: u64,
    quota_bytes: u64,
    usage_files: u64,
    quota_files: u64,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageFileEntryResponsePayload {
    path: String,
    dir: bool,
    size_bytes: Option<u64>,
    updated_at: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageFileReadSuccessResponsePayload {
    ok: bool,
    path: String,
    data_base64: String,
    size_bytes: u64,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageFileWriteSuccessResponsePayload {
    ok: bool,
    path: String,
    size_bytes: u64,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageFileDeleteSuccessResponsePayload {
    ok: bool,
    path: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageFileListSuccessResponsePayload {
    ok: bool,
    path: String,
    entries: Vec<StorageFileEntryResponsePayload>,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageKVEntryResponsePayload {
    key: String,
    size_bytes: u64,
    updated_at: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageKVGetSuccessResponsePayload {
    ok: bool,
    key: String,
    value_base64: String,
    size_bytes: u64,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageKVPutSuccessResponsePayload {
    ok: bool,
    key: String,
    size_bytes: u64,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageKVDeleteSuccessResponsePayload {
    ok: bool,
    key: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageKVListSuccessResponsePayload {
    ok: bool,
    prefix: Option<String>,
    entries: Vec<StorageKVEntryResponsePayload>,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageSQLiteValueResponsePayload {
    #[serde(rename = "null")]
    null_value: Option<bool>,
    int: Option<i64>,
    float: Option<f64>,
    text: Option<String>,
    blob_base64: Option<String>,
}

impl StorageSQLiteValueResponsePayload {
    fn is_exactly_typed(&self) -> bool {
        let variants = usize::from(self.null_value.is_some())
            + usize::from(self.int.is_some())
            + usize::from(self.float.is_some())
            + usize::from(self.text.is_some())
            + usize::from(self.blob_base64.is_some());
        variants == 1 && self.null_value.unwrap_or(true)
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageSQLiteExecSuccessResponsePayload {
    ok: bool,
    database: String,
    rows_affected: u64,
    last_insert_id: Option<u64>,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct StorageSQLiteQuerySuccessResponsePayload {
    ok: bool,
    database: String,
    columns: Vec<String>,
    rows: Vec<Vec<StorageSQLiteValueResponsePayload>>,
    usage: StorageUsageResponsePayload,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct NetworkDestinationResponsePayload {
    transport: String,
    scheme: Option<String>,
    host: String,
    port: u16,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct NetworkGrantSuccessResponsePayload {
    ok: bool,
    grant_id: String,
    plugin_instance_id: String,
    active_fingerprint: String,
    resource_scope: NetworkResourceScope,
    policy_revision: u64,
    management_revision: u64,
    revoke_epoch: u64,
    connector_id: String,
    transport: String,
    destination: NetworkDestinationResponsePayload,
    runtime_generation_id: String,
    target_classifier_version: String,
    expires_at: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(dead_code)]
struct NetworkExecuteSuccessResponsePayload {
    ok: bool,
    transport: String,
    destination: NetworkDestinationResponsePayload,
    status_code: Option<u16>,
    headers: Option<HashMap<String, Vec<String>>>,
    message_type: Option<String>,
    body_base64: Option<String>,
    payload_base64: Option<String>,
    stream_id: Option<String>,
    bytes_read: Option<u64>,
    chunk_count: Option<u64>,
    grant_id: String,
    connector_id: String,
    runtime_generation_id: String,
}

fn parse_hostcall_response_frame<T: DeserializeOwned>(
    input: &str,
    expected_frame_type: &'static str,
) -> IpcResult<(RawIPCFrame, HostcallResponsePayload<T>)> {
    let frame = parse_raw_frame(input)?;
    if frame.ipc_version != RUST_IPC_VERSION {
        return Err(protocol_violation("unsupported ipc_version"));
    }
    if frame.frame_type != expected_frame_type {
        return Err(protocol_violation(
            "unexpected hostcall response frame type",
        ));
    }
    if frame.request_id.trim().is_empty() {
        return Err(invalid_field("hostcall response request_id"));
    }
    let runtime_generation_id = frame
        .runtime_generation_id
        .as_deref()
        .filter(|value| !value.trim().is_empty())
        .ok_or_else(|| missing_field("hostcall response runtime_generation_id"))?;
    if runtime_generation_id.trim().is_empty() {
        return Err(invalid_field("hostcall response runtime_generation_id"));
    }
    let discriminator: BooleanResponseDiscriminator = serde_json::from_str(frame.payload.get())
        .map_err(|_| decode_failed("hostcall response discriminator"))?;
    let payload = if discriminator.ok {
        HostcallResponsePayload::Success(
            serde_json::from_str(frame.payload.get())
                .map_err(|_| decode_failed("hostcall success response payload"))?,
        )
    } else {
        HostcallResponsePayload::Failure(
            serde_json::from_str(frame.payload.get())
                .map_err(|_| decode_failed("hostcall failure response payload"))?,
        )
    };
    Ok((frame, payload))
}

fn validate_hostcall_response_identity(
    frame: &RawIPCFrame,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    _label: &'static str,
) -> IpcResult<()> {
    if frame.request_id != expected_request_id {
        return Err(protocol_violation("hostcall response request_id mismatch"));
    }
    if frame.runtime_generation_id.as_deref() != Some(expected_runtime_generation_id) {
        return Err(protocol_violation(
            "hostcall response runtime_generation_id mismatch",
        ));
    }
    Ok(())
}

fn validated_hostcall_failure(failure: HostcallFailureResponsePayload) -> IpcResult<IpcError> {
    if failure.ok {
        return Err(protocol_violation(
            "hostcall failure response ok must be false",
        ));
    }
    if failure.error_origin != ERROR_ORIGIN_HOSTCALL {
        return Err(protocol_violation(
            "hostcall response error_origin must be hostcall",
        ));
    }
    let code = failure.code.trim();
    if !is_stable_worker_error_code(code) {
        return Err(invalid_field("hostcall response code"));
    }
    let message = failure.message.trim();
    if message.is_empty() || message.len() > 4096 {
        return Err(invalid_field("hostcall response message"));
    }
    Ok(IpcError::RemoteFailure {
        code: code.to_string(),
    })
}

pub fn open_handle_frame(
    request_id: &str,
    runtime_generation_id: &str,
    identity: &WorkerInvocationIdentity,
) -> String {
    format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"package_hash\":\"{}\",\"artifact\":\"{}\",\"artifact_sha256\":\"{}\"}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_OPEN_HANDLE,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&identity.package_hash),
        escape_json_string(&identity.artifact),
        escape_json_string(&identity.artifact_sha256)
    )
}

pub fn compile_flight_register_frame(
    parent_request_id: &str,
    runtime_generation_id: &str,
    identity: &WorkerInvocationIdentity,
) -> String {
    compile_flight_lifecycle_frame(
        FRAME_TYPE_COMPILE_FLIGHT_REGISTER,
        parent_request_id,
        runtime_generation_id,
        identity,
    )
}

pub fn compile_flight_complete_frame(
    parent_request_id: &str,
    runtime_generation_id: &str,
    identity: &WorkerInvocationIdentity,
) -> String {
    compile_flight_lifecycle_frame(
        FRAME_TYPE_COMPILE_FLIGHT_COMPLETE,
        parent_request_id,
        runtime_generation_id,
        identity,
    )
}

fn compile_flight_lifecycle_frame(
    frame_type: &str,
    parent_request_id: &str,
    runtime_generation_id: &str,
    identity: &WorkerInvocationIdentity,
) -> String {
    let artifact_request_id = format!("{parent_request_id}:artifact");
    let request_id = if frame_type == FRAME_TYPE_COMPILE_FLIGHT_REGISTER {
        format!("{artifact_request_id}:register")
    } else {
        format!("{artifact_request_id}:complete")
    };
    format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"parent_request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"artifact_request_id\":\"{}\",\"package_hash\":\"{}\",\"artifact\":\"{}\",\"artifact_sha256\":\"{}\",\"wasm_abi_version\":\"{}\"}}}}",
        RUST_IPC_VERSION,
        frame_type,
        escape_json_string(&request_id),
        escape_json_string(parent_request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&artifact_request_id),
        escape_json_string(&identity.package_hash),
        escape_json_string(&identity.artifact),
        escape_json_string(&identity.artifact_sha256),
        WASM_ABI_VERSION,
    )
}

pub fn validate_open_handle_response(
    input: &str,
    expected_request_id: &str,
    expected_parent_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_identity: &WorkerInvocationIdentity,
) -> IpcResult<()> {
    parse_open_handle_success_response(
        input,
        expected_request_id,
        expected_parent_request_id,
        expected_runtime_generation_id,
        expected_identity,
    )?;
    Ok(())
}

fn parse_open_handle_success_response(
    input: &str,
    expected_request_id: &str,
    expected_parent_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_identity: &WorkerInvocationIdentity,
) -> IpcResult<OpenHandleSuccessResponsePayload> {
    let (frame, response) = parse_hostcall_response_frame::<OpenHandleSuccessResponsePayload>(
        input,
        FRAME_TYPE_OPEN_HANDLE,
    )?;
    validate_hostcall_response_identity(
        &frame,
        expected_request_id,
        expected_runtime_generation_id,
        "open_handle",
    )?;
    if frame.parent_request_id.as_deref() != Some(expected_parent_request_id) {
        return Err(protocol_violation("open_handle parent_request_id mismatch"));
    }
    let success = match response {
        HostcallResponsePayload::Success(success) if success.ok => success,
        HostcallResponsePayload::Success(_) => {
            return Err(protocol_violation(
                "open_handle success response ok must be true",
            ));
        }
        HostcallResponsePayload::Failure(failure) => {
            return Err(validated_hostcall_failure(failure)?);
        }
    };
    if success.package_hash != expected_identity.package_hash
        || success.artifact != expected_identity.artifact
        || success.sha256 != expected_identity.artifact_sha256
    {
        return Err(protocol_violation("open_handle artifact identity mismatch"));
    }
    if success.content_base64.trim().is_empty() {
        return Err(invalid_field("content_base64"));
    }
    Ok(success)
}

pub fn open_handle_content_base64(
    input: &str,
    expected_request_id: &str,
    expected_parent_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_identity: &WorkerInvocationIdentity,
) -> IpcResult<String> {
    let success = parse_open_handle_success_response(
        input,
        expected_request_id,
        expected_parent_request_id,
        expected_runtime_generation_id,
        expected_identity,
    )?;
    Ok(success.content_base64)
}

pub fn worker_success_result_json(
    identity: &WorkerInvocationIdentity,
    wasm_byte_len: usize,
    storage_file_result_json: Option<&str>,
    storage_kv_result_json: Option<&str>,
    storage_sqlite_result_json: Option<&str>,
    network_execute_result_json: Option<&str>,
) -> String {
    worker_success_result_json_with_network_results(
        identity,
        wasm_byte_len,
        storage_file_result_json,
        storage_kv_result_json,
        storage_sqlite_result_json,
        network_execute_result_json.into_iter().collect(),
    )
}

pub fn worker_success_result_json_with_network_results(
    identity: &WorkerInvocationIdentity,
    wasm_byte_len: usize,
    storage_file_result_json: Option<&str>,
    storage_kv_result_json: Option<&str>,
    storage_sqlite_result_json: Option<&str>,
    network_execute_result_jsons: Vec<&str>,
) -> String {
    let storage_file = storage_file_result_json
        .map(|result| format!(",\"storage_file\":{result}"))
        .unwrap_or_default();
    let storage_kv = storage_kv_result_json
        .map(|result| format!(",\"storage_kv\":{result}"))
        .unwrap_or_default();
    let storage_sqlite = storage_sqlite_result_json
        .map(|result| format!(",\"storage_sqlite\":{result}"))
        .unwrap_or_default();
    let stream_id = first_network_stream_id(&network_execute_result_jsons)
        .map(|value| format!(",\"stream_id\":\"{}\"", escape_json_string(&value)))
        .unwrap_or_default();
    let network_execute = network_success_fields(network_execute_result_jsons);
    format!(
        "{{\"data\":{{\"method\":\"{}\",\"worker_id\":\"{}\",\"backend\":\"executed wasm worker scaffold\",\"transport\":\"rust runtime ipc\",\"wasm_abi\":\"{}\",\"wasm_byte_len\":{}{}{}{}{}}}{}}}",
        escape_json_string(&identity.method),
        escape_json_string(&identity.worker_id),
        WASM_ABI_VERSION,
        wasm_byte_len,
        storage_file,
        storage_kv,
        storage_sqlite,
        network_execute,
        stream_id
    )
}

fn network_success_fields(results: Vec<&str>) -> String {
    let mut fields = String::new();
    for (index, result) in results.into_iter().enumerate() {
        let field = if index == 0 {
            "network_execute".to_string()
        } else {
            format!(
                "network_execute_{}",
                network_result_transport(result)
                    .filter(|transport| !transport.is_empty())
                    .unwrap_or_else(|| index.to_string())
            )
        };
        fields.push_str(&format!(",\"{}\":{}", escape_json_string(&field), result));
    }
    fields
}

#[derive(Deserialize)]
struct NetworkResultProjection {
    transport: Option<String>,
    stream_id: Option<String>,
}

fn parse_network_result_projection(result: &str) -> Option<NetworkResultProjection> {
    serde_json::from_str(result).ok()
}

fn network_result_transport(result: &str) -> Option<String> {
    parse_network_result_projection(result)
        .and_then(|result| result.transport)
        .map(|value| {
            value
                .chars()
                .map(|ch| {
                    if ch.is_ascii_alphanumeric() {
                        ch.to_ascii_lowercase()
                    } else {
                        '_'
                    }
                })
                .collect::<String>()
                .trim_matches('_')
                .to_string()
        })
}

fn first_network_stream_id(results: &[&str]) -> Option<String> {
    results
        .iter()
        .filter_map(|result| parse_network_result_projection(result)?.stream_id)
        .find(|stream_id| !stream_id.trim().is_empty())
}

pub fn storage_file_payload_json(input: &str, expected_operation: &str) -> IpcResult<String> {
    match expected_operation {
        "read" => successful_hostcall_payload_json::<StorageFileReadSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            |payload| payload.ok,
        ),
        "write" => successful_hostcall_payload_json::<StorageFileWriteSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            |payload| payload.ok,
        ),
        "delete" => successful_hostcall_payload_json::<StorageFileDeleteSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            |payload| payload.ok,
        ),
        "list" => successful_hostcall_payload_json::<StorageFileListSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            |payload| payload.ok,
        ),
        _ => Err(invalid_field("storage_file response operation")),
    }
}

pub fn storage_kv_payload_json(input: &str, expected_operation: &str) -> IpcResult<String> {
    match expected_operation {
        "get" => successful_hostcall_payload_json::<StorageKVGetSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            |payload| payload.ok,
        ),
        "put" => successful_hostcall_payload_json::<StorageKVPutSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            |payload| payload.ok,
        ),
        "delete" => successful_hostcall_payload_json::<StorageKVDeleteSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            |payload| payload.ok,
        ),
        "list" => successful_hostcall_payload_json::<StorageKVListSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            |payload| payload.ok,
        ),
        _ => Err(invalid_field("storage_kv response operation")),
    }
}

pub fn storage_sqlite_payload_json(input: &str, expected_operation: &str) -> IpcResult<String> {
    match expected_operation {
        "exec" => successful_hostcall_payload_json::<StorageSQLiteExecSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_SQLITE,
            |payload| payload.ok,
        ),
        "query" => successful_hostcall_payload_json::<StorageSQLiteQuerySuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_SQLITE,
            |payload| {
                payload.ok
                    && payload
                        .rows
                        .iter()
                        .flatten()
                        .all(StorageSQLiteValueResponsePayload::is_exactly_typed)
            },
        ),
        _ => Err(invalid_field("storage_sqlite response operation")),
    }
}

pub fn network_execute_payload_json(input: &str) -> IpcResult<String> {
    successful_hostcall_payload_json::<NetworkExecuteSuccessResponsePayload, _>(
        input,
        FRAME_TYPE_NETWORK_EXECUTE,
        |payload| payload.ok,
    )
}

fn successful_hostcall_payload_json<T, F>(
    input: &str,
    frame_type: &'static str,
    is_success: F,
) -> IpcResult<String>
where
    T: DeserializeOwned,
    F: FnOnce(&T) -> bool,
{
    let (frame, response) = parse_hostcall_response_frame::<T>(input, frame_type)?;
    match response {
        HostcallResponsePayload::Success(payload) if is_success(&payload) => {
            Ok(frame.payload.get().to_string())
        }
        HostcallResponsePayload::Success(_) => Err(protocol_violation(
            "hostcall success response ok must be true",
        )),
        HostcallResponsePayload::Failure(failure) => Err(validated_hostcall_failure(failure)?),
    }
}

fn parse_validated_hostcall_success<T, F>(
    input: &str,
    frame_type: &'static str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    is_success: F,
) -> IpcResult<T>
where
    T: DeserializeOwned,
    F: FnOnce(&T) -> bool,
{
    let (frame, response) = parse_hostcall_response_frame::<T>(input, frame_type)?;
    validate_hostcall_response_identity(
        &frame,
        expected_request_id,
        expected_runtime_generation_id,
        frame_type,
    )?;
    match response {
        HostcallResponsePayload::Success(payload) if is_success(&payload) => Ok(payload),
        HostcallResponsePayload::Success(_) => Err(protocol_violation(
            "hostcall success response ok must be true",
        )),
        HostcallResponsePayload::Failure(failure) => Err(validated_hostcall_failure(failure)?),
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HandleGrantValidationRequest {
    pub handle_grant_token: String,
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub owner_session_hash: String,
    pub owner_user_hash: String,
    pub owner_env_hash: String,
    pub session_channel_id_hash: String,
    pub handle_id: String,
    pub method: String,
    pub resource_scope: NetworkResourceScope,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
}

pub fn validate_handle_grant_frame(
    request_id: &str,
    runtime_generation_id: &str,
    req: &HandleGrantValidationRequest,
) -> IpcResult<String> {
    for (value, field) in [
        (request_id, "request_id"),
        (runtime_generation_id, "runtime_generation_id"),
        (&req.handle_grant_token, "handle_grant_token"),
        (&req.plugin_instance_id, "plugin_instance_id"),
        (&req.active_fingerprint, "active_fingerprint"),
        (&req.runtime_instance_id, "runtime_instance_id"),
        (&req.runtime_generation_id, "runtime_generation_id"),
        (&req.runtime_shard_id, "runtime_shard_id"),
        (&req.handle_id, "handle_id"),
        (&req.method, "method"),
    ] {
        if value.trim().is_empty() {
            return Err(invalid_field(field));
        }
    }
    if runtime_generation_id != req.runtime_generation_id {
        return Err(protocol_violation(
            "validate_handle_grant runtime_generation_id mismatch",
        ));
    }
    if !req.resource_scope.valid() {
        return Err(invalid_field("handle grant resource scope"));
    }
    if req.owner_session_hash.trim().is_empty()
        || req.owner_user_hash.trim().is_empty()
        || req.owner_env_hash.trim().is_empty()
        || req.session_channel_id_hash.trim().is_empty()
    {
        return Err(invalid_field("handle grant session audience"));
    }
    if req.resource_scope.owner_env_hash != req.owner_env_hash
        || (req.resource_scope.kind == "user"
            && req.resource_scope.owner_user_hash != req.owner_user_hash)
    {
        return Err(invalid_field("handle grant resource scope"));
    }
    validate_revision_binding(
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
    )?;
    let resource_scope = serde_json::to_string(&req.resource_scope)
        .map_err(|_| encode_failed("handle grant resource scope"))?;
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"handle_grant_token\":\"{}\",\"plugin_instance_id\":\"{}\",\"active_fingerprint\":\"{}\",\"runtime_instance_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"runtime_shard_id\":\"{}\",\"owner_session_hash\":\"{}\",\"owner_user_hash\":\"{}\",\"owner_env_hash\":\"{}\",\"session_channel_id_hash\":\"{}\",\"handle_id\":\"{}\",\"method\":\"{}\",\"resource_scope\":{},\"policy_revision\":{},\"management_revision\":{},\"revoke_epoch\":{}}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_VALIDATE_HANDLE_GRANT,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&req.handle_grant_token),
        escape_json_string(&req.plugin_instance_id),
        escape_json_string(&req.active_fingerprint),
        escape_json_string(&req.runtime_instance_id),
        escape_json_string(&req.runtime_generation_id),
        escape_json_string(&req.runtime_shard_id),
        escape_json_string(&req.owner_session_hash),
        escape_json_string(&req.owner_user_hash),
        escape_json_string(&req.owner_env_hash),
        escape_json_string(&req.session_channel_id_hash),
        escape_json_string(&req.handle_id),
        escape_json_string(&req.method),
        resource_scope,
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch
    ))
}

pub fn validate_handle_grant_response(
    input: &str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_handle_id: &str,
    expected_method: &str,
    expected_resource_scope: &NetworkResourceScope,
) -> IpcResult<()> {
    let (frame, response) = parse_hostcall_response_frame::<HandleGrantSuccessResponsePayload>(
        input,
        FRAME_TYPE_VALIDATE_HANDLE_GRANT,
    )?;
    validate_hostcall_response_identity(
        &frame,
        expected_request_id,
        expected_runtime_generation_id,
        "validate_handle_grant",
    )?;
    let success = match response {
        HostcallResponsePayload::Success(success) if success.ok => success,
        HostcallResponsePayload::Success(_) => {
            return Err(protocol_violation(
                "validate_handle_grant success response ok must be true",
            ));
        }
        HostcallResponsePayload::Failure(failure) => {
            return Err(validated_hostcall_failure(failure)?);
        }
    };
    if success.handle_id != expected_handle_id || success.method != expected_method {
        return Err(protocol_violation(
            "validate_handle_grant audience mismatch",
        ));
    }
    if success.runtime_generation_id != expected_runtime_generation_id {
        return Err(protocol_violation(
            "validate_handle_grant payload runtime_generation_id mismatch",
        ));
    }
    if !success.resource_scope.valid() || success.resource_scope != *expected_resource_scope {
        return Err(protocol_violation(
            "validate_handle_grant payload resource_scope mismatch",
        ));
    }
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageFileRequest {
    pub handle_grant_token: String,
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub handle_id: String,
    pub method: String,
    pub resource_scope: NetworkResourceScope,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
    pub operation: String,
    pub store_id: String,
    pub path: String,
    pub data_base64: String,
    pub max_bytes: u64,
    pub max_entries: u64,
    pub recursive: bool,
}

pub fn storage_file_frame(
    request_id: &str,
    runtime_generation_id: &str,
    req: &StorageFileRequest,
) -> IpcResult<String> {
    if !req.resource_scope.valid() {
        return Err(invalid_field("storage file resource scope"));
    }
    validate_revision_binding(
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
    )?;
    let resource_scope = serde_json::to_string(&req.resource_scope)
        .map_err(|_| encode_failed("storage file resource scope"))?;
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"handle_grant_token\":\"{}\",\"plugin_instance_id\":\"{}\",\"active_fingerprint\":\"{}\",\"runtime_instance_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"runtime_shard_id\":\"{}\",\"handle_id\":\"{}\",\"method\":\"{}\",\"resource_scope\":{},\"policy_revision\":{},\"management_revision\":{},\"revoke_epoch\":{},\"operation\":\"{}\",\"store_id\":\"{}\",\"path\":\"{}\",\"data_base64\":\"{}\",\"max_bytes\":{},\"max_entries\":{},\"recursive\":{}}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_STORAGE_FILE,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&req.handle_grant_token),
        escape_json_string(&req.plugin_instance_id),
        escape_json_string(&req.active_fingerprint),
        escape_json_string(&req.runtime_instance_id),
        escape_json_string(&req.runtime_generation_id),
        escape_json_string(&req.runtime_shard_id),
        escape_json_string(&req.handle_id),
        escape_json_string(&req.method),
        resource_scope,
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
        escape_json_string(&req.operation),
        escape_json_string(&req.store_id),
        escape_json_string(&req.path),
        escape_json_string(&req.data_base64),
        req.max_bytes,
        req.max_entries,
        if req.recursive { "true" } else { "false" }
    ))
}

pub fn validate_storage_file_response(
    input: &str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_operation: &str,
) -> IpcResult<()> {
    match expected_operation {
        "read" => parse_validated_hostcall_success::<StorageFileReadSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "write" => parse_validated_hostcall_success::<StorageFileWriteSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "delete" => parse_validated_hostcall_success::<StorageFileDeleteSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "list" => parse_validated_hostcall_success::<StorageFileListSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_FILE,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        _ => Err(invalid_field("storage_file response operation")),
    }?;
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageKVRequest {
    pub handle_grant_token: String,
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub handle_id: String,
    pub method: String,
    pub resource_scope: NetworkResourceScope,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
    pub operation: String,
    pub store_id: String,
    pub key: String,
    pub value_base64: String,
    pub prefix: String,
    pub max_bytes: u64,
    pub max_entries: u64,
}

pub fn storage_kv_frame(
    request_id: &str,
    runtime_generation_id: &str,
    req: &StorageKVRequest,
) -> IpcResult<String> {
    if !req.resource_scope.valid() {
        return Err(invalid_field("storage kv resource scope"));
    }
    validate_revision_binding(
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
    )?;
    let resource_scope = serde_json::to_string(&req.resource_scope)
        .map_err(|_| encode_failed("storage kv resource scope"))?;
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"handle_grant_token\":\"{}\",\"plugin_instance_id\":\"{}\",\"active_fingerprint\":\"{}\",\"runtime_instance_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"runtime_shard_id\":\"{}\",\"handle_id\":\"{}\",\"method\":\"{}\",\"resource_scope\":{},\"policy_revision\":{},\"management_revision\":{},\"revoke_epoch\":{},\"operation\":\"{}\",\"store_id\":\"{}\",\"key\":\"{}\",\"value_base64\":\"{}\",\"prefix\":\"{}\",\"max_bytes\":{},\"max_entries\":{}}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_STORAGE_KV,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&req.handle_grant_token),
        escape_json_string(&req.plugin_instance_id),
        escape_json_string(&req.active_fingerprint),
        escape_json_string(&req.runtime_instance_id),
        escape_json_string(&req.runtime_generation_id),
        escape_json_string(&req.runtime_shard_id),
        escape_json_string(&req.handle_id),
        escape_json_string(&req.method),
        resource_scope,
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
        escape_json_string(&req.operation),
        escape_json_string(&req.store_id),
        escape_json_string(&req.key),
        escape_json_string(&req.value_base64),
        escape_json_string(&req.prefix),
        req.max_bytes,
        req.max_entries
    ))
}

pub fn validate_storage_kv_response(
    input: &str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_operation: &str,
) -> IpcResult<()> {
    match expected_operation {
        "get" => parse_validated_hostcall_success::<StorageKVGetSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "put" => parse_validated_hostcall_success::<StorageKVPutSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "delete" => parse_validated_hostcall_success::<StorageKVDeleteSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "list" => parse_validated_hostcall_success::<StorageKVListSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_KV,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        _ => Err(invalid_field("storage_kv response operation")),
    }?;
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageSQLiteRequest {
    pub handle_grant_token: String,
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub handle_id: String,
    pub method: String,
    pub resource_scope: NetworkResourceScope,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
    pub operation: String,
    pub store_id: String,
    pub database: String,
    pub sql: String,
    pub args_json: String,
    pub max_rows: u64,
    pub max_response_bytes: u64,
    pub timeout_ms: u64,
}

pub fn storage_sqlite_frame(
    request_id: &str,
    runtime_generation_id: &str,
    req: &StorageSQLiteRequest,
) -> IpcResult<String> {
    if !req.resource_scope.valid() {
        return Err(invalid_field("storage sqlite resource scope"));
    }
    validate_revision_binding(
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
    )?;
    let resource_scope = serde_json::to_string(&req.resource_scope)
        .map_err(|_| encode_failed("storage sqlite resource scope"))?;
    let args_json = if req.args_json.trim().is_empty() {
        "[]"
    } else {
        req.args_json.trim()
    };
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"handle_grant_token\":\"{}\",\"plugin_instance_id\":\"{}\",\"active_fingerprint\":\"{}\",\"runtime_instance_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"runtime_shard_id\":\"{}\",\"handle_id\":\"{}\",\"method\":\"{}\",\"resource_scope\":{},\"policy_revision\":{},\"management_revision\":{},\"revoke_epoch\":{},\"operation\":\"{}\",\"store_id\":\"{}\",\"database\":\"{}\",\"sql\":\"{}\",\"args\":{},\"max_rows\":{},\"max_response_bytes\":{},\"timeout_ms\":{}}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_STORAGE_SQLITE,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&req.handle_grant_token),
        escape_json_string(&req.plugin_instance_id),
        escape_json_string(&req.active_fingerprint),
        escape_json_string(&req.runtime_instance_id),
        escape_json_string(&req.runtime_generation_id),
        escape_json_string(&req.runtime_shard_id),
        escape_json_string(&req.handle_id),
        escape_json_string(&req.method),
        resource_scope,
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
        escape_json_string(&req.operation),
        escape_json_string(&req.store_id),
        escape_json_string(&req.database),
        escape_json_string(&req.sql),
        args_json,
        req.max_rows,
        req.max_response_bytes,
        req.timeout_ms
    ))
}

pub fn validate_storage_sqlite_response(
    input: &str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_operation: &str,
) -> IpcResult<()> {
    match expected_operation {
        "exec" => parse_validated_hostcall_success::<StorageSQLiteExecSuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_SQLITE,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| payload.ok,
        )
        .map(|_| ()),
        "query" => parse_validated_hostcall_success::<StorageSQLiteQuerySuccessResponsePayload, _>(
            input,
            FRAME_TYPE_STORAGE_SQLITE,
            expected_request_id,
            expected_runtime_generation_id,
            |payload| {
                payload.ok
                    && payload
                        .rows
                        .iter()
                        .flatten()
                        .all(StorageSQLiteValueResponsePayload::is_exactly_typed)
            },
        )
        .map(|_| ()),
        _ => Err(invalid_field("storage_sqlite response operation")),
    }?;
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NetworkResourceScope {
    pub kind: String,
    pub owner_env_hash: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub owner_user_hash: String,
}

impl NetworkResourceScope {
    fn valid(&self) -> bool {
        valid_owner_hash(&self.owner_env_hash)
            && match self.kind.as_str() {
                "user" => valid_owner_hash(&self.owner_user_hash),
                "environment" => self.owner_user_hash.is_empty(),
                _ => false,
            }
    }
}

fn valid_owner_hash(value: &str) -> bool {
    let bytes = value.as_bytes();
    (1..=256).contains(&bytes.len())
        && bytes[0].is_ascii_alphanumeric()
        && bytes[1..]
            .iter()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(*byte, b'.' | b'_' | b':' | b'-'))
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NetworkGrantRequest {
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub resource_scope: NetworkResourceScope,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
    pub connector_id: String,
    pub transport: String,
    pub destination: String,
    pub ttl_ms: u64,
}

pub fn network_grant_frame(
    request_id: &str,
    runtime_generation_id: &str,
    req: &NetworkGrantRequest,
) -> IpcResult<String> {
    if !req.resource_scope.valid() {
        return Err(invalid_field("network resource scope"));
    }
    validate_revision_binding(
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
    )?;
    let resource_scope = serde_json::to_string(&req.resource_scope)
        .map_err(|_| encode_failed("network resource scope"))?;
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"plugin_instance_id\":\"{}\",\"active_fingerprint\":\"{}\",\"resource_scope\":{},\"runtime_instance_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"runtime_shard_id\":\"{}\",\"policy_revision\":{},\"management_revision\":{},\"revoke_epoch\":{},\"connector_id\":\"{}\",\"transport\":\"{}\",\"destination\":\"{}\",\"ttl_ms\":{}}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_NETWORK_GRANT,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&req.plugin_instance_id),
        escape_json_string(&req.active_fingerprint),
        resource_scope,
        escape_json_string(&req.runtime_instance_id),
        escape_json_string(&req.runtime_generation_id),
        escape_json_string(&req.runtime_shard_id),
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
        escape_json_string(&req.connector_id),
        escape_json_string(&req.transport),
        escape_json_string(&req.destination),
        req.ttl_ms
    ))
}

pub fn validate_network_grant_response(
    input: &str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_connector_id: &str,
    expected_transport: &str,
    expected_resource_scope: &NetworkResourceScope,
) -> IpcResult<()> {
    let success = parse_validated_hostcall_success::<NetworkGrantSuccessResponsePayload, _>(
        input,
        FRAME_TYPE_NETWORK_GRANT,
        expected_request_id,
        expected_runtime_generation_id,
        |payload| payload.ok,
    )?;
    let grant_suffix = success.grant_id.strip_prefix("netgrant_");
    if grant_suffix.is_none_or(|suffix| {
        suffix.len() != 32 || !suffix.bytes().all(|byte| byte.is_ascii_hexdigit())
    }) {
        return Err(invalid_field("network grant id"));
    }
    if success.connector_id != expected_connector_id || success.transport != expected_transport {
        return Err(protocol_violation("network_grant audience mismatch"));
    }
    if !success.resource_scope.valid() || success.resource_scope != *expected_resource_scope {
        return Err(protocol_violation("network_grant resource scope mismatch"));
    }
    validate_revision_binding(
        success.policy_revision,
        success.management_revision,
        success.revoke_epoch,
    )?;
    if success.runtime_generation_id != expected_runtime_generation_id {
        return Err(protocol_violation(
            "network_grant payload runtime_generation_id mismatch",
        ));
    }
    if success.target_classifier_version != "target-classifier-v2" {
        return Err(protocol_violation(
            "network_grant target classifier version mismatch",
        ));
    }
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NetworkExecuteRequest {
    pub plugin_id: String,
    pub plugin_instance_id: String,
    pub active_fingerprint: String,
    pub resource_scope: NetworkResourceScope,
    pub runtime_instance_id: String,
    pub runtime_generation_id: String,
    pub runtime_shard_id: String,
    pub policy_revision: u64,
    pub management_revision: u64,
    pub revoke_epoch: u64,
    pub connector_id: String,
    pub transport: String,
    pub destination: String,
    pub ttl_ms: u64,
    pub operation: String,
    pub method: String,
    pub path: String,
    pub query_json: String,
    pub headers_json: String,
    pub message_type: String,
    pub body_base64: String,
    pub payload_base64: String,
    pub max_request_bytes: u64,
    pub max_response_bytes: u64,
    pub max_chunk_bytes: u64,
    pub max_buffered_bytes: u64,
    pub timeout_ms: u64,
    pub stream_id: String,
    pub stream_method: String,
    pub stream_effect: String,
    pub stream_execution: String,
    pub surface_instance_id: String,
    pub owner_session_hash: String,
    pub owner_user_hash: String,
    pub owner_env_hash: String,
    pub session_channel_id_hash: String,
    pub bridge_channel_id: String,
    pub content_type: String,
}

pub fn network_execute_frame(
    request_id: &str,
    runtime_generation_id: &str,
    req: &NetworkExecuteRequest,
) -> IpcResult<String> {
    if !req.resource_scope.valid() {
        return Err(invalid_field("network resource scope"));
    }
    validate_revision_binding(
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
    )?;
    let query_json = if req.query_json.trim().is_empty() {
        "{}"
    } else {
        req.query_json.trim()
    };
    let headers_json = if req.headers_json.trim().is_empty() {
        "{}"
    } else {
        req.headers_json.trim()
    };
    let query: serde_json::Value =
        serde_json::from_str(query_json).map_err(|_| invalid_field("network execute query"))?;
    if !query.is_object() {
        return Err(invalid_field("network execute query"));
    }
    let headers: serde_json::Value =
        serde_json::from_str(headers_json).map_err(|_| invalid_field("network execute headers"))?;
    if !headers.is_object() {
        return Err(invalid_field("network execute headers"));
    }
    let resource_scope = serde_json::to_string(&req.resource_scope)
        .map_err(|_| encode_failed("network resource scope"))?;
    Ok(format!(
        "{{\"ipc_version\":\"{}\",\"frame_type\":\"{}\",\"request_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"payload\":{{\"plugin_id\":\"{}\",\"plugin_instance_id\":\"{}\",\"active_fingerprint\":\"{}\",\"resource_scope\":{},\"runtime_instance_id\":\"{}\",\"runtime_generation_id\":\"{}\",\"runtime_shard_id\":\"{}\",\"policy_revision\":{},\"management_revision\":{},\"revoke_epoch\":{},\"connector_id\":\"{}\",\"transport\":\"{}\",\"destination\":\"{}\",\"ttl_ms\":{},\"operation\":\"{}\",\"method\":\"{}\",\"path\":\"{}\",\"query\":{},\"headers\":{},\"message_type\":\"{}\",\"body_base64\":\"{}\",\"payload_base64\":\"{}\",\"max_request_bytes\":{},\"max_response_bytes\":{},\"max_chunk_bytes\":{},\"max_buffered_bytes\":{},\"timeout_ms\":{},\"stream_id\":\"{}\",\"stream_method\":\"{}\",\"stream_effect\":\"{}\",\"stream_execution\":\"{}\",\"surface_instance_id\":\"{}\",\"owner_session_hash\":\"{}\",\"owner_user_hash\":\"{}\",\"owner_env_hash\":\"{}\",\"session_channel_id_hash\":\"{}\",\"bridge_channel_id\":\"{}\",\"content_type\":\"{}\"}}}}",
        RUST_IPC_VERSION,
        FRAME_TYPE_NETWORK_EXECUTE,
        escape_json_string(request_id),
        escape_json_string(runtime_generation_id),
        escape_json_string(&req.plugin_id),
        escape_json_string(&req.plugin_instance_id),
        escape_json_string(&req.active_fingerprint),
        resource_scope,
        escape_json_string(&req.runtime_instance_id),
        escape_json_string(&req.runtime_generation_id),
        escape_json_string(&req.runtime_shard_id),
        req.policy_revision,
        req.management_revision,
        req.revoke_epoch,
        escape_json_string(&req.connector_id),
        escape_json_string(&req.transport),
        escape_json_string(&req.destination),
        req.ttl_ms,
        escape_json_string(&req.operation),
        escape_json_string(&req.method),
        escape_json_string(&req.path),
        query_json,
        headers_json,
        escape_json_string(&req.message_type),
        escape_json_string(&req.body_base64),
        escape_json_string(&req.payload_base64),
        req.max_request_bytes,
        req.max_response_bytes,
        req.max_chunk_bytes,
        req.max_buffered_bytes,
        req.timeout_ms,
        escape_json_string(&req.stream_id),
        escape_json_string(&req.stream_method),
        escape_json_string(&req.stream_effect),
        escape_json_string(&req.stream_execution),
        escape_json_string(&req.surface_instance_id),
        escape_json_string(&req.owner_session_hash),
        escape_json_string(&req.owner_user_hash),
        escape_json_string(&req.owner_env_hash),
        escape_json_string(&req.session_channel_id_hash),
        escape_json_string(&req.bridge_channel_id),
        escape_json_string(&req.content_type)
    ))
}

pub fn validate_network_execute_response(
    input: &str,
    expected_request_id: &str,
    expected_runtime_generation_id: &str,
    expected_connector_id: &str,
    expected_transport: &str,
) -> IpcResult<()> {
    let success = parse_validated_hostcall_success::<NetworkExecuteSuccessResponsePayload, _>(
        input,
        FRAME_TYPE_NETWORK_EXECUTE,
        expected_request_id,
        expected_runtime_generation_id,
        |payload| payload.ok,
    )?;
    if success.connector_id != expected_connector_id || success.transport != expected_transport {
        return Err(protocol_violation("network_execute audience mismatch"));
    }
    if success.runtime_generation_id != expected_runtime_generation_id {
        return Err(protocol_violation(
            "network_execute payload runtime_generation_id mismatch",
        ));
    }
    Ok(())
}

pub fn validate_hello_frame(input: &str) -> IpcResult<(String, String, String)> {
    let parsed = parse_hello_frame(input)?;
    Ok((
        parsed.request_id,
        parsed.runtime_generation_id,
        parsed.channel_nonce,
    ))
}

pub fn parse_hello_frame(input: &str) -> IpcResult<HelloFrame> {
    let frame: RawIPCFrame = serde_json::from_str(input).map_err(|err| {
        if err.to_string().contains("missing field `request_id`") {
            missing_field("request_id")
        } else if err
            .to_string()
            .contains("missing field `runtime_generation_id`")
        {
            missing_field("runtime_generation_id")
        } else {
            decode_failed("hello frame")
        }
    })?;
    if frame.ipc_version != RUST_IPC_VERSION {
        return Err(protocol_violation("unsupported ipc_version"));
    }
    if frame.frame_type != FRAME_TYPE_HELLO {
        return Err(protocol_violation("expected hello frame"));
    }
    if frame.request_id.trim().is_empty() {
        return Err(invalid_field("request_id"));
    }
    let runtime_generation_id = frame
        .runtime_generation_id
        .as_deref()
        .ok_or_else(|| missing_field("runtime_generation_id"))?;
    if runtime_generation_id.trim().is_empty() {
        return Err(invalid_field("runtime_generation_id"));
    }
    let payload: HelloPayload = serde_json::from_str(frame.payload.get()).map_err(|err| {
        if err.to_string().contains("missing field `channel_nonce`") {
            missing_field("channel_nonce")
        } else {
            decode_failed("hello payload")
        }
    })?;
    let target =
        RuntimeTarget::parse(&payload.target).map_err(|_| invalid_field("hello target"))?;
    if payload.host_process_id == 0 || payload.started_unix_nano == 0 {
        return Err(invalid_field("hello process metadata"));
    }
    if payload.host_ipc_version != RUST_IPC_VERSION {
        return Err(protocol_violation("unsupported host_ipc_version"));
    }
    if payload.host_wasm_abi != WASM_ABI_VERSION {
        return Err(protocol_violation("unsupported host_wasm_abi"));
    }
    if payload.contract_set_sha256 != CONTRACT_SET_SHA256 {
        return Err(protocol_violation("contract_set_sha256 mismatch"));
    }
    if payload.channel_nonce.trim().is_empty() {
        return Err(invalid_field("channel_nonce"));
    }
    let limits = payload.limits.validate()?;
    let runtime_lease_public_keys =
        parse_runtime_lease_public_key_payloads(payload.runtime_lease_public_keys)?;
    Ok(HelloFrame {
        request_id: frame.request_id,
        runtime_generation_id: runtime_generation_id.to_string(),
        target,
        contract_set_sha256: payload.contract_set_sha256,
        channel_nonce: payload.channel_nonce,
        runtime_lease_public_keys,
        limits,
    })
}

pub fn parse_frame_identity(input: &str) -> IpcResult<FrameIdentity> {
    let frame: RawIPCFrame = serde_json::from_str(input).map_err(|err| {
        let message = err.to_string();
        if message.contains("missing field `ipc_version`") {
            missing_field("ipc_version")
        } else if message.contains("missing field `frame_type`") {
            missing_field("frame_type")
        } else if message.contains("missing field `request_id`") {
            missing_field("request_id")
        } else if message.contains("missing field `runtime_generation_id`") {
            missing_field("runtime_generation_id")
        } else if message.contains("missing field `payload`") {
            missing_field("payload")
        } else {
            decode_failed("IPC frame")
        }
    })?;
    validated_frame_identity(&frame)
}

fn validated_frame_identity(frame: &RawIPCFrame) -> IpcResult<FrameIdentity> {
    if frame.ipc_version != RUST_IPC_VERSION {
        return Err(protocol_violation("unsupported ipc_version"));
    }
    if frame.frame_type.trim().is_empty() {
        return Err(invalid_field("frame_type"));
    }
    if frame.request_id.trim().is_empty() {
        return Err(invalid_field("request_id"));
    }
    let runtime_generation_id = frame
        .runtime_generation_id
        .as_deref()
        .ok_or_else(|| missing_field("runtime_generation_id"))?;
    if runtime_generation_id.trim().is_empty() {
        return Err(invalid_field("runtime_generation_id"));
    }
    if frame
        .parent_request_id
        .as_deref()
        .is_some_and(|value| value.trim().is_empty())
    {
        return Err(invalid_field("parent_request_id"));
    }
    Ok(FrameIdentity {
        frame_type: frame.frame_type.clone(),
        request_id: frame.request_id.clone(),
        parent_request_id: frame.parent_request_id.clone(),
        runtime_generation_id: runtime_generation_id.to_string(),
    })
}

pub fn decode_runtime_input_frame(input: &str) -> IpcResult<RuntimeInputFrame> {
    let frame = parse_raw_frame(input)?;
    let identity = validated_frame_identity(&frame)?;
    match identity.frame_type.as_str() {
        FRAME_TYPE_INVOKE_WORKER => {
            let invocation = parsed_worker_invocation(&identity, frame.payload.as_ref());
            Ok(RuntimeInputFrame::InvokeWorker(Box::new(
                WorkerInvocationInput {
                    identity,
                    invocation,
                },
            )))
        }
        FRAME_TYPE_CANCEL_INVOKE => {
            if identity.parent_request_id.is_some() {
                return Err(protocol_violation(
                    "cancel_invoke must not have parent_request_id",
                ));
            }
            let payload: CancelInvokePayload = serde_json::from_str(frame.payload.get())
                .map_err(|_| decode_failed("cancel_invoke payload"))?;
            if payload.invocation_request_id.trim().is_empty() {
                return Err(invalid_field("cancel invocation_request_id"));
            }
            Ok(RuntimeInputFrame::CancelInvoke(CancelInvocationInput {
                identity,
                invocation_request_id: payload.invocation_request_id,
            }))
        }
        FRAME_TYPE_OPEN_HANDLE
        | FRAME_TYPE_VALIDATE_HANDLE_GRANT
        | FRAME_TYPE_STORAGE_FILE
        | FRAME_TYPE_STORAGE_KV
        | FRAME_TYPE_STORAGE_SQLITE
        | FRAME_TYPE_NETWORK_GRANT
        | FRAME_TYPE_NETWORK_EXECUTE => {
            if identity.parent_request_id.is_none() {
                return Err(missing_field("runtime hostcall response parent_request_id"));
            }
            Ok(RuntimeInputFrame::HostcallResponse(
                RuntimeHostcallResponseInput {
                    identity,
                    raw_frame: input.to_string(),
                },
            ))
        }
        _ => Ok(RuntimeInputFrame::Unsupported(identity)),
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkerInvocationIdentity {
    pub package_hash: String,
    pub artifact: String,
    pub artifact_sha256: String,
    pub worker_id: String,
    pub method: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WorkerResponseV2 {
    Success(String),
    Failure { code: String, message: String },
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RawWorkerResponseV2<'a> {
    ok: bool,
    #[serde(borrow)]
    data: Option<&'a serde_json::value::RawValue>,
    error_code: Option<String>,
    message: Option<String>,
}

pub fn worker_request_json_v2(input: &str) -> IpcResult<String> {
    parse_worker_invocation(input)?.worker_request_json_v2()
}

pub fn runtime_lease_memory_limit_bytes(input: &str) -> IpcResult<usize> {
    parse_worker_invocation(input)?.memory_limit_bytes()
}

pub fn worker_storage_handle_grant(input: &str, store_id: &str) -> IpcResult<String> {
    parse_worker_invocation(input)?.storage_handle_grant(store_id)
}

pub fn validate_worker_storage_broker_access(
    input: &str,
    store_id: &str,
    operation: &str,
) -> IpcResult<()> {
    parse_worker_invocation(input)?.validate_storage_broker_access(store_id, operation)
}

pub fn validate_worker_network_broker_access(
    input: &str,
    connector_id: &str,
    transport: &str,
    operation: &str,
    http_method: &str,
) -> IpcResult<()> {
    parse_worker_invocation(input)?.validate_network_broker_access(
        connector_id,
        transport,
        operation,
        http_method,
    )
}

pub fn parse_worker_response_v2(input: &str) -> IpcResult<WorkerResponseV2> {
    let response: RawWorkerResponseV2<'_> =
        serde_json::from_str(input).map_err(|_| decode_failed("worker response"))?;
    if response.ok {
        if response.error_code.is_some() || response.message.is_some() {
            return Err(protocol_violation(
                "worker success response contains failure fields",
            ));
        }
        let data = response
            .data
            .ok_or_else(|| missing_field("worker success response data"))?;
        return Ok(WorkerResponseV2::Success(data.get().to_string()));
    }
    if response.data.is_some() {
        return Err(protocol_violation(
            "worker failure response contains success data",
        ));
    }
    let error_code = response
        .error_code
        .ok_or_else(|| missing_field("worker failure response error_code"))?;
    let message = response
        .message
        .ok_or_else(|| missing_field("worker failure response message"))?;
    if !is_stable_worker_error_code(&error_code) {
        return Err(invalid_field("worker failure response error_code"));
    }
    if message.trim().is_empty() || message.len() > 4096 {
        return Err(invalid_field("worker failure response message"));
    }
    Ok(WorkerResponseV2::Failure {
        code: error_code,
        message,
    })
}

fn is_stable_worker_error_code(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 128
        && value.chars().enumerate().all(|(index, ch)| {
            ch.is_ascii_uppercase() || ch.is_ascii_digit() || (index > 0 && ch == '_')
        })
        && value
            .chars()
            .next()
            .is_some_and(|ch| ch.is_ascii_uppercase())
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct WorkerLeaseReplayKey {
    pub lease_id: String,
    pub lease_nonce: String,
    pub expires_at_unix_ms: i64,
}

pub fn parse_worker_lease_replay_key(input: &str) -> IpcResult<WorkerLeaseReplayKey> {
    parse_worker_invocation(input)?.replay_key()
}

pub fn parse_worker_invocation_identity(input: &str) -> IpcResult<WorkerInvocationIdentity> {
    parse_worker_invocation(input)?.identity()
}

pub fn worker_invocation_not_implemented_message(identity: &WorkerInvocationIdentity) -> String {
    format!(
        "runtime worker execution is not implemented for {}:{}",
        identity.worker_id, identity.method
    )
}

pub fn validate_worker_artifact_bytes(
    identity: &WorkerInvocationIdentity,
    content: &[u8],
) -> IpcResult<()> {
    let actual = format!("sha256:{}", lowercase_hex(&Sha256::digest(content)));
    if actual != identity.artifact_sha256 {
        return Err(protocol_violation(
            "worker artifact content does not match artifact_sha256",
        ));
    }
    Ok(())
}

fn is_sha256_ref(value: &str) -> bool {
    let Some(hex) = value.strip_prefix("sha256:") else {
        return false;
    };
    hex.len() == 64
        && hex
            .chars()
            .all(|ch| ch.is_ascii_hexdigit() && !ch.is_ascii_uppercase())
}

fn is_worker_artifact_path(value: &str) -> bool {
    if !value.starts_with("workers/") || !value.ends_with(".wasm") {
        return false;
    }
    if value.contains('\\') || value.contains("//") {
        return false;
    }
    value
        .split('/')
        .all(|part| !part.is_empty() && part != "." && part != "..")
}

#[cfg(test)]
mod tests {
    use super::*;
    use ed25519_dalek::{Signer, SigningKey};
    use serde_json::Value;
    use std::fs;
    use std::path::PathBuf;

    fn runtime_limits() -> RuntimeLimits {
        RuntimeLimits {
            worker_count: 8,
            queue_capacity: 32,
            per_plugin_concurrency: 4,
            module_cache_entries: 64,
            module_cache_source_bytes: 128 * 1024 * 1024,
        }
    }

    fn invalid_runtime_limits() -> RuntimeLimits {
        RuntimeLimits {
            worker_count: 1,
            per_plugin_concurrency: 2,
            ..runtime_limits()
        }
    }

    fn process_containment() -> ProcessContainmentEvidence {
        ProcessContainmentEvidence {
            schema_version: "redevplugin.process_containment.v1".to_string(),
            profile: "linux-runtime-v1".to_string(),
            seccomp_policy_sha256:
                "6305735925c1fbacaf4950df2e535d3a11cebec8ab7aa16ce37fca3c31745543".to_string(),
            no_new_privs: true,
            seccomp_tsync: true,
            process_creation_denied: true,
            reexec_denied: true,
            active: true,
        }
    }

    #[test]
    fn runtime_limit_constants_match_the_ipc_schema() {
        let schema: Value = serde_json::from_str(contract_fixture(
            redevplugin_contracts::ContractId::RUST_IPC_SCHEMA,
        ))
        .expect("IPC schema");
        let properties = schema["$defs"]["runtime_limits"]["properties"]
            .as_object()
            .expect("runtime limit properties");
        for (field, minimum, maximum) in [
            (
                "worker_count",
                MIN_RUNTIME_WORKER_COUNT,
                MAX_RUNTIME_WORKER_COUNT,
            ),
            (
                "queue_capacity",
                MIN_RUNTIME_QUEUE_CAPACITY,
                MAX_RUNTIME_QUEUE_CAPACITY,
            ),
            (
                "per_plugin_concurrency",
                MIN_RUNTIME_PER_PLUGIN_CONCURRENCY,
                MAX_RUNTIME_PER_PLUGIN_CONCURRENCY,
            ),
            (
                "module_cache_entries",
                MIN_RUNTIME_MODULE_CACHE_ENTRIES,
                MAX_RUNTIME_MODULE_CACHE_ENTRIES,
            ),
            (
                "module_cache_source_bytes",
                MIN_RUNTIME_MODULE_CACHE_SOURCE_BYTES,
                MAX_RUNTIME_MODULE_CACHE_SOURCE_BYTES,
            ),
        ] {
            let property = properties.get(field).expect("runtime limit property");
            assert_eq!(property["minimum"].as_u64(), Some(minimum as u64));
            assert_eq!(property["maximum"].as_u64(), Some(maximum as u64));
        }
    }

    #[test]
    fn runtime_limits_enforce_all_platform_bounds() {
        RuntimeLimits {
            worker_count: MAX_RUNTIME_WORKER_COUNT,
            queue_capacity: MAX_RUNTIME_QUEUE_CAPACITY,
            per_plugin_concurrency: MAX_RUNTIME_PER_PLUGIN_CONCURRENCY,
            module_cache_entries: MAX_RUNTIME_MODULE_CACHE_ENTRIES,
            module_cache_source_bytes: MAX_RUNTIME_MODULE_CACHE_SOURCE_BYTES,
        }
        .validate()
        .expect("maximum runtime limits");

        for invalid in [
            RuntimeLimits {
                worker_count: 0,
                queue_capacity: 0,
                per_plugin_concurrency: 0,
                module_cache_entries: 0,
                module_cache_source_bytes: 0,
            },
            RuntimeLimits {
                worker_count: MAX_RUNTIME_WORKER_COUNT + 1,
                ..runtime_limits()
            },
            RuntimeLimits {
                queue_capacity: MAX_RUNTIME_QUEUE_CAPACITY + 1,
                ..runtime_limits()
            },
            RuntimeLimits {
                per_plugin_concurrency: MAX_RUNTIME_PER_PLUGIN_CONCURRENCY + 1,
                ..runtime_limits()
            },
            invalid_runtime_limits(),
            RuntimeLimits {
                module_cache_entries: MAX_RUNTIME_MODULE_CACHE_ENTRIES + 1,
                ..runtime_limits()
            },
            RuntimeLimits {
                module_cache_source_bytes: MAX_RUNTIME_MODULE_CACHE_SOURCE_BYTES + 1,
                ..runtime_limits()
            },
        ] {
            assert!(matches!(
                invalid.validate(),
                Err(IpcError::ProtocolViolation { .. })
            ));
        }
    }

    fn user_resource_scope() -> NetworkResourceScope {
        NetworkResourceScope {
            kind: "user".to_string(),
            owner_env_hash: "env_hash".to_string(),
            owner_user_hash: "user_hash".to_string(),
        }
    }

    fn environment_resource_scope() -> NetworkResourceScope {
        NetworkResourceScope {
            kind: "environment".to_string(),
            owner_env_hash: "env_hash".to_string(),
            owner_user_hash: String::new(),
        }
    }

    fn handle_grant_validation_request() -> HandleGrantValidationRequest {
        HandleGrantValidationRequest {
            handle_grant_token: "handle_grant.secret".to_string(),
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            owner_session_hash: "session_hash".to_string(),
            owner_user_hash: "user_hash".to_string(),
            owner_env_hash: "env_hash".to_string(),
            session_channel_id_hash: "channel_hash".to_string(),
            handle_id: "storage:db".to_string(),
            method: "storage.sqlite".to_string(),
            resource_scope: user_resource_scope(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 3,
        }
    }

    #[test]
    fn resource_scopes_match_the_closed_owner_hash_contract() {
        let maximum_hash = format!("a{}", "b".repeat(255));
        for valid in [
            NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: maximum_hash.clone(),
                owner_user_hash: "user.hash:_-1".to_string(),
            },
            NetworkResourceScope {
                kind: "environment".to_string(),
                owner_env_hash: maximum_hash,
                owner_user_hash: String::new(),
            },
        ] {
            assert!(valid.valid(), "valid resource scope rejected: {valid:?}");
        }

        for invalid in [
            NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: String::new(),
                owner_user_hash: "user_hash".to_string(),
            },
            NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: " env_hash".to_string(),
                owner_user_hash: "user_hash".to_string(),
            },
            NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: "env/hash".to_string(),
                owner_user_hash: "user_hash".to_string(),
            },
            NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: "env_hash".to_string(),
                owner_user_hash: "user_hash ".to_string(),
            },
            NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: "env_hash".to_string(),
                owner_user_hash: String::new(),
            },
            NetworkResourceScope {
                kind: "environment".to_string(),
                owner_env_hash: "env_hash".to_string(),
                owner_user_hash: " ".to_string(),
            },
            NetworkResourceScope {
                kind: "environment".to_string(),
                owner_env_hash: "a".repeat(257),
                owner_user_hash: String::new(),
            },
            NetworkResourceScope {
                kind: "environment".to_string(),
                owner_env_hash: "\u{73af}\u{5883}".to_string(),
                owner_user_hash: String::new(),
            },
        ] {
            assert!(
                !invalid.valid(),
                "invalid resource scope accepted: {invalid:?}"
            );
        }
    }

    fn closed_worker_frame(lease: &str, invocation: &str) -> String {
        format!(
            r#"{{"ipc_version":"rust-ipc-v6","frame_type":"invoke_worker","request_id":"r1","runtime_generation_id":"g1","payload":{{"lease":{lease},"method":"worker.echo","invocation":{invocation}}}}}"#
        )
    }

    fn worker_lease_from_value(value: &serde_json::Value) -> WorkerLeasePayload {
        serde_json::from_value(value.clone()).expect("typed worker lease")
    }

    fn worker_lease_from_object(
        value: &serde_json::Map<String, serde_json::Value>,
    ) -> WorkerLeasePayload {
        worker_lease_from_value(&serde_json::Value::Object(value.clone()))
    }

    fn hello_frame(channel_nonce: Option<&str>, public_keys: &str) -> String {
        let channel_nonce = channel_nonce
            .map(|value| format!(",\"channel_nonce\":\"{value}\""))
            .unwrap_or_default();
        format!(
            r#"{{"ipc_version":"rust-ipc-v6","frame_type":"hello","request_id":"r1","runtime_generation_id":"g1","payload":{{"target":"linux/amd64","host_process_id":1,"host_ipc_version":"rust-ipc-v6","host_wasm_abi":"redevplugin-wasm-worker-v2","contract_set_sha256":"{CONTRACT_SET_SHA256}","started_unix_nano":1{channel_nonce},"runtime_lease_public_keys":{public_keys},"limits":{{"worker_count":8,"queue_capacity":32,"per_plugin_concurrency":4,"module_cache_entries":64,"module_cache_source_bytes":134217728}}}}}}"#
        )
    }

    fn hostcall_response_frame(frame_type: &str, payload: &str) -> String {
        format!(
            r#"{{"ipc_version":"rust-ipc-v6","frame_type":"{frame_type}","request_id":"r1","runtime_generation_id":"g1","payload":{payload}}}"#
        )
    }

    fn without_payload_field(frame: &str, field: &str) -> String {
        let mut value: Value = serde_json::from_str(frame).expect("hostcall response frame");
        value["payload"]
            .as_object_mut()
            .expect("hostcall response payload")
            .remove(field)
            .unwrap_or_else(|| panic!("hostcall response payload missing {field}"));
        serde_json::to_string(&value).expect("hostcall response json")
    }

    fn validate_test_hostcall_response<T: DeserializeOwned>(
        frame_type: &'static str,
        payload: &str,
    ) -> IpcResult<()> {
        let frame = hostcall_response_frame(frame_type, payload);
        let (_, response) = parse_hostcall_response_frame::<T>(&frame, frame_type)?;
        match response {
            HostcallResponsePayload::Success(_) => Ok(()),
            HostcallResponsePayload::Failure(failure) => {
                validated_hostcall_failure(failure).map(|_| ())
            }
        }
    }

    fn assert_closed_hostcall_response_union<T: DeserializeOwned>(
        frame_type: &'static str,
        success_payload: &str,
        success_field: &str,
    ) {
        validate_test_hostcall_response::<T>(frame_type, success_payload)
            .unwrap_or_else(|err| panic!("valid {frame_type} success response: {err}"));

        let success_prefix = success_payload
            .strip_suffix('}')
            .expect("success response object");
        let duplicate_success_field = success_payload.replacen(
            success_field,
            &format!("{success_field},{success_field}"),
            1,
        );
        let failure =
            r#"{"ok":false,"code":"HOSTCALL_FAILED","message":"failed","error_origin":"hostcall"}"#;
        let failure_prefix = failure.strip_suffix('}').expect("failure response object");
        let invalid = [
            format!(r#"{success_prefix},"future":true}}"#),
            success_payload.replacen(r#""ok":true"#, r#""ok":true,"ok":false"#, 1),
            success_payload.replacen(r#""ok":true"#, r#""ok":true,"OK":false"#, 1),
            duplicate_success_field,
            format!(r#"{success_prefix},"code":"HOSTCALL_FAILED"}}"#),
            format!(r#"{failure_prefix},{success_field}}}"#),
            format!(r#"{failure_prefix},"future":true}}"#),
            r#"{"ok":false,"code":"HOSTCALL_FAILED","message":"failed"}"#.to_string(),
            r#"{"ok":false,"code":"HOSTCALL_FAILED","message":"failed","error_origin":"runtime"}"#
                .to_string(),
        ];
        for payload in invalid {
            assert!(
                validate_test_hostcall_response::<T>(frame_type, &payload).is_err(),
                "{frame_type} accepted non-closed response payload {payload}"
            );
        }
    }

    #[test]
    fn hostcall_response_unions_reject_ambiguous_or_extended_payloads() {
        assert_closed_hostcall_response_union::<OpenHandleSuccessResponsePayload>(
            FRAME_TYPE_OPEN_HANDLE,
            r#"{"ok":true,"package_hash":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","artifact":"workers/backend.wasm","sha256":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","content_base64":"AGFzbQ=="}"#,
            r#""content_base64":"AGFzbQ==""#,
        );
        assert_closed_hostcall_response_union::<HandleGrantSuccessResponsePayload>(
            FRAME_TYPE_VALIDATE_HANDLE_GRANT,
            r#"{"ok":true,"handle_grant_id":"grant_1","handle_id":"storage:settings","method":"storage.kv","runtime_generation_id":"g1","resource_scope":{"kind":"user","owner_env_hash":"env_hash","owner_user_hash":"user_hash"}}"#,
            r#""handle_id":"storage:settings""#,
        );
        assert_closed_hostcall_response_union::<StorageFileDeleteSuccessResponsePayload>(
            FRAME_TYPE_STORAGE_FILE,
            r#"{"ok":true,"path":"notes/a.txt"}"#,
            r#""path":"notes/a.txt""#,
        );
        assert_closed_hostcall_response_union::<StorageKVDeleteSuccessResponsePayload>(
            FRAME_TYPE_STORAGE_KV,
            r#"{"ok":true,"key":"settings/theme"}"#,
            r#""key":"settings/theme""#,
        );
        assert_closed_hostcall_response_union::<StorageSQLiteExecSuccessResponsePayload>(
            FRAME_TYPE_STORAGE_SQLITE,
            r#"{"ok":true,"database":"plugin.sqlite","rows_affected":1,"usage":{"plugin_instance_id":"plugini_1","store_id":"db","usage_bytes":1,"quota_bytes":100,"usage_files":1,"quota_files":10}}"#,
            r#""database":"plugin.sqlite""#,
        );
        assert_closed_hostcall_response_union::<NetworkGrantSuccessResponsePayload>(
            FRAME_TYPE_NETWORK_GRANT,
            r#"{"ok":true,"grant_id":"netgrant_00112233445566778899aabbccddeeff","plugin_instance_id":"plugini_1","active_fingerprint":"sha256:active","resource_scope":{"kind":"user","owner_env_hash":"env_hash","owner_user_hash":"user_hash"},"policy_revision":1,"management_revision":2,"revoke_epoch":3,"connector_id":"api","transport":"http","destination":{"transport":"http","scheme":"https","host":"api.example.com","port":443},"runtime_generation_id":"g1","target_classifier_version":"target-classifier-v2","expires_at":"2026-06-30T10:00:30Z"}"#,
            r#""grant_id":"netgrant_00112233445566778899aabbccddeeff""#,
        );
        assert_closed_hostcall_response_union::<NetworkExecuteSuccessResponsePayload>(
            FRAME_TYPE_NETWORK_EXECUTE,
            r#"{"ok":true,"transport":"http","destination":{"transport":"http","scheme":"https","host":"api.example.com","port":443},"status_code":200,"grant_id":"netgrant_00112233445566778899aabbccddeeff","connector_id":"api","runtime_generation_id":"g1"}"#,
            r#""status_code":200"#,
        );
    }

    #[test]
    fn validates_hello_frame() {
        let public_key = base64::engine::general_purpose::STANDARD.encode([7u8; 32]);
        let input = hello_frame(
            Some("nonce_1234567890"),
            &format!(
                r#"[{{"algorithm":"ed25519","key_id":"host_ephemeral_key_1","public_key_base64":"{public_key}"}}]"#
            ),
        );
        let (request_id, generation_id, channel_nonce) =
            validate_hello_frame(&input).expect("valid hello");
        assert_eq!(request_id, "r1");
        assert_eq!(generation_id, "g1");
        assert_eq!(channel_nonce, "nonce_1234567890");
        let parsed = parse_hello_frame(&input).expect("typed hello");
        assert_eq!(parsed.target, RuntimeTarget::LinuxAmd64);
        assert_eq!(parsed.limits, runtime_limits());
    }

    #[test]
    fn runtime_target_enum_covers_only_canonical_linux_targets() {
        for (value, expected) in [
            ("linux/amd64", RuntimeTarget::LinuxAmd64),
            ("linux/arm64", RuntimeTarget::LinuxArm64),
        ] {
            let parsed = RuntimeTarget::parse(value).expect("canonical runtime target");
            assert_eq!(parsed, expected);
            assert_eq!(parsed.as_str(), value);
        }
        for value in [
            "linux/x86_64",
            "darwin/amd64",
            "darwin/arm64",
            "windows/amd64",
            "Linux/amd64",
            "linux-amd64",
        ] {
            assert_eq!(
                RuntimeTarget::parse(value).unwrap_err(),
                IpcError::ProtocolViolation {
                    message: "unsupported runtime target"
                }
            );
        }
    }

    #[test]
    fn rejects_noncanonical_hello_targets() {
        let public_key = base64::engine::general_purpose::STANDARD.encode([7u8; 32]);
        let valid = hello_frame(
            Some("nonce_1234567890"),
            &format!(
                r#"[{{"algorithm":"ed25519","key_id":"host_ephemeral_key_1","public_key_base64":"{public_key}"}}]"#
            ),
        );
        for invalid in [
            valid.replace("linux/amd64", "darwin/amd64"),
            valid.replace("linux/amd64", "linux/x86_64"),
        ] {
            assert_eq!(
                parse_hello_frame(&invalid).unwrap_err(),
                IpcError::InvalidField {
                    field: "hello target"
                }
            );
        }
    }

    #[test]
    fn rejects_v2_and_invalid_runtime_limits() {
        let public_key = base64::engine::general_purpose::STANDARD.encode([7u8; 32]);
        let valid = hello_frame(
            Some("nonce_1234567890"),
            &format!(
                r#"[{{"algorithm":"ed25519","key_id":"host_ephemeral_key_1","public_key_base64":"{public_key}"}}]"#
            ),
        );
        assert!(parse_hello_frame(&valid.replace("rust-ipc-v6", "rust-ipc-v2")).is_err());
        assert!(
            parse_hello_frame(&valid.replacen("\"worker_count\":8", "\"worker_count\":0", 1))
                .is_err()
        );
        assert!(
            parse_hello_frame(&valid.replacen(
                "\"module_cache_source_bytes\":134217728",
                "\"module_cache_source_bytes\":134217729",
                1,
            ))
            .is_err()
        );
        assert!(
            parse_hello_frame(&valid.replacen(
                "\"per_plugin_concurrency\":4",
                "\"per_plugin_concurrency\":9",
                1,
            ))
            .is_err()
        );
    }

    #[test]
    fn runtime_route_capacities_are_closed_derivations_of_hello_limits() {
        let limits = runtime_limits().validate().unwrap();
        assert_eq!(limits.hostcall_active_route_capacity(), limits.worker_count);
        assert_eq!(
            limits.hostcall_canceled_route_capacity().unwrap(),
            limits.worker_count + limits.queue_capacity
        );
        assert_eq!(limits.compile_flight_route_capacity(), limits.worker_count);
    }

    #[test]
    fn decodes_invalid_worker_input_once_into_a_typed_runtime_variant() {
        let input = r#"{"ipc_version":"rust-ipc-v6","frame_type":"invoke_worker","request_id":"invoke-invalid","runtime_generation_id":"g1","payload":{"method":"worker.echo","invocation":{}}}"#;
        let decoded = decode_runtime_input_frame(input).expect("outer IPC frame decodes");
        let RuntimeInputFrame::InvokeWorker(worker) = decoded else {
            panic!("invoke_worker must use the typed worker variant");
        };
        assert_eq!(worker.identity.request_id, "invoke-invalid");
        assert_eq!(worker.identity.runtime_generation_id, "g1");
        assert!(worker.invocation.is_err());
    }

    #[test]
    fn parses_cancel_and_binds_parent_request_id() {
        let cancel = r#"{"ipc_version":"rust-ipc-v6","frame_type":"cancel_invoke","request_id":"cancel-1","runtime_generation_id":"g1","payload":{"invocation_request_id":"invoke-1"}}"#;
        assert_eq!(parse_cancel_invoke(cancel).unwrap(), "invoke-1");
        let ack = cancel_invoke_ack_frame("cancel-1", "g1", "invoke-1", "running")
            .expect("cancel acknowledgement frame");
        assert!(ack.contains(r#""frame_type":"cancel_invoke_ack""#));
        let hostcall = bind_parent_request_id(
            r#"{"ipc_version":"rust-ipc-v6","frame_type":"open_handle","request_id":"invoke-1:artifact","runtime_generation_id":"g1","payload":{}}"#,
            "invoke-1",
        )
        .unwrap();
        assert_eq!(
            parse_frame_identity(&hostcall).unwrap().parent_request_id,
            Some("invoke-1".to_string())
        );
    }

    #[test]
    fn closed_ipc_decoding_rejects_ambiguous_or_extended_frames() {
        let valid = r#"{"ipc_version":"rust-ipc-v6","frame_type":"heartbeat","request_id":"outer","runtime_generation_id":"g1","payload":{"request_id":"nested"}}"#;
        let identity = parse_frame_identity(valid).expect("top-level frame identity");
        assert_eq!(identity.request_id, "outer");

        for invalid in [
            format!("{valid}{{}}"),
            valid.replace(r#""payload""#, r#""unknown":true,"payload""#),
            valid.replace(
                r#""request_id":"outer""#,
                r#""request_id":"outer","request_id":"replayed""#,
            ),
        ] {
            assert!(parse_frame_identity(&invalid).is_err(), "{invalid}");
        }
    }

    #[test]
    fn runtime_hostcall_response_requires_nonempty_parent_request_id() {
        let without_parent = r#"{"ipc_version":"rust-ipc-v6","frame_type":"open_handle","request_id":"r1:artifact","runtime_generation_id":"g1","payload":{"ok":false,"code":"ARTIFACT_HANDLE_FAILED","message":"unavailable","error_origin":"hostcall"}}"#;
        assert!(decode_runtime_input_frame(without_parent).is_err());
        let empty_parent = without_parent.replace(
            r#""request_id":"r1:artifact""#,
            r#""request_id":"r1:artifact","parent_request_id":"""#,
        );
        assert!(decode_runtime_input_frame(&empty_parent).is_err());
    }

    #[test]
    fn closed_worker_decoding_rejects_unknown_duplicate_and_trailing_fields() {
        let valid = closed_worker_frame(
            r#"{"plugin_instance_id":"plugini_1","runtime_shard_id":"runtime_shard_signed","policy_revision":1,"management_revision":2,"revoke_epoch":1}"#,
            r#"{"plugin_id":"com.example.worker","plugin_instance_id":"plugini_1","active_fingerprint":"sha256:active","runtime_instance_id":"runtime_1","runtime_generation_id":"g1","method":"worker.echo"}"#,
        );
        let context = parse_worker_invocation_context(&valid).expect("closed worker invocation");
        assert_eq!(context.runtime_shard_id, "runtime_shard_signed");

        for invalid in [
            format!("{valid}{{}}"),
            valid.replace(
                r#""method":"worker.echo"}}}"#,
                r#""method":"worker.echo","unknown":true}}}"#,
            ),
            valid.replace(
                r#""plugin_instance_id":"plugini_1","active_fingerprint""#,
                r#""plugin_instance_id":"plugini_1","plugin_instance_id":"plugini_2","active_fingerprint""#,
            ),
            valid.replace(
                r#""runtime_instance_id":"runtime_1""#,
                r#""runtime_instance_id":"runtime_1","runtime_shard_id":"runtime_shard_spoofed""#,
            ),
            valid.replace(
                r#""revoke_epoch":1}"#,
                r#""revoke_epoch":1,"unknown":true}"#,
            ),
        ] {
            assert!(parse_worker_invocation_context(&invalid).is_err(), "{invalid}");
        }
    }

    #[test]
    fn worker_frame_initial_decode_types_params_and_broker_access() {
        for invocation in [
            r#"{"method":"worker.echo","params":[]}"#,
            r#"{"method":"worker.echo","broker_access":{"unknown":true}}"#,
            r#"{"method":"worker.echo","broker_access":{"storage":[{"store_id":"notes","scope":"user","operations":["read"],"unknown":true}]}}"#,
        ] {
            let error = match parse_worker_invocation(&closed_worker_frame("{}", invocation)) {
                Ok(_) => panic!("typed invocation fields must fail during initial frame decode"),
                Err(error) => error,
            };
            assert_eq!(
                error,
                IpcError::DecodeFailed {
                    context: "worker frame payload"
                }
            );
        }

        let invocation = r#"{"method":"worker.echo","params":{"title":"Launch notes","body":"<script>&\u2028"},"broker_access":{"storage":[{"store_id":"notes","scope":"user","operations":["read"]}]}}"#;
        serde_json::from_str::<WorkerInvocationPayload>(invocation)
            .expect("direct typed worker invocation payload");
        let parsed = parse_worker_invocation(&closed_worker_frame("{}", invocation))
            .expect("typed worker invocation");
        assert_eq!(
            parsed.params_json.as_deref(),
            Some(r#"{"body":"<script>&\u2028","title":"Launch notes"}"#)
        );
        assert_eq!(
            parsed.broker_access_json.as_deref(),
            Some(r#"{"storage":[{"store_id":"notes","scope":"user","operations":["read"]}]}"#)
        );
    }

    #[test]
    fn rejects_hello_frame_without_channel_nonce() {
        let input = hello_frame(None, "[]");
        assert_eq!(
            validate_hello_frame(&input),
            Err(IpcError::MissingField {
                field: "channel_nonce"
            })
        );
    }

    #[test]
    fn parses_runtime_lease_public_keys_from_hello() {
        let public_key = base64::engine::general_purpose::STANDARD.encode([7u8; 32]);
        let input = hello_frame(
            Some("nonce_1234567890"),
            &format!(
                r#"[{{"algorithm":"ed25519","key_id":"host_ephemeral_key_1","public_key_base64":"{public_key}"}}]"#
            ),
        );
        let keys = parse_runtime_lease_public_keys(&input).expect("keys");
        assert_eq!(
            keys,
            vec![RuntimeLeasePublicKey {
                key_id: "host_ephemeral_key_1".to_string(),
                public_key: [7u8; 32],
            }]
        );
    }

    #[test]
    fn rejects_hello_without_runtime_lease_public_keys() {
        let missing = hello_frame(Some("nonce_1234567890"), "null");
        assert!(parse_runtime_lease_public_keys(&missing).is_err());
        let empty = hello_frame(Some("nonce_1234567890"), "[]");
        assert!(parse_runtime_lease_public_keys(&empty).is_err());
    }

    #[test]
    fn verifies_worker_runtime_lease_signature() {
        let signing_key = runtime_lease_signing_key_for_test(7);
        let frame = signed_runtime_lease_invocation_for_test(&signing_key, None);
        let key = RuntimeLeasePublicKey {
            key_id: "host_ephemeral_key_1".to_string(),
            public_key: signing_key.verifying_key().to_bytes(),
        };
        verify_worker_runtime_lease_signature(&frame, &[key]).expect("signed lease");
    }

    #[test]
    fn rejects_tampered_worker_runtime_lease_signature() {
        let signing_key = runtime_lease_signing_key_for_test(7);
        let frame =
            signed_runtime_lease_invocation_for_test(&signing_key, Some(("revoke_epoch", "14")));
        let key = RuntimeLeasePublicKey {
            key_id: "host_ephemeral_key_1".to_string(),
            public_key: signing_key.verifying_key().to_bytes(),
        };
        let err = verify_worker_runtime_lease_signature(&frame, &[key])
            .expect_err("tampered lease should fail");
        assert_eq!(
            err,
            IpcError::InvalidField {
                field: "runtime lease signature"
            }
        );
    }

    #[test]
    fn rejects_unsigned_worker_runtime_lease_when_keys_are_configured() {
        let signing_key = runtime_lease_signing_key_for_test(7);
        let signed = signed_runtime_lease_invocation_for_test(&signing_key, None);
        let mut frame: serde_json::Value = serde_json::from_str(&signed).expect("signed frame");
        frame["payload"]["lease"]
            .as_object_mut()
            .expect("lease object")
            .remove("signature");
        let frame = serde_json::to_string(&frame).expect("unsigned frame");
        let err = verify_worker_runtime_lease_signature(
            &frame,
            &[RuntimeLeasePublicKey {
                key_id: "host_ephemeral_key_1".to_string(),
                public_key: signing_key.verifying_key().to_bytes(),
            }],
        )
        .expect_err("missing signature should fail");
        assert_eq!(err, IpcError::MissingField { field: "signature" });
    }

    #[test]
    fn runtime_lease_signature_requires_closed_contract_fields() {
        let signing_key = runtime_lease_signing_key_for_test(7);
        let signed = signed_runtime_lease_invocation_for_test(&signing_key, None);
        let frame: serde_json::Value = serde_json::from_str(&signed).expect("signed frame");
        let lease = frame["payload"]["lease"]
            .as_object()
            .expect("lease object")
            .clone();
        let method = frame["payload"]["method"].as_str().expect("method");

        for field in [
            "plugin_id",
            "plugin_version",
            "active_fingerprint",
            "owner_env_hash",
            "target_descriptor_hashes",
            "limits",
            "policy_revision",
            "management_revision",
            "revoke_epoch",
            "runtime_shard_id",
            "runtime_instance_id",
            "ipc_channel_id",
            "connection_nonce",
        ] {
            let mut missing = lease.clone();
            missing.remove(field);
            assert!(
                runtime_lease_signature_payload_json(&worker_lease_from_object(&missing), method)
                    .is_err(),
                "accepted lease without {field}"
            );
        }

        let mut zero_revoke_epoch = lease.clone();
        zero_revoke_epoch.insert("revoke_epoch".to_string(), serde_json::Value::from(0));
        assert_eq!(
            runtime_lease_signature_payload_json(
                &worker_lease_from_object(&zero_revoke_epoch),
                method
            )
            .unwrap_err(),
            IpcError::InvalidField {
                field: "revoke_epoch"
            }
        );

        for field in [
            "timeout_ms",
            "memory_bytes",
            "max_payload_bytes",
            "max_stream_bytes_per_sec",
        ] {
            let mut missing = lease.clone();
            missing["limits"]
                .as_object_mut()
                .expect("limits object")
                .remove(field);
            assert!(
                runtime_lease_signature_payload_json(&worker_lease_from_object(&missing), method)
                    .is_err(),
                "accepted limits without {field}"
            );
        }

        let mut zero_limits = lease;
        for field in [
            "timeout_ms",
            "max_payload_bytes",
            "max_stream_bytes_per_sec",
        ] {
            zero_limits["limits"][field] = serde_json::Value::from(0);
        }
        let canonical =
            runtime_lease_signature_payload_json(&worker_lease_from_object(&zero_limits), method)
                .expect("zero-valued optional quota dimensions remain explicit");
        for field in [
            "\"timeout_ms\":0",
            "\"max_payload_bytes\":0",
            "\"max_stream_bytes_per_sec\":0",
        ] {
            assert!(
                canonical.contains(field),
                "canonical payload omitted {field}"
            );
        }
    }

    #[test]
    fn rejects_worker_runtime_lease_without_runtime_keys() {
        let signing_key = runtime_lease_signing_key_for_test(7);
        let frame = signed_runtime_lease_invocation_for_test(&signing_key, None);
        let err = verify_worker_runtime_lease_signature(&frame, &[])
            .expect_err("missing runtime keyring should fail closed");
        assert_eq!(
            err,
            IpcError::MissingField {
                field: "runtime lease public keys"
            }
        );
    }

    #[test]
    fn validates_worker_runtime_lease_expiry_and_execution_binding() {
        let frame = runtime_lease_invocation_fixture();
        validate_worker_runtime_lease(frame, 1_783_161_901_000)
            .expect("current runtime lease binding");

        let expired = validate_worker_runtime_lease(frame, 1_783_161_930_000)
            .expect_err("expired runtime lease must fail closed");
        assert_eq!(
            expired,
            IpcError::ProtocolViolation {
                message: "runtime execution lease is expired"
            }
        );

        let mut mismatched: Value = serde_json::from_str(frame).expect("invocation fixture");
        mismatched["payload"]["invocation"]["stream_id"] =
            Value::String("stream_other".to_string());
        let mismatch = validate_worker_runtime_lease(
            &serde_json::to_string(&mismatched).expect("mismatched invocation"),
            1_783_161_901_000,
        )
        .expect_err("execution handle mismatch must fail closed");
        assert_eq!(mismatch, IpcError::InvalidField { field: "stream_id" });

        let mut wrong_environment: Value = serde_json::from_str(frame).expect("invocation fixture");
        wrong_environment["payload"]["invocation"]["owner_env_hash"] =
            Value::String("environment_other".to_string());
        let environment_mismatch = validate_worker_runtime_lease(
            &serde_json::to_string(&wrong_environment).expect("mismatched environment"),
            1_783_161_901_000,
        )
        .expect_err("environment mismatch must fail closed");
        assert_eq!(
            environment_mismatch,
            IpcError::InvalidField {
                field: "owner_env_hash"
            }
        );

        let mut tampered_params: Value = serde_json::from_str(frame).expect("invocation fixture");
        tampered_params["payload"]["invocation"]["params"]["message"] =
            Value::String("tampered".to_string());
        let params_mismatch = validate_worker_runtime_lease(
            &serde_json::to_string(&tampered_params).expect("tampered invocation"),
            1_783_161_901_000,
        )
        .expect_err("tampered params must fail closed");
        assert_eq!(
            params_mismatch,
            IpcError::ProtocolViolation {
                message: "worker invocation params_sha256 does not match params"
            }
        );

        let mut unbound_target: Value = serde_json::from_str(frame).expect("invocation fixture");
        unbound_target["payload"]["lease"]["target_descriptor_hashes"] = serde_json::json!([
            "method:sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
            "worker:sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
        ]);
        let target_mismatch = validate_worker_runtime_lease(
            &serde_json::to_string(&unbound_target).expect("unbound invocation"),
            1_783_161_901_000,
        )
        .expect_err("unbound invocation target must fail closed");
        assert_eq!(
            target_mismatch,
            IpcError::ProtocolViolation {
                message: "runtime lease does not bind the worker invocation target"
            }
        );
    }

    #[test]
    fn runtime_lease_signature_payload_matches_go_canonical_order() {
        let lease = serde_json::json!({
            "lease_id": "rel_lease_signature",
            "token_id": "rel_token_signature",
            "lease_nonce": "nonce_1234567890",
            "runtime_generation_id": "rtgen_1",
            "plugin_instance_id": "plugini_1",
            "plugin_id": "com.example.worker",
            "plugin_version": "1.2.3",
            "active_fingerprint": "sha256:active",
            "issued_at_unix_ms": 1783161900000_i64,
            "method": "worker.echo",
            "effect": "read",
            "execution": "sync",
            "audit_correlation_id": "audit_lease_signature",
            "surface_instance_id": "surface_runtime",
            "owner_session_hash": "session_hash",
            "owner_user_hash": "user_hash",
            "owner_env_hash": "env_hash",
            "session_channel_id_hash": "channel_hash",
            "bridge_channel_id": "bridge_runtime",
            "target_descriptor_hashes": [
                "method:sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                "worker:sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
            ],
            "limits": {
                "timeout_ms": 2000,
                "memory_bytes": 65536,
                "max_payload_bytes": 4096,
                "max_stream_bytes_per_sec": 1024
            },
            "policy_revision": 11,
            "management_revision": 12,
            "revoke_epoch": 13,
            "runtime_shard_id": "rtshard_1",
            "runtime_instance_id": "rtinst_1",
            "ipc_channel_id": "ipc_1",
            "connection_nonce": "connection_nonce_1234567890",
            "key_id": "host_ephemeral_key_1",
            "signature": "ed25519:not-part-of-the-payload",
            "expires_at_unix_ms": 1783161930000_i64
        });
        let payload =
            runtime_lease_signature_payload_json(&worker_lease_from_value(&lease), "worker.echo")
                .expect("payload");
        assert_eq!(
            payload,
            r#"{"schema_version":"redevplugin.runtime_execution_lease.v1","token_kind":"runtime_execution_lease","lease_id":"rel_lease_signature","token_id":"rel_token_signature","lease_nonce":"nonce_1234567890","plugin_instance_id":"plugini_1","plugin_id":"com.example.worker","plugin_version":"1.2.3","active_fingerprint":"sha256:active","issued_at_unix_ms":1783161900000,"method":"worker.echo","effect":"read","execution":"sync","audit_correlation_id":"audit_lease_signature","surface_instance_id":"surface_runtime","owner_session_hash":"session_hash","owner_user_hash":"user_hash","owner_env_hash":"env_hash","session_channel_id_hash":"channel_hash","bridge_channel_id":"bridge_runtime","target_descriptor_hashes":["method:sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","worker:sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"],"limits":{"timeout_ms":2000,"memory_bytes":65536,"max_payload_bytes":4096,"max_stream_bytes_per_sec":1024},"policy_revision":11,"management_revision":12,"revoke_epoch":13,"expires_at_unix_ms":1783161930000,"runtime_shard_id":"rtshard_1","runtime_instance_id":"rtinst_1","runtime_generation_id":"rtgen_1","ipc_channel_id":"ipc_1","connection_nonce":"connection_nonce_1234567890","key_id":"host_ephemeral_key_1"}"#
        );
        assert!(!payload.contains("not-part-of-the-payload"));
    }

    #[test]
    fn runtime_lease_signature_shared_fixture_matches_go() {
        let fixture: serde_json::Value =
            serde_json::from_str(include_str!("../testdata/runtime-lease-signature-v1.json"))
                .expect("shared runtime lease fixture");
        let lease = fixture
            .get("lease")
            .and_then(|value| value.as_object())
            .expect("fixture lease");
        let method = fixture
            .get("method")
            .and_then(|value| value.as_str())
            .expect("fixture method");
        let canonical = fixture
            .get("canonical_payload")
            .and_then(|value| value.as_str())
            .expect("fixture canonical payload");
        assert_eq!(
            runtime_lease_signature_payload_json(&worker_lease_from_object(lease), method,)
                .expect("canonical payload"),
            canonical
        );
        let public_key: [u8; 32] = base64::engine::general_purpose::STANDARD
            .decode(
                fixture
                    .get("public_key_base64")
                    .and_then(|value| value.as_str())
                    .expect("fixture public key")
                    .as_bytes(),
            )
            .expect("fixture public key base64")
            .try_into()
            .expect("fixture public key length");
        verify_worker_runtime_lease_signature(
            runtime_lease_invocation_fixture(),
            &[RuntimeLeasePublicKey {
                key_id: "host_ephemeral_fixture_v1".to_string(),
                public_key,
            }],
        )
        .expect("shared runtime lease fixture signature");
    }

    #[test]
    fn renders_hello_ack_frame() {
        let actual_target = RuntimeTarget::LinuxAmd64;
        let frame = hello_ack_frame(HelloAckFrameRequest {
            request_id: "r1",
            runtime_generation_id: "g1",
            channel_nonce: "nonce_1",
            runtime_version: "0.0.0-dev",
            actual_target: &actual_target,
            wasm_abi_version: WASM_ABI_VERSION,
            limits: runtime_limits(),
            process_containment: &process_containment(),
        })
        .expect("hello acknowledgement frame");
        assert!(frame.contains(r#""frame_type":"hello_ack""#));
        assert!(frame.contains(r#""request_id":"r1""#));
        assert!(frame.contains(r#""runtime_generation_id":"g1""#));
        assert!(frame.contains(r#""actual_target":"linux/amd64""#));
        assert!(frame.contains(r#""rust_ipc_version":"rust-ipc-v6""#));
        assert!(frame.contains(&format!(r#""contract_set_sha256":"{CONTRACT_SET_SHA256}""#)));
        assert!(frame.contains(r#""channel_nonce":"nonce_1""#));
        assert!(frame.contains(r#""worker_count":8"#));
        assert!(frame.contains(r#""module_cache_source_bytes":134217728"#));
    }

    #[test]
    fn hello_ack_frame_rejects_invalid_runtime_limits() {
        let actual_target = RuntimeTarget::LinuxAmd64;
        assert!(matches!(
            hello_ack_frame(HelloAckFrameRequest {
                request_id: "r1",
                runtime_generation_id: "g1",
                channel_nonce: "nonce_1",
                runtime_version: "0.0.0-dev",
                actual_target: &actual_target,
                wasm_abi_version: WASM_ABI_VERSION,
                limits: invalid_runtime_limits(),
                process_containment: &process_containment(),
            }),
            Err(IpcError::ProtocolViolation { .. })
        ));
    }

    #[test]
    fn renders_error_response_frame() {
        let frame = error_response_frame(
            FRAME_TYPE_INVOKE_WORKER_RESULT,
            "r1",
            "g1",
            ResponseError::runtime(ERR_WASM_WORKER_FAILED, "runtime worker execution failed")
                .expect("runtime response error"),
        )
        .expect("runtime error response frame");
        assert!(frame.contains(r#""frame_type":"invoke_worker_result""#));
        assert!(frame.contains(r#""ok":false"#));
        assert!(frame.contains(r#""code":"WASM_WORKER_FAILED""#));
        assert!(frame.contains(r#""error_origin":"runtime""#));

        let plugin_frame = error_response_frame(
            FRAME_TYPE_INVOKE_WORKER_RESULT,
            "r2",
            "g1",
            ResponseError::plugin("NOTE_NOT_FOUND", "note was not found")
                .expect("plugin response error"),
        )
        .expect("plugin error response frame");
        assert!(plugin_frame.contains(r#""error_origin":"plugin""#));
    }

    #[test]
    fn error_response_rejects_empty_code() {
        assert_eq!(
            ResponseError::runtime(" ", "failed"),
            Err(IpcError::EmptyResponseErrorCode),
        );
    }

    #[test]
    fn error_response_rejects_empty_message() {
        assert_eq!(
            ResponseError::runtime("FAILED", " "),
            Err(IpcError::EmptyResponseErrorMessage),
        );
    }

    #[test]
    fn success_response_rejects_invalid_result_json_without_panicking() {
        assert_eq!(
            success_response_frame(FRAME_TYPE_INVOKE_WORKER_RESULT, "r1", "g1", "{"),
            Err(IpcError::InvalidResponseResultJson),
        );
    }

    #[test]
    fn ipc_errors_are_cloneable_comparable_and_have_stable_redacted_display() {
        let error = IpcError::RemoteFailure {
            code: "NETWORK_TARGET_DENIED".to_string(),
        };
        assert_eq!(error.clone(), error);
        assert_eq!(
            error.to_string(),
            "hostcall response failed with code NETWORK_TARGET_DENIED"
        );

        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"network_execute","request_id":"r1:network_execute","runtime_generation_id":"g1","payload":{"ok":false,"code":"NETWORK_TARGET_DENIED","message":"bearer secret-token https://api.example.com/path?token=secret /Users/private/key","error_origin":"hostcall"}}"#;
        let error =
            validate_network_execute_response(failed, "r1:network_execute", "g1", "api", "http")
                .expect_err("hostcall failure");
        let display = error.to_string();
        assert_eq!(
            display,
            "hostcall response failed with code NETWORK_TARGET_DENIED"
        );
        for sensitive in [
            "secret-token",
            "api.example.com",
            "token=secret",
            "/Users/private",
        ] {
            assert!(!display.contains(sensitive), "display leaked {sensitive}");
        }
    }

    #[test]
    fn ipc_golden_fixtures_match_rust_frame_contract() {
        let fixtures = [
            "valid_hello_ack.json",
            "valid_invoke_worker_result.json",
            "valid_validate_handle_grant.json",
            "missing_required.json",
            "replay_frame.json",
            "runtime_generation_mismatch.json",
            "unknown_enum.json",
        ];
        for fixture_name in fixtures {
            let fixture = load_ipc_fixture(fixture_name);
            assert_eq!(
                fixture["want_error"].as_bool(),
                Some(
                    fixture_name != "valid_hello_ack.json"
                        && fixture_name != "valid_invoke_worker_result.json"
                        && fixture_name != "valid_validate_handle_grant.json"
                ),
                "fixture {fixture_name} want_error mismatch"
            );
            let frame = fixture.get("frame").expect("fixture frame").clone();
            let frame_json = serde_json::to_string(&frame).expect("compact frame");
            match fixture_name {
                "valid_hello_ack.json" => {
                    let actual_target = RuntimeTarget::parse(
                        frame["payload"]["actual_target"]
                            .as_str()
                            .expect("actual_target"),
                    )
                    .expect("actual_target");
                    let encoded = hello_ack_frame(HelloAckFrameRequest {
                        request_id: fixture["request_id"].as_str().expect("request_id"),
                        runtime_generation_id: fixture["runtime_generation_id"]
                            .as_str()
                            .expect("runtime_generation_id"),
                        channel_nonce: fixture["channel_nonce"].as_str().expect("channel_nonce"),
                        runtime_version: frame["payload"]["runtime_version"]
                            .as_str()
                            .expect("runtime_version"),
                        actual_target: &actual_target,
                        wasm_abi_version: WASM_ABI_VERSION,
                        limits: runtime_limits(),
                        process_containment: &process_containment(),
                    })
                    .expect("hello acknowledgement frame");
                    assert_json_eq(&frame_json, &encoded, fixture_name);
                }
                "valid_invoke_worker_result.json" => {
                    let result =
                        serde_json::to_string(&frame["payload"]["result"]).expect("compact result");
                    let encoded = success_response_frame(
                        FRAME_TYPE_INVOKE_WORKER_RESULT,
                        fixture["request_id"].as_str().expect("request_id"),
                        fixture["runtime_generation_id"]
                            .as_str()
                            .expect("runtime_generation_id"),
                        &result,
                    )
                    .expect("success response frame");
                    assert_json_eq(&frame_json, &encoded, fixture_name);
                }
                "valid_validate_handle_grant.json" => {
                    let encoded = validate_handle_grant_frame(
                        fixture["request_id"].as_str().expect("request_id"),
                        fixture["runtime_generation_id"]
                            .as_str()
                            .expect("runtime_generation_id"),
                        &handle_grant_validation_request(),
                    )
                    .expect("validate handle grant frame");
                    let encoded = bind_parent_request_id(
                        &encoded,
                        fixture["parent_request_id"]
                            .as_str()
                            .expect("parent_request_id"),
                    )
                    .expect("bind validate handle grant parent");
                    assert_json_eq(&frame_json, &encoded, fixture_name);
                }
                "missing_required.json" => {
                    assert_eq!(
                        parse_frame_identity(&frame_json),
                        Err(IpcError::MissingField {
                            field: "request_id"
                        }),
                        "fixture {fixture_name} should reject missing request_id"
                    );
                }
                "replay_frame.json" => {
                    let identity = parse_frame_identity(&frame_json).expect("parse replay fixture");
                    assert_ne!(
                        identity.request_id,
                        fixture["request_id"].as_str().expect("expected request_id"),
                        "fixture {fixture_name} should replay a different request_id"
                    );
                }
                "runtime_generation_mismatch.json" => {
                    let identity = parse_frame_identity(&frame_json)
                        .expect("parse runtime generation mismatch fixture");
                    assert_ne!(
                        identity.runtime_generation_id,
                        fixture["runtime_generation_id"]
                            .as_str()
                            .expect("expected runtime_generation_id"),
                        "fixture {fixture_name} should carry mismatched runtime generation"
                    );
                }
                "unknown_enum.json" => {
                    let identity =
                        parse_frame_identity(&frame_json).expect("parse unknown enum fixture");
                    assert_ne!(
                        identity.frame_type, FRAME_TYPE_INVOKE_WORKER_RESULT,
                        "fixture {fixture_name} should use an unknown frame type"
                    );
                }
                _ => panic!("unhandled fixture {fixture_name}"),
            }
        }
    }

    fn load_ipc_fixture(name: &str) -> Value {
        let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        path.push("testdata");
        path.push("ipc");
        path.push(name);
        let raw = fs::read_to_string(&path).unwrap_or_else(|err| {
            panic!("read fixture {}: {err}", path.display());
        });
        serde_json::from_str(&raw).unwrap_or_else(|err| {
            panic!("decode fixture {}: {err}", path.display());
        })
    }

    fn assert_json_eq(actual: &str, expected: &str, label: &str) {
        let actual: Value = serde_json::from_str(actual).expect("actual json");
        let expected: Value = serde_json::from_str(expected).expect("expected json");
        assert_eq!(actual, expected, "{label} json mismatch");
    }

    #[test]
    fn renders_revoke_epoch_ack_result_json() {
        let result =
            revoke_epoch_ack_result_json(&environment_resource_scope(), "plugini_1", 7, 2, 3, 4)
                .expect("valid revoke result");
        assert!(
            result
                .contains(r#""resource_scope":{"kind":"environment","owner_env_hash":"env_hash"}"#)
        );
        assert!(result.contains(r#""plugin_instance_id":"plugini_1""#));
        assert!(result.contains(r#""revoke_epoch":7"#));
        assert!(result.contains(r#""closed_socket_count":2"#));
        assert!(result.contains(r#""closed_stream_count":3"#));
        assert!(result.contains(r#""closed_storage_handle_count":4"#));
    }

    #[test]
    fn renders_heartbeat_ack_result_json() {
        let result = heartbeat_ack_result_json(
            "runtime_gen_1",
            101,
            5000,
            100,
            RuntimeHeartbeatStatus {
                active_invocations: 2,
                queued_invocations: 3,
                limits: runtime_limits(),
                module_cache: ModuleCacheMetrics {
                    hits: 4,
                    misses: 5,
                    compiles: 1,
                    entries: 1,
                    source_bytes: 1024,
                },
            },
        )
        .expect("heartbeat acknowledgement result");
        assert!(result.contains(r#""runtime_generation_id":"runtime_gen_1""#));
        assert!(result.contains(r#""runtime_unix_nano":101"#));
        assert!(result.contains(r#""max_staleness_ms":5000"#));
        assert!(result.contains(r#""host_sent_unix_nano":100"#));
    }

    #[test]
    fn heartbeat_ack_result_rejects_invalid_runtime_limits() {
        assert!(matches!(
            heartbeat_ack_result_json(
                "runtime_gen_1",
                101,
                5000,
                100,
                RuntimeHeartbeatStatus {
                    active_invocations: 0,
                    queued_invocations: 0,
                    limits: invalid_runtime_limits(),
                    module_cache: ModuleCacheMetrics {
                        hits: 0,
                        misses: 0,
                        compiles: 0,
                        entries: 0,
                        source_bytes: 0,
                    },
                },
            ),
            Err(IpcError::ProtocolViolation { .. })
        ));
    }

    #[test]
    fn renders_open_handle_frame() {
        let identity = WorkerInvocationIdentity {
            package_hash: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                .to_string(),
            artifact: "workers/backend.wasm".to_string(),
            artifact_sha256:
                "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
                    .to_string(),
            worker_id: "backend".to_string(),
            method: "worker.echo".to_string(),
        };
        let frame = open_handle_frame("r1", "g1", &identity);
        assert!(frame.contains(r#""frame_type":"open_handle""#));
        assert!(frame.contains(r#""artifact":"workers/backend.wasm""#));
    }

    #[test]
    fn renders_compile_flight_lifecycle_frames() {
        let identity = WorkerInvocationIdentity {
            package_hash: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                .to_string(),
            artifact: "workers/backend.wasm".to_string(),
            artifact_sha256:
                "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
                    .to_string(),
            worker_id: "backend".to_string(),
            method: "worker.echo".to_string(),
        };
        for (frame_type, actual) in [
            (
                FRAME_TYPE_COMPILE_FLIGHT_REGISTER,
                compile_flight_register_frame("invoke-1", "generation-1", &identity),
            ),
            (
                FRAME_TYPE_COMPILE_FLIGHT_COMPLETE,
                compile_flight_complete_frame("invoke-1", "generation-1", &identity),
            ),
        ] {
            let suffix = if frame_type == FRAME_TYPE_COMPILE_FLIGHT_REGISTER {
                "register"
            } else {
                "complete"
            };
            let expected = format!(
                r#"{{"ipc_version":"rust-ipc-v6","frame_type":"{frame_type}","request_id":"invoke-1:artifact:{suffix}","parent_request_id":"invoke-1","runtime_generation_id":"generation-1","payload":{{"artifact_request_id":"invoke-1:artifact","package_hash":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","artifact":"workers/backend.wasm","artifact_sha256":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","wasm_abi_version":"redevplugin-wasm-worker-v2"}}}}"#
            );
            assert_json_eq(&actual, &expected, frame_type);
        }
    }

    #[test]
    fn validates_open_handle_response() {
        let identity = WorkerInvocationIdentity {
            package_hash: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                .to_string(),
            artifact: "workers/backend.wasm".to_string(),
            artifact_sha256:
                "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
                    .to_string(),
            worker_id: "backend".to_string(),
            method: "worker.echo".to_string(),
        };
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"open_handle","request_id":"r1:artifact","parent_request_id":"r1","runtime_generation_id":"g1","payload":{"ok":true,"package_hash":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","artifact":"workers/backend.wasm","sha256":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","content_base64":"AAE="}}"#;
        validate_open_handle_response(frame, "r1:artifact", "r1", "g1", &identity)
            .expect("valid open_handle");
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"open_handle","request_id":"r1:artifact","parent_request_id":"r1","runtime_generation_id":"g1","payload":{"ok":false,"code":"ARTIFACT_HANDLE_FAILED","message":"unavailable","error_origin":"hostcall"}}"#;
        let err = validate_open_handle_response(failed, "r1:artifact", "r1", "g1", &identity)
            .expect_err("failed open_handle response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "ARTIFACT_HANDLE_FAILED".to_string()
            }
        );
    }

    #[test]
    fn renders_validate_handle_grant_frame() {
        let request = handle_grant_validation_request();
        let frame = validate_handle_grant_frame("r1:handle", "g1", &request)
            .expect("valid handle grant frame");
        assert!(frame.contains(r#""frame_type":"validate_handle_grant""#));
        assert!(frame.contains(r#""handle_id":"storage:db""#));
        assert!(frame.contains(r#""runtime_instance_id":"runtime_1""#));
        assert!(frame.contains(r#""runtime_shard_id":"runtime_shard_1""#));
        assert!(frame.contains(r#""owner_session_hash":"session_hash""#));
        assert!(frame.contains(r#""owner_user_hash":"user_hash""#));
        assert!(frame.contains(r#""owner_env_hash":"env_hash""#));
        assert!(frame.contains(r#""session_channel_id_hash":"channel_hash""#));
        assert!(frame.contains(r#""policy_revision":1"#));

        let mut missing_session = request.clone();
        missing_session.owner_session_hash.clear();
        assert_eq!(
            validate_handle_grant_frame("r1:handle", "g1", &missing_session).unwrap_err(),
            IpcError::InvalidField {
                field: "handle grant session audience"
            }
        );

        let mut mismatched_scope = request.clone();
        mismatched_scope.owner_env_hash = "env_other".to_string();
        assert_eq!(
            validate_handle_grant_frame("r1:handle", "g1", &mismatched_scope).unwrap_err(),
            IpcError::InvalidField {
                field: "handle grant resource scope"
            }
        );

        let mut missing_token = request.clone();
        missing_token.handle_grant_token.clear();
        assert_eq!(
            validate_handle_grant_frame("r1:handle", "g1", &missing_token).unwrap_err(),
            IpcError::InvalidField {
                field: "handle_grant_token"
            }
        );
        assert_eq!(
            validate_handle_grant_frame("", "g1", &request).unwrap_err(),
            IpcError::InvalidField {
                field: "request_id"
            }
        );
        assert_eq!(
            validate_handle_grant_frame("r1:handle", "g2", &request).unwrap_err(),
            IpcError::ProtocolViolation {
                message: "validate_handle_grant runtime_generation_id mismatch"
            }
        );
        let mut unsafe_revision = request;
        unsafe_revision.policy_revision = MAX_JSON_SAFE_INTEGER + 1;
        assert_eq!(
            validate_handle_grant_frame("r1:handle", "g1", &unsafe_revision).unwrap_err(),
            IpcError::InvalidField {
                field: "policy_revision"
            }
        );
    }

    #[test]
    fn zero_revoke_epoch_returns_typed_errors() {
        let revoke = r#"{"ipc_version":"rust-ipc-v6","frame_type":"revoke_epoch","request_id":"r1","runtime_generation_id":"g1","payload":{"resource_scope":{"kind":"environment","owner_env_hash":"env_hash"},"plugin_instance_id":"plugini_1","revoke_epoch":0}}"#;
        assert_eq!(
            parse_revoke_epoch_request(revoke).unwrap_err(),
            IpcError::InvalidField {
                field: "revoke_epoch"
            }
        );
        let worker = closed_worker_frame(
            r#"{"runtime_shard_id":"runtime_shard_signed","policy_revision":1,"management_revision":2,"revoke_epoch":0}"#,
            r#"{"plugin_id":"com.example.worker","plugin_instance_id":"plugini_1","active_fingerprint":"sha256:active","runtime_instance_id":"runtime_1","runtime_generation_id":"g1","method":"worker.echo"}"#,
        );
        assert_eq!(
            parse_worker_invocation_context(&worker).unwrap_err(),
            IpcError::InvalidField {
                field: "revoke_epoch"
            }
        );
        assert_eq!(
            revoke_epoch_ack_result_json(&environment_resource_scope(), "plugini_1", 0, 0, 0, 0)
                .unwrap_err(),
            IpcError::InvalidField {
                field: "revoke_epoch"
            }
        );

        let mut handle_grant = handle_grant_validation_request();
        handle_grant.revoke_epoch = 0;
        assert_eq!(
            validate_handle_grant_frame("r1:handle", "g1", &handle_grant).unwrap_err(),
            IpcError::InvalidField {
                field: "revoke_epoch"
            }
        );

        let mut network_grant = NetworkGrantRequest {
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            resource_scope: user_resource_scope(),
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 0,
            connector_id: "api".to_string(),
            transport: "http".to_string(),
            destination: "https://api.example.com".to_string(),
            ttl_ms: 30000,
        };
        assert_eq!(
            network_grant_frame("r1:network_grant", "g1", &network_grant).unwrap_err(),
            IpcError::InvalidField {
                field: "revoke_epoch"
            }
        );
        network_grant.revoke_epoch = 1;
        network_grant.resource_scope.owner_env_hash = " env_hash".to_string();
        assert_eq!(
            network_grant_frame("r1:network_grant", "g1", &network_grant).unwrap_err(),
            IpcError::InvalidField {
                field: "network resource scope"
            }
        );
    }

    #[test]
    fn validates_handle_grant_response() {
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"validate_handle_grant","request_id":"r1:handle","runtime_generation_id":"g1","payload":{"ok":true,"handle_grant_id":"h1","handle_id":"storage:db","method":"storage.sqlite","runtime_generation_id":"g1","resource_scope":{"kind":"user","owner_env_hash":"env_hash","owner_user_hash":"user_hash"},"max_total_bytes":4096}}"#;
        validate_handle_grant_response(
            frame,
            "r1:handle",
            "g1",
            "storage:db",
            "storage.sqlite",
            &user_resource_scope(),
        )
        .expect("valid handle grant");
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"validate_handle_grant","request_id":"r1:handle","runtime_generation_id":"g1","payload":{"ok":false,"code":"HANDLE_GRANT_VALIDATION_FAILED","message":"denied","error_origin":"hostcall"}}"#;
        let err = validate_handle_grant_response(
            failed,
            "r1:handle",
            "g1",
            "storage:db",
            "storage.sqlite",
            &user_resource_scope(),
        )
        .expect_err("failed handle grant response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "HANDLE_GRANT_VALIDATION_FAILED".to_string()
            }
        );
    }

    #[test]
    fn renders_storage_file_frame() {
        let mut req = StorageFileRequest {
            handle_grant_token: "handle_grant.secret".to_string(),
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            handle_id: "storage:workspace".to_string(),
            method: "storage.files".to_string(),
            resource_scope: user_resource_scope(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 3,
            operation: "read".to_string(),
            store_id: "workspace".to_string(),
            path: "notes/today.txt".to_string(),
            data_base64: "".to_string(),
            max_bytes: 1024,
            max_entries: 10,
            recursive: false,
        };
        let frame =
            storage_file_frame("r1:storage_file", "g1", &req).expect("valid storage file frame");
        assert!(frame.contains(r#""frame_type":"storage_file""#));
        assert!(frame.contains(r#""handle_id":"storage:workspace""#));
        assert!(frame.contains(r#""method":"storage.files""#));
        assert!(frame.contains(r#""operation":"read""#));
        req.policy_revision = MAX_JSON_SAFE_INTEGER + 1;
        assert_eq!(
            storage_file_frame("r1:storage_file", "g1", &req).unwrap_err(),
            IpcError::InvalidField {
                field: "policy_revision"
            }
        );
    }

    #[test]
    fn validates_storage_file_response() {
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_file","request_id":"r1:storage_file","runtime_generation_id":"g1","payload":{"ok":true,"path":"notes/today.txt","data_base64":"aGVsbG8=","size_bytes":5,"usage":{"plugin_instance_id":"plugini_1","store_id":"workspace","usage_bytes":5,"quota_bytes":100,"usage_files":1,"quota_files":10}}}"#;
        validate_storage_file_response(frame, "r1:storage_file", "g1", "read")
            .expect("valid storage file response");
        let payload = storage_file_payload_json(frame, "read").expect("storage file payload");
        assert!(payload.contains(r#""path":"notes/today.txt""#));
        for wrong_operation in ["write", "delete", "list"] {
            assert!(
                validate_storage_file_response(frame, "r1:storage_file", "g1", wrong_operation)
                    .is_err(),
                "read response accepted as {wrong_operation}"
            );
        }
        let mixed = frame.replace(r#""usage":{"#, r#""entries":[],"usage":{"#);
        assert!(validate_storage_file_response(&mixed, "r1:storage_file", "g1", "read").is_err());
        let missing = without_payload_field(frame, "data_base64");
        assert!(validate_storage_file_response(&missing, "r1:storage_file", "g1", "read").is_err());
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_file","request_id":"r1:storage_file","runtime_generation_id":"g1","payload":{"ok":false,"code":"STORAGE_FILE_NOT_FOUND","message":"missing","error_origin":"hostcall"}}"#;
        let err = validate_storage_file_response(failed, "r1:storage_file", "g1", "read")
            .expect_err("failed storage file response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "STORAGE_FILE_NOT_FOUND".to_string()
            }
        );
        let missing_origin = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_file","request_id":"r1:storage_file","runtime_generation_id":"g1","payload":{"ok":false,"code":"STORAGE_FILE_NOT_FOUND","message":"missing"}}"#;
        let err = validate_storage_file_response(missing_origin, "r1:storage_file", "g1", "read")
            .expect_err("hostcall origin is required");
        assert_eq!(
            err,
            IpcError::DecodeFailed {
                context: "hostcall failure response payload"
            }
        );
        let spoofed_origin = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_file","request_id":"r1:storage_file","runtime_generation_id":"g1","payload":{"ok":false,"code":"STORAGE_FILE_NOT_FOUND","message":"missing","error_origin":"plugin"}}"#;
        let err = validate_storage_file_response(spoofed_origin, "r1:storage_file", "g1", "read")
            .expect_err("hostcall origin cannot be spoofed");
        assert_eq!(
            err,
            IpcError::ProtocolViolation {
                message: "hostcall response error_origin must be hostcall"
            }
        );
    }

    #[test]
    fn validates_all_storage_file_success_operations() {
        let usage = r#"{"plugin_instance_id":"plugini_1","store_id":"workspace","usage_bytes":5,"quota_bytes":100,"usage_files":1,"quota_files":10}"#;
        let cases = [
            (
                "read",
                format!(
                    r#"{{"ok":true,"path":"a.txt","data_base64":"YQ==","size_bytes":1,"usage":{usage}}}"#
                ),
            ),
            (
                "write",
                format!(r#"{{"ok":true,"path":"a.txt","size_bytes":1,"usage":{usage}}}"#),
            ),
            ("delete", r#"{"ok":true,"path":"a.txt"}"#.to_string()),
            (
                "list",
                format!(r#"{{"ok":true,"path":"","entries":[],"usage":{usage}}}"#),
            ),
        ];
        for (operation, payload) in &cases {
            let frame = hostcall_response_frame(FRAME_TYPE_STORAGE_FILE, payload);
            validate_storage_file_response(&frame, "r1", "g1", operation)
                .unwrap_or_else(|err| panic!("{operation} response: {err}"));
            for (other, _) in &cases {
                if other != operation {
                    assert!(validate_storage_file_response(&frame, "r1", "g1", other).is_err());
                }
            }
        }
    }

    #[test]
    fn renders_storage_kv_frame() {
        let mut req = StorageKVRequest {
            handle_grant_token: "handle_grant.secret".to_string(),
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            handle_id: "storage:settings".to_string(),
            method: "storage.kv".to_string(),
            resource_scope: user_resource_scope(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 3,
            operation: "put".to_string(),
            store_id: "settings".to_string(),
            key: "demo/last_broker_run".to_string(),
            value_base64: "aGVsbG8=".to_string(),
            prefix: String::new(),
            max_bytes: 0,
            max_entries: 10,
        };
        let frame = storage_kv_frame("r1:storage_kv", "g1", &req).expect("valid storage kv frame");
        assert!(frame.contains(r#""frame_type":"storage_kv""#));
        assert!(frame.contains(r#""handle_id":"storage:settings""#));
        assert!(frame.contains(r#""method":"storage.kv""#));
        assert!(frame.contains(r#""operation":"put""#));
        req.management_revision = MAX_JSON_SAFE_INTEGER + 1;
        assert_eq!(
            storage_kv_frame("r1:storage_kv", "g1", &req).unwrap_err(),
            IpcError::InvalidField {
                field: "management_revision"
            }
        );
        assert!(frame.contains(r#""key":"demo/last_broker_run""#));
    }

    #[test]
    fn validates_storage_kv_response() {
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_kv","request_id":"r1:storage_kv","runtime_generation_id":"g1","payload":{"ok":true,"key":"demo/last_broker_run","value_base64":"aGVsbG8=","size_bytes":5,"usage":{"plugin_instance_id":"plugini_1","store_id":"settings","usage_bytes":5,"quota_bytes":100,"usage_files":1,"quota_files":10}}}"#;
        validate_storage_kv_response(frame, "r1:storage_kv", "g1", "get")
            .expect("valid storage kv response");
        let payload = storage_kv_payload_json(frame, "get").expect("storage kv payload");
        assert!(payload.contains(r#""key":"demo/last_broker_run""#));
        for wrong_operation in ["put", "delete", "list"] {
            assert!(
                validate_storage_kv_response(frame, "r1:storage_kv", "g1", wrong_operation)
                    .is_err()
            );
        }
        let mixed = frame.replace(r#""usage":{"#, r#""entries":[],"usage":{"#);
        assert!(validate_storage_kv_response(&mixed, "r1:storage_kv", "g1", "get").is_err());
        let missing = without_payload_field(frame, "value_base64");
        assert!(validate_storage_kv_response(&missing, "r1:storage_kv", "g1", "get").is_err());
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_kv","request_id":"r1:storage_kv","runtime_generation_id":"g1","payload":{"ok":false,"code":"STORAGE_KV_NOT_FOUND","message":"missing","error_origin":"hostcall"}}"#;
        let err = validate_storage_kv_response(failed, "r1:storage_kv", "g1", "get")
            .expect_err("failed storage kv response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "STORAGE_KV_NOT_FOUND".to_string()
            }
        );
    }

    #[test]
    fn validates_all_storage_kv_success_operations() {
        let usage = r#"{"plugin_instance_id":"plugini_1","store_id":"settings","usage_bytes":5,"quota_bytes":100,"usage_files":1,"quota_files":10}"#;
        let cases = [
            (
                "get",
                format!(
                    r#"{{"ok":true,"key":"theme","value_base64":"ZGFyaw==","size_bytes":4,"usage":{usage}}}"#
                ),
            ),
            (
                "put",
                format!(r#"{{"ok":true,"key":"theme","size_bytes":4,"usage":{usage}}}"#),
            ),
            ("delete", r#"{"ok":true,"key":"theme"}"#.to_string()),
            (
                "list",
                format!(r#"{{"ok":true,"prefix":"","entries":[],"usage":{usage}}}"#),
            ),
        ];
        for (operation, payload) in &cases {
            let frame = hostcall_response_frame(FRAME_TYPE_STORAGE_KV, payload);
            validate_storage_kv_response(&frame, "r1", "g1", operation)
                .unwrap_or_else(|err| panic!("{operation} response: {err}"));
            for (other, _) in &cases {
                if other != operation {
                    assert!(validate_storage_kv_response(&frame, "r1", "g1", other).is_err());
                }
            }
        }
    }

    #[test]
    fn renders_worker_success_with_storage_result() {
        let identity = WorkerInvocationIdentity {
            package_hash: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                .to_string(),
            artifact: "workers/backend.wasm".to_string(),
            artifact_sha256:
                "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
                    .to_string(),
            worker_id: "backend".to_string(),
            method: "worker.echo".to_string(),
        };
        let result = worker_success_result_json(
            &identity,
            42,
            Some(r#"{"ok":true,"path":"notes/from-wasm.txt","size_bytes":5}"#),
            Some(r#"{"ok":true,"key":"demo/last_broker_run","size_bytes":12}"#),
            Some(r#"{"ok":true,"database":"plugin.sqlite","rows_affected":1}"#),
            Some(r#"{"ok":true,"transport":"http","status_code":201,"stream_id":"stream_http_1"}"#),
        );
        assert!(result.contains(r#""storage_file":{"ok":true"#));
        assert!(result.contains(r#""storage_kv":{"ok":true"#));
        assert!(result.contains(r#""storage_sqlite":{"ok":true"#));
        assert!(result.contains(r#""network_execute":{"ok":true"#));
        assert!(result.contains(r#""stream_id":"stream_http_1""#));
        assert!(result.contains(r#""wasm_byte_len":42"#));
    }

    #[test]
    fn renders_storage_sqlite_frame() {
        let mut req = StorageSQLiteRequest {
            handle_grant_token: "handle_grant.secret".to_string(),
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            handle_id: "storage:db".to_string(),
            method: "storage.sqlite".to_string(),
            resource_scope: user_resource_scope(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 3,
            operation: "query".to_string(),
            store_id: "db".to_string(),
            database: "plugin.sqlite".to_string(),
            sql: "SELECT title FROM events WHERE score = ?".to_string(),
            args_json: r#"[{"int":7}]"#.to_string(),
            max_rows: 10,
            max_response_bytes: 4096,
            timeout_ms: 1000,
        };
        let frame = storage_sqlite_frame("r1:storage_sqlite", "g1", &req)
            .expect("valid storage sqlite frame");
        assert!(frame.contains(r#""frame_type":"storage_sqlite""#));
        assert!(frame.contains(r#""handle_id":"storage:db""#));
        assert!(frame.contains(r#""method":"storage.sqlite""#));
        assert!(frame.contains(r#""operation":"query""#));
        assert!(frame.contains(r#""args":[{"int":7}]"#));
        req.policy_revision = MAX_JSON_SAFE_INTEGER + 1;
        assert_eq!(
            storage_sqlite_frame("r1:storage_sqlite", "g1", &req).unwrap_err(),
            IpcError::InvalidField {
                field: "policy_revision"
            }
        );
    }

    #[test]
    fn validates_storage_sqlite_response() {
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_sqlite","request_id":"r1:storage_sqlite","runtime_generation_id":"g1","payload":{"ok":true,"database":"plugin.sqlite","columns":["title"],"rows":[[{"text":"stored from wasm"}]],"usage":{"plugin_instance_id":"plugini_1","store_id":"db","usage_bytes":5,"quota_bytes":100,"usage_files":1,"quota_files":10}}}"#;
        validate_storage_sqlite_response(frame, "r1:storage_sqlite", "g1", "query")
            .expect("valid storage sqlite response");
        let payload = storage_sqlite_payload_json(frame, "query").expect("storage sqlite payload");
        assert!(payload.contains(r#""database":"plugin.sqlite""#));
        assert!(
            validate_storage_sqlite_response(frame, "r1:storage_sqlite", "g1", "exec").is_err()
        );
        let mixed = frame.replace(
            r#""columns":["title"]"#,
            r#""rows_affected":1,"columns":["title"]"#,
        );
        assert!(
            validate_storage_sqlite_response(&mixed, "r1:storage_sqlite", "g1", "query").is_err()
        );
        let missing = without_payload_field(frame, "rows");
        assert!(
            validate_storage_sqlite_response(&missing, "r1:storage_sqlite", "g1", "query").is_err()
        );
        let empty_blob = frame.replace(r#"{"text":"stored from wasm"}"#, r#"{"blob_base64":""}"#);
        validate_storage_sqlite_response(&empty_blob, "r1:storage_sqlite", "g1", "query")
            .expect("empty SQLite blob is a valid typed value");
        for invalid_value in [
            r#"{}"#,
            r#"{"null":false}"#,
            r#"{"int":1,"text":"ambiguous"}"#,
        ] {
            let invalid = frame.replace(r#"{"text":"stored from wasm"}"#, invalid_value);
            assert!(
                validate_storage_sqlite_response(&invalid, "r1:storage_sqlite", "g1", "query")
                    .is_err(),
                "accepted invalid SQLite value {invalid_value}"
            );
        }
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"storage_sqlite","request_id":"r1:storage_sqlite","runtime_generation_id":"g1","payload":{"ok":false,"code":"STORAGE_SQLITE_RESULT_TOO_LARGE","message":"too large","error_origin":"hostcall"}}"#;
        let err = validate_storage_sqlite_response(failed, "r1:storage_sqlite", "g1", "query")
            .expect_err("failed storage sqlite response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "STORAGE_SQLITE_RESULT_TOO_LARGE".to_string()
            }
        );
    }

    #[test]
    fn validates_all_storage_sqlite_success_operations() {
        let usage = r#"{"plugin_instance_id":"plugini_1","store_id":"db","usage_bytes":5,"quota_bytes":100,"usage_files":1,"quota_files":10}"#;
        let cases = [
            (
                "exec",
                format!(
                    r#"{{"ok":true,"database":"plugin.sqlite","rows_affected":0,"usage":{usage}}}"#
                ),
            ),
            (
                "query",
                format!(
                    r#"{{"ok":true,"database":"plugin.sqlite","columns":[],"rows":[],"usage":{usage}}}"#
                ),
            ),
        ];
        for (operation, payload) in &cases {
            let frame = hostcall_response_frame(FRAME_TYPE_STORAGE_SQLITE, payload);
            validate_storage_sqlite_response(&frame, "r1", "g1", operation)
                .unwrap_or_else(|err| panic!("{operation} response: {err}"));
            for (other, _) in &cases {
                if other != operation {
                    assert!(validate_storage_sqlite_response(&frame, "r1", "g1", other).is_err());
                }
            }
        }
    }

    #[test]
    fn renders_network_grant_frame() {
        let mut req = NetworkGrantRequest {
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            resource_scope: NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: "env_hash".to_string(),
                owner_user_hash: "user_hash".to_string(),
            },
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 3,
            connector_id: "api".to_string(),
            transport: "http".to_string(),
            destination: "https://api.example.com".to_string(),
            ttl_ms: 30000,
        };
        let frame =
            network_grant_frame("r1:network_grant", "g1", &req).expect("network grant frame");
        assert!(frame.contains(r#""frame_type":"network_grant""#));
        assert!(frame.contains(r#""connector_id":"api""#));
        assert!(frame.contains(r#""transport":"http""#));
        assert!(frame.contains(r#""resource_scope":{"kind":"user","owner_env_hash":"env_hash","owner_user_hash":"user_hash"}"#));
        assert!(frame.contains(r#""ttl_ms":30000"#));
        req.management_revision = MAX_JSON_SAFE_INTEGER + 1;
        assert_eq!(
            network_grant_frame("r1:network_grant", "g1", &req).unwrap_err(),
            IpcError::InvalidField {
                field: "management_revision"
            }
        );
    }

    #[test]
    fn validates_network_grant_response() {
        let scope = NetworkResourceScope {
            kind: "user".to_string(),
            owner_env_hash: "env_hash".to_string(),
            owner_user_hash: "user_hash".to_string(),
        };
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"network_grant","request_id":"r1:network_grant","runtime_generation_id":"g1","payload":{"ok":true,"grant_id":"netgrant_00112233445566778899aabbccddeeff","plugin_instance_id":"plugini_1","active_fingerprint":"sha256:active","resource_scope":{"kind":"user","owner_env_hash":"env_hash","owner_user_hash":"user_hash"},"policy_revision":1,"management_revision":2,"revoke_epoch":3,"connector_id":"api","transport":"http","destination":{"transport":"http","scheme":"https","host":"api.example.com","port":443},"runtime_generation_id":"g1","target_classifier_version":"target-classifier-v2","expires_at":"2026-06-30T10:00:30Z"}}"#;
        validate_network_grant_response(frame, "r1:network_grant", "g1", "api", "http", &scope)
            .expect("valid network grant response");
        let unsafe_revision = frame.replace(
            r#""policy_revision":1"#,
            r#""policy_revision":9007199254740992"#,
        );
        assert_eq!(
            validate_network_grant_response(
                &unsafe_revision,
                "r1:network_grant",
                "g1",
                "api",
                "http",
                &scope,
            )
            .unwrap_err(),
            IpcError::InvalidField {
                field: "policy_revision"
            }
        );
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"network_grant","request_id":"r1:network_grant","runtime_generation_id":"g1","payload":{"ok":false,"code":"NETWORK_TARGET_DENIED","message":"blocked","error_origin":"hostcall"}}"#;
        let err = validate_network_grant_response(
            failed,
            "r1:network_grant",
            "g1",
            "api",
            "http",
            &scope,
        )
        .expect_err("failed network grant response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "NETWORK_TARGET_DENIED".to_string()
            }
        );
    }

    #[test]
    fn renders_network_execute_frame() {
        let mut req = NetworkExecuteRequest {
            plugin_id: "com.example.worker".to_string(),
            plugin_instance_id: "plugini_1".to_string(),
            active_fingerprint: "sha256:active".to_string(),
            resource_scope: NetworkResourceScope {
                kind: "user".to_string(),
                owner_env_hash: "env_hash".to_string(),
                owner_user_hash: "user_hash".to_string(),
            },
            runtime_instance_id: "runtime_1".to_string(),
            runtime_generation_id: "g1".to_string(),
            runtime_shard_id: "runtime_shard_1".to_string(),
            policy_revision: 1,
            management_revision: 2,
            revoke_epoch: 3,
            connector_id: "api".to_string(),
            transport: "http".to_string(),
            destination: "https://api.example.com".to_string(),
            ttl_ms: 30000,
            operation: "http".to_string(),
            method: "POST".to_string(),
            path: "/v1/worker".to_string(),
            query_json: r#"{"lang":["en"],"units":["metric"]}"#.to_string(),
            headers_json: r#"{"X-Test":["ok"]}"#.to_string(),
            message_type: "".to_string(),
            body_base64: "e30=".to_string(),
            payload_base64: "".to_string(),
            max_request_bytes: 1024,
            max_response_bytes: 2048,
            max_chunk_bytes: 256,
            max_buffered_bytes: 65536,
            timeout_ms: 2000,
            stream_id: "stream_1".to_string(),
            stream_method: "worker.echo".to_string(),
            stream_effect: "read".to_string(),
            stream_execution: "subscription".to_string(),
            surface_instance_id: "surface_1".to_string(),
            owner_session_hash: "session_hash".to_string(),
            owner_user_hash: "user_hash".to_string(),
            owner_env_hash: "env_hash".to_string(),
            session_channel_id_hash: "channel_hash".to_string(),
            bridge_channel_id: "bridge_1".to_string(),
            content_type: "text/plain".to_string(),
        };
        let frame =
            network_execute_frame("r1:network_execute", "g1", &req).expect("network execute frame");
        assert!(frame.contains(r#""frame_type":"network_execute""#));
        assert!(frame.contains(r#""operation":"http""#));
        assert!(frame.contains(r#""headers":{"X-Test":["ok"]}"#));
        assert!(frame.contains(r#""query":{"lang":["en"],"units":["metric"]}"#));
        req.policy_revision = MAX_JSON_SAFE_INTEGER + 1;
        assert_eq!(
            network_execute_frame("r1:network_execute", "g1", &req).unwrap_err(),
            IpcError::InvalidField {
                field: "policy_revision"
            }
        );
        req.policy_revision = 1;
        assert!(frame.contains(r#""body_base64":"e30=""#));
        assert!(frame.contains(r#""stream_id":"stream_1""#));
        assert!(frame.contains(r#""owner_session_hash":"session_hash""#));
        assert!(frame.contains(r#""max_chunk_bytes":256"#));
        assert!(frame.contains(r#""timeout_ms":2000"#));

        req.query_json = "[]".to_string();
        assert_eq!(
            network_execute_frame("r1:network_execute", "g1", &req),
            Err(IpcError::InvalidField {
                field: "network execute query"
            })
        );
        req.query_json = "{}".to_string();
        req.headers_json = "[".to_string();
        assert_eq!(
            network_execute_frame("r1:network_execute", "g1", &req),
            Err(IpcError::InvalidField {
                field: "network execute headers"
            })
        );
        req.headers_json = "{}".to_string();
        req.resource_scope.owner_user_hash.clear();
        assert_eq!(
            network_execute_frame("r1:network_execute", "g1", &req),
            Err(IpcError::InvalidField {
                field: "network resource scope"
            })
        );
    }

    #[test]
    fn validates_network_execute_response() {
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"network_execute","request_id":"r1:network_execute","runtime_generation_id":"g1","payload":{"ok":true,"transport":"http","destination":{"transport":"http","scheme":"https","host":"api.example.com","port":443},"status_code":201,"headers":{"X-Worker":["ok"]},"body_base64":"e30=","grant_id":"netgrant_00112233445566778899aabbccddeeff","connector_id":"api","runtime_generation_id":"g1"}}"#;
        validate_network_execute_response(frame, "r1:network_execute", "g1", "api", "http")
            .expect("valid network execute response");
        let failed = r#"{"ipc_version":"rust-ipc-v6","frame_type":"network_execute","request_id":"r1:network_execute","runtime_generation_id":"g1","payload":{"ok":false,"code":"NETWORK_RESPONSE_TOO_LARGE","message":"too large","error_origin":"hostcall"}}"#;
        let err =
            validate_network_execute_response(failed, "r1:network_execute", "g1", "api", "http")
                .expect_err("failed network execute response");
        assert_eq!(
            err,
            IpcError::RemoteFailure {
                code: "NETWORK_RESPONSE_TOO_LARGE".to_string()
            }
        );
    }

    #[test]
    fn parses_worker_invocation_identity() {
        let frame = closed_worker_frame(
            "{}",
            r#"{"package_hash":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","artifact":"workers/backend.wasm","artifact_sha256":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","worker_id":"backend","method":"worker.echo"}"#,
        );
        let identity = parse_worker_invocation_identity(&frame).expect("valid invocation");
        assert_eq!(
            identity.package_hash,
            "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
        );
        assert_eq!(identity.artifact, "workers/backend.wasm");
        assert_eq!(identity.worker_id, "backend");
    }

    #[test]
    fn validates_worker_artifact_content_hash() {
        let identity = WorkerInvocationIdentity {
            package_hash: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                .to_string(),
            artifact: "workers/backend.wasm".to_string(),
            artifact_sha256:
                "sha256:2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
                    .to_string(),
            worker_id: "backend".to_string(),
            method: "worker.echo".to_string(),
        };
        validate_worker_artifact_bytes(&identity, b"hello").unwrap();
        assert!(validate_worker_artifact_bytes(&identity, b"tampered").is_err());
    }

    #[test]
    fn rejects_unsupported_worker_runtime_contract() {
        let valid = runtime_lease_invocation_fixture();
        parse_worker_invocation(valid)
            .unwrap()
            .validate_worker_contract()
            .unwrap();
        let invalid = valid.replacen(
            "redevplugin-wasm-worker-v2",
            "redevplugin-wasm-worker-v99",
            1,
        );
        assert!(
            parse_worker_invocation(&invalid)
                .unwrap()
                .validate_worker_contract()
                .is_err()
        );
    }

    #[test]
    fn projects_closed_worker_request_v2() {
        let frame = r#"{"ipc_version":"rust-ipc-v6","frame_type":"invoke_worker","request_id":"r1","runtime_generation_id":"g1","payload":{"lease":{},"method":"notes.save","invocation":{"plugin_id":"com.example.notes","plugin_instance_id":"plugini_1","storage_handle_grants":{"notes":"handle-secret"},"method":"notes.save","params":{"title":"Launch notes","body":"Ship the examples"}}}}"#;

        let request = worker_request_json_v2(frame).expect("worker request projection");

        assert_eq!(
            request,
            r#"{"schema_version":"redevplugin.worker_request.v2","method":"notes.save","params":{"body":"Ship the examples","title":"Launch notes"}}"#
        );
        assert!(!request.contains("handle-secret"));
        assert!(!request.contains("plugin_instance_id"));
    }

    #[test]
    fn reads_positive_worker_memory_limit_from_signed_lease() {
        let frame = closed_worker_frame(
            r#"{"limits":{"memory_bytes":33554432}}"#,
            r#"{"method":"worker.echo"}"#,
        );
        assert_eq!(
            runtime_lease_memory_limit_bytes(&frame).expect("memory limit"),
            33_554_432
        );
        for lease in [
            r#"{"limits":{}}"#,
            r#"{"limits":{"memory_bytes":0}}"#,
            r#"{"limits":{"memory_bytes":268435457}}"#,
        ] {
            assert!(
                runtime_lease_memory_limit_bytes(&closed_worker_frame(
                    lease,
                    r#"{"method":"worker.echo"}"#,
                ))
                .is_err()
            );
        }
    }

    #[test]
    fn read_effect_rejects_mutating_storage_broker_operations() {
        for operation in ["write", "delete", "put", "exec"] {
            let frame = format!(
                r#"{{"effect":"read","broker_access":{{"storage":[{{"store_id":"store","scope":"user","operations":["{operation}"]}}]}}}}"#
            );
            let frame = closed_worker_frame("{}", &frame);
            let err = validate_worker_storage_broker_access(&frame, "store", operation)
                .expect_err("read methods must not mutate storage");
            assert_eq!(
                err,
                IpcError::ProtocolViolation {
                    message: "worker method with read effect cannot perform the storage mutation"
                }
            );
        }
    }

    #[test]
    fn read_effect_allows_declared_http_post_network_request() {
        let frame = closed_worker_frame(
            "{}",
            r#"{"effect":"read","broker_access":{"network":[{"connector_id":"search","transport":"http","scope":"user","operations":["http"],"http_methods":["POST"]}]}}"#,
        );
        validate_worker_network_broker_access(&frame, "search", "http", "http", "POST")
            .expect("HTTP verbs do not define method effect");
    }

    #[test]
    fn parses_worker_response_v2_success_and_rejects_extra_authority() {
        let success =
            parse_worker_response_v2(r#"{"ok":true,"data":{"saved":true,"id":"note_1"}}"#)
                .expect("worker success response");
        assert_eq!(
            success,
            WorkerResponseV2::Success(r#"{"saved":true,"id":"note_1"}"#.to_string())
        );

        let error = parse_worker_response_v2(
            r#"{"ok":true,"data":{"saved":true},"gateway_token":"secret"}"#,
        )
        .expect_err("extra response authority must fail closed");
        assert_eq!(
            error,
            IpcError::DecodeFailed {
                context: "worker response"
            }
        );
        assert!(!error.to_string().contains("gateway_token"));
        assert!(!error.to_string().contains("secret"));
    }

    #[test]
    fn worker_response_v2_enforces_closed_success_and_failure_branches() {
        let failure = parse_worker_response_v2(
            r#"{"ok":false,"error_code":"WORKER_FAILED","message":"failed"}"#,
        )
        .expect("worker failure response");
        assert_eq!(
            failure,
            WorkerResponseV2::Failure {
                code: "WORKER_FAILED".to_string(),
                message: "failed".to_string(),
            }
        );

        for invalid in [
            r#"{"ok":true}"#,
            r#"{"ok":true,"data":{},"error_code":"WORKER_FAILED"}"#,
            r#"{"ok":false,"data":{},"error_code":"WORKER_FAILED","message":"failed"}"#,
            r#"{"ok":false,"message":"failed"}"#,
            r#"{"ok":false,"error_code":"WORKER_FAILED"}"#,
        ] {
            assert!(
                parse_worker_response_v2(invalid).is_err(),
                "accepted ambiguous worker response {invalid}"
            );
        }
    }

    #[test]
    fn worker_response_v2_preserves_large_raw_success_payload() {
        let payload = "x".repeat(512 * 1024 - 64);
        let input = format!(r#"{{"ok":true,"data":{{"payload":"{payload}"}}}}"#);
        let response = parse_worker_response_v2(&input).expect("large worker response");
        match response {
            WorkerResponseV2::Success(data) => {
                assert_eq!(data.len(), payload.len() + r#"{"payload":""}"#.len());
                assert!(data.ends_with("\"}"));
            }
            WorkerResponseV2::Failure { .. } => panic!("expected success response"),
        }
    }

    #[test]
    fn rejects_worker_invocation_without_artifact_identity() {
        let frame = closed_worker_frame("{}", r#"{"artifact":"../backend.wasm"}"#);
        let err = parse_worker_invocation_identity(&frame).expect_err("invalid invocation");
        assert_eq!(
            err,
            IpcError::MissingField {
                field: "package_hash"
            }
        );
        let frame = closed_worker_frame(
            "{}",
            r#"{"package_hash":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","artifact":"workers/../backend.wasm","artifact_sha256":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","worker_id":"backend","method":"worker.echo"}"#,
        );
        let err = parse_worker_invocation_identity(&frame).expect_err("invalid artifact");
        assert_eq!(err, IpcError::InvalidField { field: "artifact" });
    }

    #[test]
    fn parses_worker_lease_replay_key() {
        let input = closed_worker_frame(
            r#"{"lease_id":"lease_1","lease_nonce":"nonce_1","expires_at_unix_ms":2000}"#,
            r#"{"method":"worker.echo"}"#,
        );
        let key = parse_worker_lease_replay_key(&input).expect("valid replay key");
        assert_eq!(key.lease_id, "lease_1");
        assert_eq!(key.lease_nonce, "nonce_1");
        assert_eq!(key.expires_at_unix_ms, 2_000);
    }

    #[test]
    fn rejects_worker_lease_replay_key_without_nonce() {
        let input = closed_worker_frame(r#"{"lease_id":"lease_1"}"#, r#"{"method":"worker.echo"}"#);
        let err = parse_worker_lease_replay_key(&input).expect_err("missing nonce should fail");
        assert_eq!(
            err,
            IpcError::MissingField {
                field: "lease_nonce"
            }
        );
    }

    fn runtime_lease_signing_key_for_test(seed_byte: u8) -> SigningKey {
        SigningKey::from_bytes(&[seed_byte; 32])
    }

    fn runtime_lease_invocation_fixture() -> &'static str {
        include_str!("../testdata/runtime-lease-signature-v1-invocation.json")
    }

    fn signed_runtime_lease_invocation_for_test(
        signing_key: &SigningKey,
        replace: Option<(&str, &str)>,
    ) -> String {
        let mut lease = serde_json::json!({
            "lease_id": "rel_lease_signature",
            "token_id": "rel_token_signature",
            "lease_nonce": "nonce_1234567890",
            "runtime_generation_id": "rtgen_1",
            "plugin_instance_id": "plugini_1",
            "plugin_id": "com.example.worker",
            "plugin_version": "1.2.3",
            "active_fingerprint": "sha256:active",
            "issued_at_unix_ms": 1783161900000_i64,
            "method": "worker.echo",
            "effect": "read",
            "execution": "sync",
            "audit_correlation_id": "audit_lease_signature",
            "surface_instance_id": "surface_runtime",
            "owner_session_hash": "session_hash",
            "owner_user_hash": "user_hash",
            "owner_env_hash": "env_hash",
            "session_channel_id_hash": "channel_hash",
            "bridge_channel_id": "bridge_runtime",
            "target_descriptor_hashes": [
                "method:sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                "worker:sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
            ],
            "limits": {
                "timeout_ms": 2000,
                "memory_bytes": 65536,
                "max_payload_bytes": 4096,
                "max_stream_bytes_per_sec": 1024
            },
            "policy_revision": 11,
            "management_revision": 12,
            "revoke_epoch": 13,
            "runtime_shard_id": "rtshard_1",
            "runtime_instance_id": "rtinst_1",
            "ipc_channel_id": "ipc_1",
            "connection_nonce": "connection_nonce_1234567890",
            "key_id": "host_ephemeral_key_1",
            "expires_at_unix_ms": 1783161930000_i64
        });
        let payload =
            runtime_lease_signature_payload_json(&worker_lease_from_value(&lease), "worker.echo")
                .expect("payload");
        let signature = signing_key.sign(payload.as_bytes());
        lease["signature"] = serde_json::Value::String(format!(
            "ed25519:{}",
            base64::engine::general_purpose::STANDARD.encode(signature.to_bytes())
        ));
        if let Some((key, value)) = replace {
            let parsed = value.parse::<u64>().expect("numeric replacement");
            lease[key] = serde_json::Value::Number(parsed.into());
        }
        format!(
            r#"{{"ipc_version":"rust-ipc-v6","frame_type":"invoke_worker","request_id":"r1","runtime_generation_id":"rtgen_1","payload":{{"lease":{},"method":"worker.echo","invocation":{{"package_hash":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","artifact":"workers/backend.wasm","artifact_sha256":"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","worker_id":"backend","method":"worker.echo"}}}}}}"#,
            serde_json::to_string(&lease).expect("lease json")
        )
    }
}