rustev_contract/
remote.rs

1//! The `rustev.remote/1` protocol documents (spec 009, 3.10) and the
2//! vendor-neutral exchange record `rustev.remote-exchange/1` (3.9.1).
3//!
4//! Serde types only: no transport, no async runtime, and no vendor, model or
5//! provider name (spec 009, I-7). Every document denies unknown fields,
6//! carries an exact schema string and parses under a declared limit set
7//! through the bounded parser ([`crate::Document::parse`]). On the wire a
8//! document is its record canonical bytes (spec 004, 5.6), which equal its
9//! canonical bytes whenever it holds no fractional number.
10//!
11//! Nothing a remote party returns is authorization (spec 009, 3.1): an
12//! infer answer is a [`RawOutput`] or a classified failure, and everything
13//! else it carries is retained in the exchange record, which no policy
14//! reads.
15
16use std::collections::BTreeMap;
17
18use serde::{Deserialize, Serialize};
19
20use crate::descriptor::BackendDescriptor;
21use crate::ids::{ArtifactId, DescriptorId};
22use crate::limits::{EXCHANGE_V1, REMOTE_V1};
23use crate::output::RawOutput;
24use crate::run::{Charge, CostBound};
25use crate::schema;
26
27/// The protocol's major version.
28pub const PROTOCOL_MAJOR: u32 = 1;
29/// The protocol's name, as recorded in exchange records.
30pub const PROTOCOL: &str = "rustev.remote/1";
31/// `rustev.remote/2` (spec 020, 2): version 1 plus the wire members spec 020
32/// defines, today the host-endpoint principal reference of an infer request
33/// (3.6). The documents keep their schema strings and carry `version: 2`.
34pub const PROTOCOL_MAJOR_V2: u32 = 2;
35pub const PROTOCOL_V2: &str = "rustev.remote/2";
36/// The most bytes of an opaque principal reference (spec 020, 3.6).
37pub const MAX_PRINCIPAL_BYTES: usize = 128;
38/// The most bytes of a principal mapping version or scope.
39pub const MAX_PRINCIPAL_LABEL_BYTES: usize = 64;
40/// The domain tag of a principal reference's evidence digest.
41pub const PRINCIPAL_DIGEST_TAG: &str = "rustev.principal-reference/1";
42
43/// The protocol name of a major version this crate defines.
44pub const fn protocol_name(major: u32) -> Option<&'static str> {
45    match major {
46        PROTOCOL_MAJOR => Some(PROTOCOL),
47        PROTOCOL_MAJOR_V2 => Some(PROTOCOL_V2),
48        _ => None,
49    }
50}
51/// Free text in a failure detail or an exchange record is cut to this many
52/// bytes (spec 009, 3.9.1; spec 003, 3.9.3).
53pub const MAX_TEXT_BYTES: usize = 256;
54/// The most items one infer request may carry, whatever the terms say, and
55/// the most members an exchange record lists.
56pub const MAX_ITEMS: usize = 32;
57/// The most route entries an exchange record keeps.
58pub const MAX_ROUTE: usize = 8;
59/// The most entries of each map (usage, privacy options, provider extras)
60/// an exchange record keeps.
61pub const MAX_ENTRIES: usize = 16;
62
63/// `s` cut to at most [`MAX_TEXT_BYTES`] bytes, on a character boundary.
64pub fn cut_text(s: &str) -> String {
65    cut_to(s, MAX_TEXT_BYTES)
66}
67
68/// `s` cut to at most `max` bytes, on a character boundary.
69pub fn cut_to(s: &str, max: usize) -> String {
70    if s.len() <= max {
71        return s.to_string();
72    }
73    let mut end = max;
74    while !s.is_char_boundary(end) {
75        end -= 1;
76    }
77    s[..end].to_string()
78}
79
80// ---------------------------------------------------------------------------
81// Vocabulary shared by the documents.
82
83/// The closed error taxonomy (spec 009, 3.6). Each code maps to exactly one
84/// spec 003 failure class; the mapping lives with the adapters, because the
85/// failure classes of the backend seam are not contract types.
86#[derive(
87    Debug,
88    Clone,
89    Copy,
90    PartialEq,
91    Eq,
92    PartialOrd,
93    Ord,
94    Hash,
95    Serialize,
96    Deserialize,
97    schemars::JsonSchema,
98)]
99#[serde(rename_all = "snake_case")]
100pub enum RemoteCode {
101    TransportUnsent,
102    TransportInterrupted,
103    RateLimited,
104    Overloaded,
105    Unauthorized,
106    Capability,
107    MalformedRequest,
108    MalformedResponse,
109    IdentityMismatch,
110    RemoteDeadline,
111    RemoteError,
112    Cancelled,
113}
114
115impl RemoteCode {
116    /// Every code, in the order of the table in spec 009, 3.6.
117    pub const ALL: [RemoteCode; 12] = [
118        RemoteCode::TransportUnsent,
119        RemoteCode::TransportInterrupted,
120        RemoteCode::RateLimited,
121        RemoteCode::Overloaded,
122        RemoteCode::Unauthorized,
123        RemoteCode::Capability,
124        RemoteCode::MalformedRequest,
125        RemoteCode::MalformedResponse,
126        RemoteCode::IdentityMismatch,
127        RemoteCode::RemoteDeadline,
128        RemoteCode::RemoteError,
129        RemoteCode::Cancelled,
130    ];
131
132    /// The code as it appears on the wire and in failure details.
133    pub const fn as_str(self) -> &'static str {
134        match self {
135            RemoteCode::TransportUnsent => "transport_unsent",
136            RemoteCode::TransportInterrupted => "transport_interrupted",
137            RemoteCode::RateLimited => "rate_limited",
138            RemoteCode::Overloaded => "overloaded",
139            RemoteCode::Unauthorized => "unauthorized",
140            RemoteCode::Capability => "capability",
141            RemoteCode::MalformedRequest => "malformed_request",
142            RemoteCode::MalformedResponse => "malformed_response",
143            RemoteCode::IdentityMismatch => "identity_mismatch",
144            RemoteCode::RemoteDeadline => "remote_deadline",
145            RemoteCode::RemoteError => "remote_error",
146            RemoteCode::Cancelled => "cancelled",
147        }
148    }
149}
150
151/// How far the remote side supports cancellation (spec 009, 3.2.4).
152#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
153#[serde(rename_all = "snake_case")]
154pub enum CancelSupport {
155    None,
156    /// A cancel is accepted; a stop is never confirmed.
157    BestEffort,
158    /// A cancel is answered with a confirmed stop and its final charge.
159    Confirmed,
160}
161
162/// What a served identity is worth (spec 009, 3.7).
163#[derive(
164    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, schemars::JsonSchema,
165)]
166#[serde(rename_all = "snake_case")]
167pub enum Disclosure {
168    /// The remote side guarantees the identity.
169    Pinned,
170    /// The remote side said so; unverified.
171    Reported,
172    /// Nothing is known. Never replaced by a requested model or a slug.
173    Unknown,
174}
175
176/// Served-identity disclosure as negotiated. `identity` is the pinned
177/// identity, present exactly when `disclosure` is `pinned`.
178#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
179#[serde(deny_unknown_fields)]
180pub struct ServedTerms {
181    pub disclosure: Disclosure,
182    pub identity: Option<String>,
183}
184
185/// Shared-state batching support (spec 009, 3.10.4).
186#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
187#[serde(rename_all = "snake_case", deny_unknown_fields)]
188pub enum Batching {
189    None,
190    SharedState { max_items: u32 },
191}
192
193/// Terms negotiated beside the descriptor (spec 009, 3.2.4). The runtime
194/// never reads them; the adapter enforces and records them.
195#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
196#[serde(deny_unknown_fields)]
197pub struct RemoteTerms {
198    /// The per-call cost disclosure, in the remote side's units. `bounded`
199    /// only when the remote side enforces that per-call maximum (3.8.1).
200    pub cost: CostBound,
201    /// False only when the remote side documents that refused requests are
202    /// not charged (3.6), which permits `observed{0}` for a refusal.
203    pub refusals_charged: bool,
204    pub cancellation: CancelSupport,
205    pub served: ServedTerms,
206    pub batching: Batching,
207    pub max_request_bytes: u64,
208    pub max_response_bytes: u64,
209    /// How long an attempt id is remembered for idempotency (3.3.2).
210    pub idempotency_window_ms: u64,
211    /// Privacy options the remote side accepts (3.8.2), by name.
212    pub privacy: Vec<String>,
213}
214
215/// A served identity as recorded (spec 009, 3.7).
216#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
217#[serde(deny_unknown_fields)]
218pub struct ServedIdentity {
219    pub identity: Option<String>,
220    pub disclosure: Disclosure,
221}
222
223impl ServedIdentity {
224    /// The explicit `unknown` value.
225    pub fn unknown() -> Self {
226        ServedIdentity {
227            identity: None,
228            disclosure: Disclosure::Unknown,
229        }
230    }
231}
232
233// ---------------------------------------------------------------------------
234// rustev.remote-describe/1
235
236/// The protocol major versions the client supports (spec 009, 3.10.1).
237#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
238#[serde(deny_unknown_fields)]
239pub struct DescribeRequest {
240    pub schema: String,
241    pub versions: Vec<u32>,
242}
243
244crate::document::document!(DescribeRequest, schema::REMOTE_DESCRIBE, REMOTE_V1);
245
246/// The chosen version and, per hosted backend, its descriptor, descriptor
247/// identity and terms.
248#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
249#[serde(deny_unknown_fields)]
250pub struct DescribeResponse {
251    pub schema: String,
252    pub version: u32,
253    pub backends: Vec<HostedBackend>,
254}
255
256crate::document::document!(DescribeResponse, schema::REMOTE_DESCRIBE, REMOTE_V1);
257
258#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
259#[serde(deny_unknown_fields)]
260pub struct HostedBackend {
261    pub descriptor: BackendDescriptor,
262    pub descriptor_id: DescriptorId,
263    pub terms: RemoteTerms,
264}
265
266// ---------------------------------------------------------------------------
267// rustev.remote-infer/1
268
269#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
270#[serde(deny_unknown_fields)]
271pub struct InferRequest {
272    pub schema: String,
273    pub version: u32,
274    pub backend_id: String,
275    /// The negotiated descriptor; a different one is `identity_mismatch`.
276    pub descriptor: DescriptorId,
277    /// The negotiated remote artifact.
278    pub artifact: ArtifactId,
279    /// The attempt's remaining budget minus the declared transit margin.
280    pub budget_ms: u64,
281    pub trace_id: String,
282    /// Privacy options requested (3.8.2); never claimed honored.
283    pub privacy: BTreeMap<String, String>,
284    pub items: Vec<InferItem>,
285    /// `rustev.remote/2` only, and only to an endpoint the host operates:
286    /// the opaque principal reference (spec 020, 3.6). Routing and isolation
287    /// context, never a credential, grant or policy input. Absent otherwise,
288    /// so a version 1 request keeps its bytes.
289    #[serde(default, skip_serializing_if = "Option::is_none")]
290    pub principal: Option<WirePrincipal>,
291}
292
293crate::document::document!(InferRequest, schema::REMOTE_INFER, REMOTE_V1);
294
295impl InferRequest {
296    /// Whether the version-specific members agree with `version`: a
297    /// principal reference is a `rustev.remote/2` member, and a present one
298    /// is canonical and within its bounds.
299    pub fn members_fit_version(&self) -> bool {
300        match &self.principal {
301            None => true,
302            Some(p) => self.version == PROTOCOL_MAJOR_V2 && p.is_canonical(),
303        }
304    }
305}
306
307/// An opaque principal reference on the wire (spec 020, 3.6): the host's
308/// bytes as lowercase hexadecimal, the host's mapping version, and the
309/// declared scope. Rustev never interprets, authenticates, authorizes or
310/// expands it.
311#[derive(Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
312#[serde(deny_unknown_fields)]
313pub struct WirePrincipal {
314    pub reference: String,
315    pub mapping_version: String,
316    pub scope: String,
317}
318
319impl std::fmt::Debug for WirePrincipal {
320    /// The reference never prints; only its evidence digest does.
321    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
322        f.debug_struct("WirePrincipal")
323            .field("reference", &"<redacted>")
324            .field("mapping_version", &self.mapping_version)
325            .field("scope", &self.scope)
326            .finish()
327    }
328}
329
330fn is_label(s: &str) -> bool {
331    !s.is_empty() && s.len() <= MAX_PRINCIPAL_LABEL_BYTES && s.bytes().all(|b| b.is_ascii_graphic())
332}
333
334fn hex(bytes: &[u8]) -> String {
335    let mut s = String::with_capacity(bytes.len() * 2);
336    for b in bytes {
337        s.push_str(&format!("{b:02x}"));
338    }
339    s
340}
341
342impl WirePrincipal {
343    /// Encode `reference` canonically. `None` when the reference is empty
344    /// or beyond [`MAX_PRINCIPAL_BYTES`], or a label is empty, beyond
345    /// [`MAX_PRINCIPAL_LABEL_BYTES`] or not printable ASCII.
346    pub fn encode(reference: &[u8], mapping_version: &str, scope: &str) -> Option<Self> {
347        let p = WirePrincipal {
348            reference: hex(reference),
349            mapping_version: mapping_version.to_string(),
350            scope: scope.to_string(),
351        };
352        p.is_canonical().then_some(p)
353    }
354
355    /// The reference's bytes, when it is canonical lowercase hexadecimal.
356    pub fn reference_bytes(&self) -> Option<Vec<u8>> {
357        let r = self.reference.as_bytes();
358        if r.is_empty() || !r.len().is_multiple_of(2) || r.len() > MAX_PRINCIPAL_BYTES * 2 {
359            return None;
360        }
361        let nibble = |c: u8| match c {
362            b'0'..=b'9' => Some(c - b'0'),
363            b'a'..=b'f' => Some(c - b'a' + 10),
364            _ => None,
365        };
366        r.chunks(2)
367            .map(|p| Some(nibble(p[0])? << 4 | nibble(p[1])?))
368            .collect()
369    }
370
371    pub fn is_canonical(&self) -> bool {
372        self.reference_bytes().is_some() && is_label(&self.mapping_version) && is_label(&self.scope)
373    }
374
375    /// What evidence holds instead of the reference: a domain-separated
376    /// digest over the mapping version, the scope and the reference.
377    pub fn evidence(&self) -> PrincipalEvidence {
378        let mut b = Vec::new();
379        for part in [&self.mapping_version, &self.scope, &self.reference] {
380            b.extend_from_slice(&(part.len() as u64).to_be_bytes());
381            b.extend_from_slice(part.as_bytes());
382        }
383        PrincipalEvidence {
384            digest: crate::canonical::tagged_digest(PRINCIPAL_DIGEST_TAG, &b),
385            mapping_version: self.mapping_version.clone(),
386        }
387    }
388}
389
390/// A principal reference as evidence holds it (spec 020, 3.2 and 3.6): its
391/// domain-separated digest and mapping version, never the reference.
392#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
393#[serde(deny_unknown_fields)]
394pub struct PrincipalEvidence {
395    pub digest: String,
396    pub mapping_version: String,
397}
398
399#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
400#[serde(deny_unknown_fields)]
401pub struct InferItem {
402    /// The idempotency key (3.3).
403    pub attempt_id: String,
404    /// The projection's exact canonical bytes, which are UTF-8 JSON.
405    pub projection: String,
406}
407
408#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
409#[serde(deny_unknown_fields)]
410pub struct InferResponse {
411    pub schema: String,
412    pub version: u32,
413    /// The remote artifact that answered.
414    pub artifact: ArtifactId,
415    pub served: ServedIdentity,
416    /// Routing metadata as reported, opaque.
417    pub route: Vec<String>,
418    pub generation_id: Option<String>,
419    /// One answer per item, each attempt id at most once.
420    pub items: Vec<ItemAnswer>,
421    /// The exchange's charge, in the remote side's units.
422    pub charge: Charge,
423    /// Provider-reported extras: retained, never supplied (3.5.5).
424    pub extras: BTreeMap<String, String>,
425}
426
427crate::document::document!(InferResponse, schema::REMOTE_INFER, REMOTE_V1);
428
429#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
430#[serde(deny_unknown_fields)]
431pub struct ItemAnswer {
432    pub attempt_id: String,
433    pub result: ItemResult,
434}
435
436#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
437#[serde(rename_all = "snake_case", deny_unknown_fields)]
438pub enum ItemResult {
439    /// The `rustev.backend-output/1` vocabulary.
440    Output(RawOutput),
441    Failure {
442        code: RemoteCode,
443        detail: String,
444    },
445    /// The attempt id is already running at the endpoint (3.3.2).
446    InProgress,
447}
448
449// ---------------------------------------------------------------------------
450// rustev.remote-cancel/1
451
452#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
453#[serde(deny_unknown_fields)]
454pub struct CancelRequest {
455    pub schema: String,
456    pub version: u32,
457    pub backend_id: String,
458    pub attempt_ids: Vec<String>,
459}
460
461crate::document::document!(CancelRequest, schema::REMOTE_CANCEL, REMOTE_V1);
462
463#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
464#[serde(deny_unknown_fields)]
465pub struct CancelResponse {
466    pub schema: String,
467    pub version: u32,
468    pub answers: Vec<CancelItem>,
469}
470
471crate::document::document!(CancelResponse, schema::REMOTE_CANCEL, REMOTE_V1);
472
473#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
474#[serde(deny_unknown_fields)]
475pub struct CancelItem {
476    pub attempt_id: String,
477    pub answer: RemoteCancelAnswer,
478}
479
480#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
481#[serde(rename_all = "snake_case", deny_unknown_fields)]
482pub enum RemoteCancelAnswer {
483    /// Work and charges stopped; the final charge, in remote units.
484    Stopped {
485        charge: Charge,
486    },
487    Unconfirmed,
488    UnknownAttempt,
489}
490
491// ---------------------------------------------------------------------------
492// rustev.remote-exchange/1
493
494/// One exchange, keyed by the attempt ids it served (spec 009, 3.9.1).
495/// Bounded: every string is cut to [`MAX_TEXT_BYTES`] and every collection
496/// to its cap by [`ExchangeRecord::bounded`], so a record never exceeds
497/// [`EXCHANGE_V1`]'s `max_bytes`. Request and response bytes are never in
498/// the record; only their digests are.
499#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
500#[serde(deny_unknown_fields)]
501pub struct ExchangeRecord {
502    pub schema: String,
503    /// The wire protocol: [`PROTOCOL`] or a foreign mapping's own label.
504    pub protocol: String,
505    /// The adapter binding identity (the adapter's artifact, 3.2.3).
506    pub binding: ArtifactId,
507    /// Every member attempt, in attempt-id order.
508    pub members: Vec<ExchangeMember>,
509    /// `sha256:` digest of the request bytes, when any were built.
510    pub request_digest: Option<String>,
511    /// `sha256:` digest of the response bytes, when any were read.
512    pub response_digest: Option<String>,
513    /// Whether any request byte left the process.
514    pub bytes_sent: bool,
515    pub status: Option<u16>,
516    /// Recorded verbatim, never slept on.
517    pub retry_after: Option<String>,
518    pub requested_model: Option<String>,
519    pub served: ServedIdentity,
520    pub route: Vec<String>,
521    pub generation_id: Option<String>,
522    /// Usage as reported (for example token counts), by name.
523    pub usage: BTreeMap<String, u64>,
524    /// Cost as reported, verbatim; units are not money.
525    pub reported_cost: Option<ReportedCost>,
526    /// The exchange's charge in deployment units, before attribution.
527    pub exchange_charge: Charge,
528    pub attribution: Attribution,
529    pub privacy_requested: BTreeMap<String, String>,
530    /// Provider-reported extras (3.5.5): argmax, confidence, explanations.
531    pub provider_extras: BTreeMap<String, String>,
532    /// The response arrived after the attempts stopped waiting; nothing in
533    /// it was supplied (3.4.4).
534    pub late: bool,
535    /// Something was cut to fit the record's bounds.
536    pub truncated: bool,
537    /// The digest and mapping version of the principal reference the
538    /// request carried (spec 020, 3.6); absent when none was sent.
539    #[serde(default, skip_serializing_if = "Option::is_none")]
540    pub principal: Option<PrincipalEvidence>,
541    /// The transport phase that ran out, when an enabled phase policy
542    /// governed the exchange (spec 020, 3.4). Absent otherwise, so a record
543    /// without one keeps its spec 009 bytes.
544    #[serde(default, skip_serializing_if = "Option::is_none")]
545    pub phase_expiry: Option<PhaseExpiry>,
546    /// The pooled connection the exchange used (spec 020, 3.5); absent
547    /// when connection reuse is disabled.
548    #[serde(default, skip_serializing_if = "Option::is_none")]
549    pub connection: Option<ConnectionObservation>,
550}
551
552crate::document::document!(ExchangeRecord, schema::REMOTE_EXCHANGE, EXCHANGE_V1);
553
554/// Whether an exchange's connection was opened for it or reused.
555#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
556#[serde(rename_all = "snake_case")]
557pub enum ConnectionReuse {
558    Fresh,
559    Reused,
560}
561
562/// A connection's age when an exchange took it, bucketed so evidence holds
563/// no timing beyond what the bound needs.
564#[derive(
565    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, schemars::JsonSchema,
566)]
567#[serde(rename_all = "snake_case")]
568pub enum AgeBucket {
569    Under1s,
570    Under10s,
571    Under60s,
572    Under600s,
573    Over600s,
574}
575
576impl AgeBucket {
577    pub fn of_ms(ms: u64) -> Self {
578        match ms {
579            0..1_000 => AgeBucket::Under1s,
580            1_000..10_000 => AgeBucket::Under10s,
581            10_000..60_000 => AgeBucket::Under60s,
582            60_000..600_000 => AgeBucket::Under600s,
583            _ => AgeBucket::Over600s,
584        }
585    }
586}
587
588/// How requests share a connection. HTTP/1.1, the only binding today,
589/// carries one request at a time.
590#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
591#[serde(rename_all = "snake_case")]
592pub enum Multiplex {
593    None,
594}
595
596/// Why a pooled connection left the pool (spec 020, 3.5).
597#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, schemars::JsonSchema)]
598#[serde(rename_all = "snake_case")]
599pub enum EvictionReason {
600    /// Idle longer than the policy's idle lifetime.
601    IdleExpired,
602    /// Older than the policy's absolute lifetime.
603    LifetimeExpired,
604    /// Used as often as the policy allows.
605    MaxUses,
606    /// More idle connections than the policy allows.
607    IdleCapacity,
608    /// Closed to make room for a connection under another key.
609    Capacity,
610    /// Found closed by the remote side or the connection.
611    Closed,
612    /// The exchange on it failed or did not finish its response.
613    Failed,
614}
615
616/// The connection an exchange used (spec 020, 3.5): fresh or reused, a
617/// nonsecret instance id, its age bucket and use count, the multiplex
618/// posture, the isolation-key digest, and why it left the pool after this
619/// exchange, if it did.
620#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
621#[serde(deny_unknown_fields)]
622pub struct ConnectionObservation {
623    pub reuse: ConnectionReuse,
624    pub instance_id: String,
625    pub age_bucket: AgeBucket,
626    pub use_count: u64,
627    pub multiplex: Multiplex,
628    pub key_digest: String,
629    pub eviction: Option<EvictionReason>,
630}
631
632/// A transport phase with its own ceiling inside the one attempt budget
633/// (spec 020, 3.4), in the order an exchange passes through them.
634#[derive(
635    Debug,
636    Clone,
637    Copy,
638    PartialEq,
639    Eq,
640    PartialOrd,
641    Ord,
642    Hash,
643    Serialize,
644    Deserialize,
645    schemars::JsonSchema,
646)]
647#[serde(rename_all = "snake_case")]
648pub enum TransportPhase {
649    Resolution,
650    Connect,
651    SecureHandshake,
652    RequestWrite,
653    FirstResponseByte,
654    ResponseBody,
655    CancellationAck,
656}
657
658impl TransportPhase {
659    /// Every phase, in exchange order.
660    pub const ALL: [TransportPhase; 7] = [
661        TransportPhase::Resolution,
662        TransportPhase::Connect,
663        TransportPhase::SecureHandshake,
664        TransportPhase::RequestWrite,
665        TransportPhase::FirstResponseByte,
666        TransportPhase::ResponseBody,
667        TransportPhase::CancellationAck,
668    ];
669}
670
671/// What bounded a phase that ran out: its own ceiling, or the remaining
672/// attempt budget when that was smaller (spec 020, 3.4).
673#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
674#[serde(rename_all = "snake_case")]
675pub enum PhaseLimit {
676    Ceiling,
677    Budget,
678}
679
680/// The phase that ran out (spec 020, 3.4): its configured ceiling, the
681/// effective allowance it started with (the lesser of the ceiling and the
682/// remaining attempt budget), the time observed in it, whether request
683/// bytes had left, and the existing code the expiry mapped to.
684#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
685#[serde(deny_unknown_fields)]
686pub struct PhaseExpiry {
687    pub phase: TransportPhase,
688    pub ceiling_ms: u64,
689    pub allowance_ms: u64,
690    pub elapsed_ms: u64,
691    pub limit: PhaseLimit,
692    pub bytes_sent: bool,
693    pub code: RemoteCode,
694}
695
696#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
697#[serde(deny_unknown_fields)]
698pub struct ExchangeMember {
699    pub attempt_id: String,
700    pub outcome: MemberOutcome,
701    /// The charge supplied to the runtime for this member, or, in a late
702    /// record, the share offered for reconciliation.
703    pub charge: Charge,
704}
705
706#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
707#[serde(rename_all = "snake_case", deny_unknown_fields)]
708pub enum MemberOutcome {
709    /// An output; `digest` is `sha256:` of its record canonical bytes.
710    Output {
711        digest: String,
712    },
713    Failure {
714        code: RemoteCode,
715        detail: String,
716    },
717}
718
719#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
720#[serde(deny_unknown_fields)]
721pub struct ReportedCost {
722    /// A decimal string exactly as reported.
723    pub amount: String,
724    pub currency: String,
725}
726
727/// How an exchange's charge reached its members (spec 009, 3.10.4).
728#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
729#[serde(rename_all = "snake_case", deny_unknown_fields)]
730pub enum Attribution {
731    /// One member carries the whole charge.
732    Single,
733    /// Integer shares in attempt-id order over the members that received
734    /// the response; members that stopped waiting report `unknown`.
735    SharedState { coalescing_window_ms: u64 },
736    /// A batch the runtime planned and dispatched whole (spec 020, 3.3):
737    /// the runtime allocates the exchange charge over its members (spec
738    /// 017, 3.5), and members that stopped waiting report `unknown`.
739    RuntimePlanned,
740}
741
742fn cut_flag(s: &mut String, truncated: &mut bool) {
743    if s.len() > MAX_TEXT_BYTES {
744        *s = cut_text(s);
745        *truncated = true;
746    }
747}
748
749fn cut_opt(s: &mut Option<String>, truncated: &mut bool) {
750    if let Some(v) = s {
751        cut_flag(v, truncated);
752    }
753}
754
755fn cut_map<V>(m: BTreeMap<String, V>, truncated: &mut bool) -> BTreeMap<String, V>
756where
757    V: CutValue,
758{
759    if m.len() > MAX_ENTRIES {
760        *truncated = true;
761    }
762    let mut out = BTreeMap::new();
763    for (k, mut v) in m.into_iter().take(MAX_ENTRIES) {
764        let mut k = k;
765        cut_flag(&mut k, truncated);
766        v.cut(truncated);
767        out.entry(k).or_insert(v);
768    }
769    out
770}
771
772trait CutValue {
773    fn cut(&mut self, truncated: &mut bool);
774}
775
776impl CutValue for String {
777    fn cut(&mut self, truncated: &mut bool) {
778        cut_flag(self, truncated);
779    }
780}
781
782impl CutValue for u64 {
783    fn cut(&mut self, _: &mut bool) {}
784}
785
786impl ExchangeRecord {
787    /// The record cut to its fixed bounds: every string to
788    /// [`MAX_TEXT_BYTES`], members to [`MAX_ITEMS`], route to [`MAX_ROUTE`]
789    /// and each map to [`MAX_ENTRIES`], with `truncated` set when anything
790    /// was cut. The schema is set.
791    pub fn bounded(mut self) -> Self {
792        let mut t = self.truncated;
793        self.schema = schema::REMOTE_EXCHANGE.to_string();
794        cut_flag(&mut self.protocol, &mut t);
795        if self.members.len() > MAX_ITEMS {
796            self.members.truncate(MAX_ITEMS);
797            t = true;
798        }
799        for m in &mut self.members {
800            cut_flag(&mut m.attempt_id, &mut t);
801            if let MemberOutcome::Failure { detail, .. } = &mut m.outcome {
802                cut_flag(detail, &mut t);
803            }
804        }
805        cut_opt(&mut self.request_digest, &mut t);
806        cut_opt(&mut self.response_digest, &mut t);
807        cut_opt(&mut self.retry_after, &mut t);
808        cut_opt(&mut self.requested_model, &mut t);
809        cut_opt(&mut self.served.identity, &mut t);
810        if self.route.len() > MAX_ROUTE {
811            self.route.truncate(MAX_ROUTE);
812            t = true;
813        }
814        for r in &mut self.route {
815            cut_flag(r, &mut t);
816        }
817        cut_opt(&mut self.generation_id, &mut t);
818        self.usage = cut_map(std::mem::take(&mut self.usage), &mut t);
819        if let Some(c) = &mut self.reported_cost {
820            cut_flag(&mut c.amount, &mut t);
821            cut_flag(&mut c.currency, &mut t);
822        }
823        self.privacy_requested = cut_map(std::mem::take(&mut self.privacy_requested), &mut t);
824        self.provider_extras = cut_map(std::mem::take(&mut self.provider_extras), &mut t);
825        if let Some(c) = &mut self.connection {
826            cut_flag(&mut c.instance_id, &mut t);
827            cut_flag(&mut c.key_digest, &mut t);
828        }
829        if let Some(p) = &mut self.principal {
830            cut_flag(&mut p.digest, &mut t);
831            if p.mapping_version.len() > MAX_PRINCIPAL_LABEL_BYTES {
832                p.mapping_version = cut_to(&p.mapping_version, MAX_PRINCIPAL_LABEL_BYTES);
833                t = true;
834            }
835        }
836        self.truncated = t;
837        self
838    }
839
840    /// True when the record is within its bounds, as [`Self::bounded`]
841    /// leaves it. A parsed record that is not is refused by
842    /// [`Self::parse_checked`].
843    pub fn is_bounded(&self) -> bool {
844        let t = |s: &str| s.len() <= MAX_TEXT_BYTES;
845        let o = |s: &Option<String>| s.as_deref().is_none_or(t);
846        t(&self.protocol)
847            && self.members.len() <= MAX_ITEMS
848            && self.members.iter().all(|m| {
849                t(&m.attempt_id)
850                    && match &m.outcome {
851                        MemberOutcome::Failure { detail, .. } => t(detail),
852                        MemberOutcome::Output { digest } => t(digest),
853                    }
854            })
855            && o(&self.request_digest)
856            && o(&self.response_digest)
857            && o(&self.retry_after)
858            && o(&self.requested_model)
859            && o(&self.served.identity)
860            && self.route.len() <= MAX_ROUTE
861            && self.route.iter().all(|r| t(r))
862            && o(&self.generation_id)
863            && self.usage.len() <= MAX_ENTRIES
864            && self.usage.keys().all(|k| t(k))
865            && self
866                .reported_cost
867                .as_ref()
868                .is_none_or(|c| t(&c.amount) && t(&c.currency))
869            && [&self.privacy_requested, &self.provider_extras]
870                .iter()
871                .all(|m| m.len() <= MAX_ENTRIES && m.iter().all(|(k, v)| t(k) && t(v)))
872            && self.principal.as_ref().is_none_or(|p| {
873                t(&p.digest) && p.mapping_version.len() <= MAX_PRINCIPAL_LABEL_BYTES
874            })
875            && self
876                .connection
877                .as_ref()
878                .is_none_or(|c| t(&c.instance_id) && t(&c.key_digest))
879    }
880
881    /// Parse under [`EXCHANGE_V1`] and refuse a record beyond its bounds.
882    pub fn parse_checked(bytes: &[u8]) -> Result<Self, crate::DocumentError> {
883        use crate::Document;
884        let r = Self::parse(bytes)?;
885        if !r.is_bounded() {
886            return Err(crate::DocumentError::Parse(
887                crate::bounded::ParseError::Schema {
888                    message: "exchange record beyond its fixed bounds".into(),
889                },
890            ));
891        }
892        Ok(r)
893    }
894}
895
896#[cfg(test)]
897mod tests {
898    use super::*;
899    use crate::Document;
900    use crate::DocumentError;
901    use crate::bounded::{BoundError, ParseError};
902    use crate::definition::{Determinism, Operation};
903    use crate::descriptor::{InputExcess, InputLimit, OperationSupport, OutputKind};
904    use crate::{Identified, limits};
905
906    const A: &str = "sha256:00000000000000000000000000000000000000000000000000000000000000aa";
907
908    fn descriptor() -> BackendDescriptor {
909        BackendDescriptor {
910            ranking: None,
911            schema: schema::BACKEND.into(),
912            backend_id: "synthetic-remote".into(),
913            artifact: ArtifactId::parse(A).unwrap(),
914            operations: vec![OperationSupport {
915                isolation: None,
916                screen: None,
917                operation: Operation::Classify,
918                output: OutputKind::Distribution,
919                max_options: 8,
920            }],
921            input_limit: InputLimit {
922                max_bytes: 4096,
923                on_excess: InputExcess::Refuse,
924            },
925            determinism: Determinism::Unspecified,
926        }
927    }
928
929    fn terms() -> RemoteTerms {
930        RemoteTerms {
931            cost: CostBound::Bounded { max_units: 3 },
932            refusals_charged: false,
933            cancellation: CancelSupport::Confirmed,
934            served: ServedTerms {
935                disclosure: Disclosure::Pinned,
936                identity: Some("synthetic-model@1".into()),
937            },
938            batching: Batching::SharedState { max_items: 4 },
939            max_request_bytes: 65_536,
940            max_response_bytes: 65_536,
941            idempotency_window_ms: 60_000,
942            privacy: vec!["no_retention".into()],
943        }
944    }
945
946    fn describe_response() -> DescribeResponse {
947        let d = descriptor();
948        DescribeResponse {
949            schema: schema::REMOTE_DESCRIBE.into(),
950            version: 1,
951            backends: vec![HostedBackend {
952                descriptor_id: d.id().unwrap(),
953                descriptor: d,
954                terms: terms(),
955            }],
956        }
957    }
958
959    fn infer_request() -> InferRequest {
960        InferRequest {
961            schema: schema::REMOTE_INFER.into(),
962            version: 1,
963            backend_id: "synthetic-remote".into(),
964            descriptor: descriptor().id().unwrap(),
965            artifact: ArtifactId::parse(A).unwrap(),
966            budget_ms: 900,
967            trace_id: "d-1".into(),
968            privacy: BTreeMap::from([("no_retention".into(), "true".into())]),
969            items: vec![InferItem {
970                attempt_id: "d-1/topic//1".into(),
971                projection: r#"{"operation":"classify"}"#.into(),
972            }],
973            principal: None,
974        }
975    }
976
977    fn infer_response() -> InferResponse {
978        InferResponse {
979            schema: schema::REMOTE_INFER.into(),
980            version: 1,
981            artifact: ArtifactId::parse(A).unwrap(),
982            served: ServedIdentity {
983                identity: Some("synthetic-model@1".into()),
984                disclosure: Disclosure::Pinned,
985            },
986            route: vec!["synthetic-route".into()],
987            generation_id: Some("gen-1".into()),
988            items: vec![
989                ItemAnswer {
990                    attempt_id: "d-1/topic//1".into(),
991                    result: ItemResult::Output(RawOutput::Distribution(BTreeMap::from([
992                        ("a".into(), 0.1),
993                        ("b".into(), 0.9),
994                    ]))),
995                },
996                ItemAnswer {
997                    attempt_id: "d-1/topic//2".into(),
998                    result: ItemResult::Failure {
999                        code: RemoteCode::RemoteDeadline,
1000                        detail: "SYNTHETIC".into(),
1001                    },
1002                },
1003                ItemAnswer {
1004                    attempt_id: "d-1/topic//3".into(),
1005                    result: ItemResult::InProgress,
1006                },
1007            ],
1008            charge: Charge::Observed { units: 3 },
1009            extras: BTreeMap::from([("argmax".into(), "b".into())]),
1010        }
1011    }
1012
1013    fn cancel_docs() -> (CancelRequest, CancelResponse) {
1014        (
1015            CancelRequest {
1016                schema: schema::REMOTE_CANCEL.into(),
1017                version: 1,
1018                backend_id: "synthetic-remote".into(),
1019                attempt_ids: vec!["x/1".into(), "x/2".into(), "x/3".into()],
1020            },
1021            CancelResponse {
1022                schema: schema::REMOTE_CANCEL.into(),
1023                version: 1,
1024                answers: vec![
1025                    CancelItem {
1026                        attempt_id: "x/1".into(),
1027                        answer: RemoteCancelAnswer::Stopped {
1028                            charge: Charge::Observed { units: 1 },
1029                        },
1030                    },
1031                    CancelItem {
1032                        attempt_id: "x/2".into(),
1033                        answer: RemoteCancelAnswer::Unconfirmed,
1034                    },
1035                    CancelItem {
1036                        attempt_id: "x/3".into(),
1037                        answer: RemoteCancelAnswer::UnknownAttempt,
1038                    },
1039                ],
1040            },
1041        )
1042    }
1043
1044    fn exchange() -> ExchangeRecord {
1045        ExchangeRecord {
1046            schema: schema::REMOTE_EXCHANGE.into(),
1047            protocol: PROTOCOL.into(),
1048            binding: ArtifactId::parse(A).unwrap(),
1049            members: vec![ExchangeMember {
1050                attempt_id: "d-1/topic//1".into(),
1051                outcome: MemberOutcome::Failure {
1052                    code: RemoteCode::RateLimited,
1053                    detail: "remote:rate_limited SYNTHETIC".into(),
1054                },
1055                charge: Charge::Observed { units: 0 },
1056            }],
1057            request_digest: Some(A.into()),
1058            response_digest: None,
1059            bytes_sent: true,
1060            status: Some(429),
1061            retry_after: Some("30".into()),
1062            requested_model: None,
1063            served: ServedIdentity::unknown(),
1064            route: vec![],
1065            generation_id: None,
1066            usage: BTreeMap::new(),
1067            reported_cost: Some(ReportedCost {
1068                amount: "0.000125".into(),
1069                currency: "SYNTHETIC".into(),
1070            }),
1071            exchange_charge: Charge::Observed { units: 0 },
1072            attribution: Attribution::Single,
1073            privacy_requested: BTreeMap::new(),
1074            provider_extras: BTreeMap::new(),
1075            late: false,
1076            truncated: false,
1077            phase_expiry: None,
1078            principal: None,
1079            connection: None,
1080        }
1081    }
1082
1083    fn round_trip<T: Document + PartialEq + std::fmt::Debug>(doc: &T) {
1084        let bytes = doc.record_canonical().unwrap();
1085        let back = T::parse(&bytes).unwrap();
1086        assert_eq!(&back, doc);
1087        assert_eq!(back.record_canonical().unwrap(), bytes);
1088    }
1089
1090    #[test]
1091    fn a_principal_reference_is_a_version_2_member_and_evidence_holds_only_its_digest() {
1092        let v1 = infer_request();
1093        assert!(v1.members_fit_version());
1094        let bytes = String::from_utf8(v1.canonical().unwrap()).unwrap();
1095        assert!(!bytes.contains("principal"), "version 1 bytes unchanged");
1096        let p = WirePrincipal::encode(b"SYNTHETIC", "map/1", "routing").unwrap();
1097        assert_eq!(p.reference, "53594e544845544943");
1098        assert_eq!(p.reference_bytes().unwrap(), b"SYNTHETIC");
1099        let mut with = infer_request();
1100        with.principal = Some(p.clone());
1101        assert!(
1102            !with.members_fit_version(),
1103            "a version 1 request carries none"
1104        );
1105        with.version = PROTOCOL_MAJOR_V2;
1106        assert!(with.members_fit_version());
1107        round_trip(&with);
1108        // Bounds and canonical encoding.
1109        assert!(WirePrincipal::encode(b"", "m", "s").is_none());
1110        assert!(WirePrincipal::encode(&[1; MAX_PRINCIPAL_BYTES + 1], "m", "s").is_none());
1111        assert!(WirePrincipal::encode(&[1; MAX_PRINCIPAL_BYTES], "m", "s").is_some());
1112        for bad in ["ABCD", "abc", "zz", ""] {
1113            let q = WirePrincipal {
1114                reference: bad.into(),
1115                ..p.clone()
1116            };
1117            assert!(!q.is_canonical(), "{bad}");
1118        }
1119        // Evidence: a digest that changes with the reference, the mapping
1120        // version and the scope, and never holds the reference.
1121        let e = p.evidence();
1122        assert!(!e.digest.contains(&p.reference));
1123        for other in [
1124            WirePrincipal::encode(b"SYNTHETIC2", "map/1", "routing").unwrap(),
1125            WirePrincipal::encode(b"SYNTHETIC", "map/2", "routing").unwrap(),
1126            WirePrincipal::encode(b"SYNTHETIC", "map/1", "other").unwrap(),
1127        ] {
1128            assert_ne!(other.evidence().digest, e.digest);
1129        }
1130        assert!(!format!("{p:?}").contains(&p.reference));
1131        let mut rec = exchange();
1132        rec.principal = Some(e);
1133        round_trip(&rec);
1134        assert!(rec.is_bounded());
1135    }
1136
1137    #[test]
1138    fn a_connection_observation_is_omitted_without_a_pool_and_round_trips_with_one() {
1139        let plain = String::from_utf8(exchange().canonical().unwrap()).unwrap();
1140        assert!(!plain.contains("connection"));
1141        let mut rec = exchange();
1142        rec.connection = Some(ConnectionObservation {
1143            reuse: ConnectionReuse::Reused,
1144            instance_id: "conn-7".into(),
1145            age_bucket: AgeBucket::of_ms(1_500),
1146            use_count: 2,
1147            multiplex: Multiplex::None,
1148            key_digest: A.into(),
1149            eviction: Some(EvictionReason::MaxUses),
1150        });
1151        round_trip(&rec);
1152        assert_eq!(
1153            rec.connection.as_ref().unwrap().age_bucket,
1154            AgeBucket::Under10s
1155        );
1156        for (ms, b) in [
1157            (0, AgeBucket::Under1s),
1158            (999, AgeBucket::Under1s),
1159            (1_000, AgeBucket::Under10s),
1160            (59_999, AgeBucket::Under60s),
1161            (60_000, AgeBucket::Under600s),
1162            (600_000, AgeBucket::Over600s),
1163        ] {
1164            assert_eq!(AgeBucket::of_ms(ms), b);
1165        }
1166    }
1167
1168    #[test]
1169    fn a_phase_expiry_is_omitted_when_absent_and_round_trips_when_present() {
1170        // Spec 020, 3.4: a record without a phase policy keeps its bytes.
1171        let plain = exchange();
1172        let bytes = String::from_utf8(plain.canonical().unwrap()).unwrap();
1173        assert!(!bytes.contains("phase_expiry"), "{bytes}");
1174        let mut timed = exchange();
1175        timed.phase_expiry = Some(PhaseExpiry {
1176            phase: TransportPhase::FirstResponseByte,
1177            ceiling_ms: 50,
1178            allowance_ms: 40,
1179            elapsed_ms: 40,
1180            limit: PhaseLimit::Budget,
1181            bytes_sent: true,
1182            code: RemoteCode::TransportInterrupted,
1183        });
1184        round_trip(&timed);
1185        let text = String::from_utf8(timed.canonical().unwrap()).unwrap();
1186        assert!(text.contains(r#""phase_expiry":{"allowance_ms":40,"bytes_sent":true,"ceiling_ms":50,"code":"transport_interrupted","elapsed_ms":40,"limit":"budget","phase":"first_response_byte"}"#), "{text}");
1187        assert!(
1188            ExchangeRecord::parse(
1189                text.replace(r#""limit":"budget""#, r#""limit":"budget","extra":1"#)
1190                    .as_bytes()
1191            )
1192            .is_err()
1193        );
1194    }
1195
1196    #[test]
1197    fn remote_documents_round_trip_through_canonical_bytes() {
1198        round_trip(&DescribeRequest {
1199            schema: schema::REMOTE_DESCRIBE.into(),
1200            versions: vec![1],
1201        });
1202        round_trip(&describe_response());
1203        round_trip(&infer_request());
1204        round_trip(&infer_response());
1205        let (req, resp) = cancel_docs();
1206        round_trip(&req);
1207        round_trip(&resp);
1208        round_trip(&exchange());
1209        // Documents without a fractional number have one form.
1210        let r = infer_request();
1211        assert_eq!(r.canonical().unwrap(), r.record_canonical().unwrap());
1212    }
1213
1214    #[test]
1215    fn remote_codes_are_closed_and_named_as_in_the_table() {
1216        for c in RemoteCode::ALL {
1217            let s = serde_json::to_string(&c).unwrap();
1218            assert_eq!(s, format!("\"{}\"", c.as_str()));
1219        }
1220        assert!(serde_json::from_str::<RemoteCode>("\"timeout\"").is_err());
1221    }
1222
1223    fn with_extra_field(doc: &impl Document, at: Option<&str>) -> Vec<u8> {
1224        let mut v: serde_json::Value = serde_json::to_value(doc).unwrap();
1225        let target = match at {
1226            None => &mut v,
1227            Some(p) => v.pointer_mut(p).unwrap(),
1228        };
1229        target
1230            .as_object_mut()
1231            .unwrap()
1232            .insert("zz_unknown".into(), serde_json::json!(1));
1233        serde_json::to_vec(&v).unwrap()
1234    }
1235
1236    fn schema_error<T: Document + std::fmt::Debug>(bytes: &[u8]) {
1237        match T::parse(bytes) {
1238            Err(DocumentError::Parse(ParseError::Schema { .. })) => {}
1239            other => panic!("{other:?}"),
1240        }
1241    }
1242
1243    #[test]
1244    fn remote_documents_refuse_unknown_fields() {
1245        schema_error::<DescribeResponse>(&with_extra_field(&describe_response(), None));
1246        schema_error::<DescribeResponse>(&with_extra_field(
1247            &describe_response(),
1248            Some("/backends/0/terms"),
1249        ));
1250        schema_error::<InferRequest>(&with_extra_field(&infer_request(), Some("/items/0")));
1251        schema_error::<InferResponse>(&with_extra_field(&infer_response(), Some("/served")));
1252        schema_error::<InferResponse>(&with_extra_field(
1253            &infer_response(),
1254            Some("/items/1/result/failure"),
1255        ));
1256        let (req, resp) = cancel_docs();
1257        schema_error::<CancelRequest>(&with_extra_field(&req, None));
1258        schema_error::<CancelResponse>(&with_extra_field(&resp, Some("/answers/0")));
1259        schema_error::<ExchangeRecord>(&with_extra_field(&exchange(), Some("/members/0")));
1260    }
1261
1262    #[test]
1263    fn remote_documents_refuse_a_wrong_schema_string() {
1264        let mut r = infer_request();
1265        r.schema = "rustev.remote-infer/2".into();
1266        let bytes = r.canonical().unwrap();
1267        assert!(matches!(
1268            InferRequest::parse(&bytes),
1269            Err(DocumentError::Schema { .. })
1270        ));
1271        // A cancel request is not an infer request, whatever its shape.
1272        let (c, _) = cancel_docs();
1273        assert!(InferRequest::parse(&c.canonical().unwrap()).is_err());
1274        let mut d = describe_response();
1275        d.schema = schema::REMOTE_INFER.into();
1276        assert!(matches!(
1277            DescribeResponse::parse(&d.canonical().unwrap()),
1278            Err(DocumentError::Schema { .. })
1279        ));
1280    }
1281
1282    #[test]
1283    fn remote_documents_enforce_their_parse_limits() {
1284        // Oversized: refused before reading content.
1285        let big = vec![b' '; limits::REMOTE_V1.max_bytes + 1];
1286        assert!(matches!(
1287            InferRequest::parse(&big),
1288            Err(DocumentError::Parse(ParseError::Bound(
1289                BoundError::TooLarge { .. }
1290            )))
1291        ));
1292        // Too deep.
1293        let deep = format!(
1294            "{}{}",
1295            "[".repeat(limits::REMOTE_V1.max_depth + 1),
1296            "]".repeat(limits::REMOTE_V1.max_depth + 1)
1297        );
1298        assert!(matches!(
1299            InferResponse::parse(deep.as_bytes()),
1300            Err(DocumentError::Parse(ParseError::Bound(
1301                BoundError::TooDeep { .. }
1302            )))
1303        ));
1304        // Too many items in one collection.
1305        let mut r = infer_request();
1306        r.items = (0..limits::REMOTE_V1.max_collection_len + 1)
1307            .map(|i| InferItem {
1308                attempt_id: format!("a/{i}"),
1309                projection: "{}".into(),
1310            })
1311            .collect();
1312        assert!(matches!(
1313            InferRequest::parse(&r.canonical().unwrap()),
1314            Err(DocumentError::Parse(ParseError::Bound(
1315                BoundError::CollectionTooLong { .. }
1316            )))
1317        ));
1318        // An exchange record parses under its own, smaller limits.
1319        let mut e = exchange();
1320        e.protocol = "p".repeat(limits::EXCHANGE_V1.max_string_bytes + 1);
1321        assert!(matches!(
1322            ExchangeRecord::parse(&e.canonical().unwrap()),
1323            Err(DocumentError::Parse(ParseError::Bound(
1324                BoundError::StringTooLong { .. }
1325            )))
1326        ));
1327    }
1328
1329    #[test]
1330    fn remote_exchange_records_are_cut_to_a_fixed_maximum_size() {
1331        // The worst case: every string at the cut and every byte escaped six
1332        // times over, every collection at its cap.
1333        let nasty = "\u{1}".repeat(4 * MAX_TEXT_BYTES);
1334        let s = || nasty.clone();
1335        let map = |n: usize| -> BTreeMap<String, String> {
1336            (0..n).map(|i| (format!("{i:03}{nasty}"), s())).collect()
1337        };
1338        let mut e = exchange();
1339        e.protocol = s();
1340        e.members = (0..MAX_ITEMS + 5)
1341            .map(|i| ExchangeMember {
1342                attempt_id: format!("{i:03}{nasty}"),
1343                outcome: MemberOutcome::Failure {
1344                    code: RemoteCode::TransportInterrupted,
1345                    detail: s(),
1346                },
1347                charge: Charge::Observed { units: u64::MAX },
1348            })
1349            .collect();
1350        e.request_digest = Some(s());
1351        e.response_digest = Some(s());
1352        e.retry_after = Some(s());
1353        e.requested_model = Some(s());
1354        e.served.identity = Some(s());
1355        e.route = vec![s(); MAX_ROUTE + 3];
1356        e.generation_id = Some(s());
1357        e.usage = (0..MAX_ENTRIES + 3)
1358            .map(|i| (format!("{i:03}{nasty}"), u64::MAX))
1359            .collect();
1360        e.reported_cost = Some(ReportedCost {
1361            amount: s(),
1362            currency: s(),
1363        });
1364        e.privacy_requested = map(MAX_ENTRIES + 3);
1365        e.provider_extras = map(MAX_ENTRIES + 3);
1366        assert!(!e.is_bounded());
1367        let b = e.bounded();
1368        assert!(b.truncated && b.is_bounded());
1369        assert_eq!(b.members.len(), MAX_ITEMS);
1370        assert!(
1371            b.members
1372                .iter()
1373                .all(|m| m.attempt_id.len() <= MAX_TEXT_BYTES)
1374        );
1375        let bytes = b.canonical().unwrap();
1376        assert!(
1377            bytes.len() <= limits::EXCHANGE_V1.max_bytes,
1378            "{} bytes",
1379            bytes.len()
1380        );
1381        assert_eq!(ExchangeRecord::parse_checked(&bytes).unwrap(), b);
1382        // A record beyond its bounds is refused even within the byte limit.
1383        let mut over = exchange();
1384        over.generation_id = Some("g".repeat(MAX_TEXT_BYTES + 1));
1385        assert!(ExchangeRecord::parse_checked(&over.canonical().unwrap()).is_err());
1386    }
1387
1388    #[test]
1389    fn remote_text_is_cut_on_a_character_boundary() {
1390        let s = "é".repeat(200);
1391        let c = cut_text(&s);
1392        assert!(c.len() <= MAX_TEXT_BYTES && c.len() >= MAX_TEXT_BYTES - 1);
1393        assert!(s.starts_with(&c));
1394        assert_eq!(cut_text("short"), "short");
1395    }
1396}
rustev_contract/remote.rs

rustev_contract/
remote.rs
