lean_ctx_protocol/
runtime_exchange.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Strict payloads for the optional local Intelligence peer, not a second Engine.
3//! Transport authenticates exact bytes before decoding and owns replay state.
4//! Only explicitly policy-approved projections belong here; never log payloads.
5
6use serde::{Deserialize, Serialize, de::DeserializeOwned};
7
8use crate::{
9    EngineInvocationV1, EngineObservationV1, EnginePolicyDecisionV1, Sha256Digest, ValidationError,
10    receipt_document::digest_bytes, validate_bounded_opaque_identifier,
11};
12
13/// Independent successor to the legacy, one-way runtime notification.
14pub const RUNTIME_EXCHANGE_VERSION: &str = "leanctx.runtime-exchange/v1";
15/// Encoded JSON limit; transport must check the length prefix before allocation.
16pub const MAX_RUNTIME_EXCHANGE_BYTES: usize = 1_048_576;
17/// A peer cannot extend an invocation beyond this host-admitted time budget.
18pub const MAX_RUNTIME_DEADLINE_MS: u64 = 60_000;
19
20/// One admitted invocation plus its exact bounded input projection.
21/// No Debug implementation: the projection may contain private user content.
22#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
23#[serde(deny_unknown_fields)]
24pub struct RuntimeRequestV1 {
25    pub protocol_version: String,
26    pub session_id: String,
27    pub request_id: String,
28    /// Strictly increasing within the authenticated session; zero is reserved.
29    pub sequence: u64,
30    pub deadline_unix_ms: u64,
31    pub invocation: EngineInvocationV1,
32    pub input: String,
33}
34
35impl RuntimeRequestV1 {
36    /// Validate structure and input binding; this does not grant authorization.
37    pub fn validate(&self) -> Result<(), ValidationError> {
38        validate_version(&self.protocol_version)?;
39        validate_bounded_opaque_identifier(&self.session_id, "runtime session_id")?;
40        validate_bounded_opaque_identifier(&self.request_id, "runtime request_id")?;
41        self.invocation.validate()?;
42        if self.sequence == 0 || self.deadline_unix_ms == 0 {
43            return Err(ValidationError::new(
44                "runtime sequence/deadline must be nonzero",
45            ));
46        }
47        if self.invocation.policy_admission.decision != EnginePolicyDecisionV1::Admitted {
48            return Err(ValidationError::new(
49                "runtime input requires host policy admission",
50            ));
51        }
52        validate_content(self.input.as_bytes(), &self.invocation.input_digest)
53    }
54
55    /// Deterministically check time admission using the receiver's clock.
56    /// The transport also enforces a monotonic I/O deadline for the entire call.
57    pub fn validate_at(&self, now_unix_ms: u64) -> Result<(), ValidationError> {
58        self.validate()?;
59        let remaining = self.deadline_unix_ms.saturating_sub(now_unix_ms);
60        if remaining == 0 || remaining > MAX_RUNTIME_DEADLINE_MS {
61            return Err(ValidationError::new(
62                "runtime deadline expired or exceeds budget",
63            ));
64        }
65        Ok(())
66    }
67
68    /// Encode the exact stable representation used by response correlation.
69    pub fn to_bytes(&self) -> Result<Vec<u8>, ValidationError> {
70        self.validate()?;
71        encode(self)
72    }
73
74    /// Decode bounded JSON, rejecting unknown fields, duplicates and bad bindings.
75    /// Equivalent JSON whitespace/key order is accepted after authentication of
76    /// the exact received bytes; decoding is not transport authentication.
77    pub fn from_bytes(bytes: &[u8], now_unix_ms: u64) -> Result<Self, ValidationError> {
78        let request: Self = decode(bytes)?;
79        request.validate_at(now_unix_ms)?;
80        Ok(request)
81    }
82
83    /// Bind all request fields, not merely a caller-provided invocation ID.
84    /// This is a semantic correlation digest of stable re-encoding, not a MAC
85    /// of the received representation. Transport authenticates bytes separately.
86    pub fn digest(&self) -> Result<Sha256Digest, ValidationError> {
87        digest_bytes(&self.to_bytes()?)
88    }
89}
90
91/// Peer result bound to the whole request; receipts remain host-owned.
92#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
93#[serde(deny_unknown_fields)]
94pub struct RuntimeResponseV1 {
95    pub protocol_version: String,
96    pub request_digest: Sha256Digest,
97    pub observation: EngineObservationV1,
98    pub output: Option<String>,
99}
100
101impl RuntimeResponseV1 {
102    /// Reuse canonical Engine lineage/status validation and check exact content.
103    pub fn validate_for(&self, request: &RuntimeRequestV1) -> Result<(), ValidationError> {
104        validate_version(&self.protocol_version)?;
105        if self.request_digest != request.digest()? {
106            return Err(ValidationError::new(
107                "runtime response belongs to another request",
108            ));
109        }
110        self.observation.validate_for(&request.invocation)?;
111        if self.observation.receipt_link.is_some() {
112            return Err(ValidationError::new(
113                "runtime peer cannot issue a host receipt",
114            ));
115        }
116        match (&self.output, &self.observation.output_digest) {
117            (Some(output), Some(digest)) => validate_content(output.as_bytes(), digest),
118            (None, None) => Ok(()),
119            _ => Err(ValidationError::new(
120                "runtime output and digest must be present together",
121            )),
122        }
123    }
124
125    /// Encode only a result that is bound to the admitted request.
126    pub fn to_bytes(&self, request: &RuntimeRequestV1) -> Result<Vec<u8>, ValidationError> {
127        self.validate_for(request)?;
128        encode(self)
129    }
130
131    /// Reject late, mismatched, malformed or oversized replies before use.
132    pub fn from_bytes(
133        bytes: &[u8],
134        request: &RuntimeRequestV1,
135        now_unix_ms: u64,
136    ) -> Result<Self, ValidationError> {
137        request.validate_at(now_unix_ms)?;
138        let response: Self = decode(bytes)?;
139        response.validate_for(request)?;
140        Ok(response)
141    }
142}
143
144fn validate_version(version: &str) -> Result<(), ValidationError> {
145    if version != RUNTIME_EXCHANGE_VERSION {
146        return Err(ValidationError::new("unsupported runtime exchange version"));
147    }
148    Ok(())
149}
150
151fn validate_content(bytes: &[u8], digest: &Sha256Digest) -> Result<(), ValidationError> {
152    if bytes.len() > MAX_RUNTIME_EXCHANGE_BYTES || digest_bytes(bytes)? != *digest {
153        return Err(ValidationError::new(
154            "runtime content exceeds limit or differs from digest",
155        ));
156    }
157    Ok(())
158}
159
160fn encode(value: &impl Serialize) -> Result<Vec<u8>, ValidationError> {
161    let bytes = serde_json::to_vec(value)
162        .map_err(|_| ValidationError::new("runtime JSON encoding failed"))?;
163    if bytes.len() > MAX_RUNTIME_EXCHANGE_BYTES {
164        return Err(ValidationError::new(
165            "runtime message exceeds encoded limit",
166        ));
167    }
168    Ok(bytes)
169}
170
171fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, ValidationError> {
172    if bytes.is_empty() || bytes.len() > MAX_RUNTIME_EXCHANGE_BYTES {
173        return Err(ValidationError::new("runtime message length is invalid"));
174    }
175    serde_json::from_slice(bytes).map_err(|_| ValidationError::new("invalid runtime JSON"))
176}
177
178#[cfg(test)]
179mod tests;
lean_ctx_protocol/runtime_exchange.rs

lean_ctx_protocol/
runtime_exchange.rs
