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            schema: schema::BACKEND.into(),
911            backend_id: "synthetic-remote".into(),
912            artifact: ArtifactId::parse(A).unwrap(),
913            operations: vec![OperationSupport {
914                isolation: None,
915                screen: None,
916                operation: Operation::Classify,
917                output: OutputKind::Distribution,
918                max_options: 8,
919            }],
920            input_limit: InputLimit {
921                max_bytes: 4096,
922                on_excess: InputExcess::Refuse,
923            },
924            determinism: Determinism::Unspecified,
925        }
926    }
927
928    fn terms() -> RemoteTerms {
929        RemoteTerms {
930            cost: CostBound::Bounded { max_units: 3 },
931            refusals_charged: false,
932            cancellation: CancelSupport::Confirmed,
933            served: ServedTerms {
934                disclosure: Disclosure::Pinned,
935                identity: Some("synthetic-model@1".into()),
936            },
937            batching: Batching::SharedState { max_items: 4 },
938            max_request_bytes: 65_536,
939            max_response_bytes: 65_536,
940            idempotency_window_ms: 60_000,
941            privacy: vec!["no_retention".into()],
942        }
943    }
944
945    fn describe_response() -> DescribeResponse {
946        let d = descriptor();
947        DescribeResponse {
948            schema: schema::REMOTE_DESCRIBE.into(),
949            version: 1,
950            backends: vec![HostedBackend {
951                descriptor_id: d.id().unwrap(),
952                descriptor: d,
953                terms: terms(),
954            }],
955        }
956    }
957
958    fn infer_request() -> InferRequest {
959        InferRequest {
960            schema: schema::REMOTE_INFER.into(),
961            version: 1,
962            backend_id: "synthetic-remote".into(),
963            descriptor: descriptor().id().unwrap(),
964            artifact: ArtifactId::parse(A).unwrap(),
965            budget_ms: 900,
966            trace_id: "d-1".into(),
967            privacy: BTreeMap::from([("no_retention".into(), "true".into())]),
968            items: vec![InferItem {
969                attempt_id: "d-1/topic//1".into(),
970                projection: r#"{"operation":"classify"}"#.into(),
971            }],
972            principal: None,
973        }
974    }
975
976    fn infer_response() -> InferResponse {
977        InferResponse {
978            schema: schema::REMOTE_INFER.into(),
979            version: 1,
980            artifact: ArtifactId::parse(A).unwrap(),
981            served: ServedIdentity {
982                identity: Some("synthetic-model@1".into()),
983                disclosure: Disclosure::Pinned,
984            },
985            route: vec!["synthetic-route".into()],
986            generation_id: Some("gen-1".into()),
987            items: vec![
988                ItemAnswer {
989                    attempt_id: "d-1/topic//1".into(),
990                    result: ItemResult::Output(RawOutput::Distribution(BTreeMap::from([
991                        ("a".into(), 0.1),
992                        ("b".into(), 0.9),
993                    ]))),
994                },
995                ItemAnswer {
996                    attempt_id: "d-1/topic//2".into(),
997                    result: ItemResult::Failure {
998                        code: RemoteCode::RemoteDeadline,
999                        detail: "SYNTHETIC".into(),
1000                    },
1001                },
1002                ItemAnswer {
1003                    attempt_id: "d-1/topic//3".into(),
1004                    result: ItemResult::InProgress,
1005                },
1006            ],
1007            charge: Charge::Observed { units: 3 },
1008            extras: BTreeMap::from([("argmax".into(), "b".into())]),
1009        }
1010    }
1011
1012    fn cancel_docs() -> (CancelRequest, CancelResponse) {
1013        (
1014            CancelRequest {
1015                schema: schema::REMOTE_CANCEL.into(),
1016                version: 1,
1017                backend_id: "synthetic-remote".into(),
1018                attempt_ids: vec!["x/1".into(), "x/2".into(), "x/3".into()],
1019            },
1020            CancelResponse {
1021                schema: schema::REMOTE_CANCEL.into(),
1022                version: 1,
1023                answers: vec![
1024                    CancelItem {
1025                        attempt_id: "x/1".into(),
1026                        answer: RemoteCancelAnswer::Stopped {
1027                            charge: Charge::Observed { units: 1 },
1028                        },
1029                    },
1030                    CancelItem {
1031                        attempt_id: "x/2".into(),
1032                        answer: RemoteCancelAnswer::Unconfirmed,
1033                    },
1034                    CancelItem {
1035                        attempt_id: "x/3".into(),
1036                        answer: RemoteCancelAnswer::UnknownAttempt,
1037                    },
1038                ],
1039            },
1040        )
1041    }
1042
1043    fn exchange() -> ExchangeRecord {
1044        ExchangeRecord {
1045            schema: schema::REMOTE_EXCHANGE.into(),
1046            protocol: PROTOCOL.into(),
1047            binding: ArtifactId::parse(A).unwrap(),
1048            members: vec![ExchangeMember {
1049                attempt_id: "d-1/topic//1".into(),
1050                outcome: MemberOutcome::Failure {
1051                    code: RemoteCode::RateLimited,
1052                    detail: "remote:rate_limited SYNTHETIC".into(),
1053                },
1054                charge: Charge::Observed { units: 0 },
1055            }],
1056            request_digest: Some(A.into()),
1057            response_digest: None,
1058            bytes_sent: true,
1059            status: Some(429),
1060            retry_after: Some("30".into()),
1061            requested_model: None,
1062            served: ServedIdentity::unknown(),
1063            route: vec![],
1064            generation_id: None,
1065            usage: BTreeMap::new(),
1066            reported_cost: Some(ReportedCost {
1067                amount: "0.000125".into(),
1068                currency: "SYNTHETIC".into(),
1069            }),
1070            exchange_charge: Charge::Observed { units: 0 },
1071            attribution: Attribution::Single,
1072            privacy_requested: BTreeMap::new(),
1073            provider_extras: BTreeMap::new(),
1074            late: false,
1075            truncated: false,
1076            phase_expiry: None,
1077            principal: None,
1078            connection: None,
1079        }
1080    }
1081
1082    fn round_trip<T: Document + PartialEq + std::fmt::Debug>(doc: &T) {
1083        let bytes = doc.record_canonical().unwrap();
1084        let back = T::parse(&bytes).unwrap();
1085        assert_eq!(&back, doc);
1086        assert_eq!(back.record_canonical().unwrap(), bytes);
1087    }
1088
1089    #[test]
1090    fn a_principal_reference_is_a_version_2_member_and_evidence_holds_only_its_digest() {
1091        let v1 = infer_request();
1092        assert!(v1.members_fit_version());
1093        let bytes = String::from_utf8(v1.canonical().unwrap()).unwrap();
1094        assert!(!bytes.contains("principal"), "version 1 bytes unchanged");
1095        let p = WirePrincipal::encode(b"SYNTHETIC", "map/1", "routing").unwrap();
1096        assert_eq!(p.reference, "53594e544845544943");
1097        assert_eq!(p.reference_bytes().unwrap(), b"SYNTHETIC");
1098        let mut with = infer_request();
1099        with.principal = Some(p.clone());
1100        assert!(
1101            !with.members_fit_version(),
1102            "a version 1 request carries none"
1103        );
1104        with.version = PROTOCOL_MAJOR_V2;
1105        assert!(with.members_fit_version());
1106        round_trip(&with);
1107        // Bounds and canonical encoding.
1108        assert!(WirePrincipal::encode(b"", "m", "s").is_none());
1109        assert!(WirePrincipal::encode(&[1; MAX_PRINCIPAL_BYTES + 1], "m", "s").is_none());
1110        assert!(WirePrincipal::encode(&[1; MAX_PRINCIPAL_BYTES], "m", "s").is_some());
1111        for bad in ["ABCD", "abc", "zz", ""] {
1112            let q = WirePrincipal {
1113                reference: bad.into(),
1114                ..p.clone()
1115            };
1116            assert!(!q.is_canonical(), "{bad}");
1117        }
1118        // Evidence: a digest that changes with the reference, the mapping
1119        // version and the scope, and never holds the reference.
1120        let e = p.evidence();
1121        assert!(!e.digest.contains(&p.reference));
1122        for other in [
1123            WirePrincipal::encode(b"SYNTHETIC2", "map/1", "routing").unwrap(),
1124            WirePrincipal::encode(b"SYNTHETIC", "map/2", "routing").unwrap(),
1125            WirePrincipal::encode(b"SYNTHETIC", "map/1", "other").unwrap(),
1126        ] {
1127            assert_ne!(other.evidence().digest, e.digest);
1128        }
1129        assert!(!format!("{p:?}").contains(&p.reference));
1130        let mut rec = exchange();
1131        rec.principal = Some(e);
1132        round_trip(&rec);
1133        assert!(rec.is_bounded());
1134    }
1135
1136    #[test]
1137    fn a_connection_observation_is_omitted_without_a_pool_and_round_trips_with_one() {
1138        let plain = String::from_utf8(exchange().canonical().unwrap()).unwrap();
1139        assert!(!plain.contains("connection"));
1140        let mut rec = exchange();
1141        rec.connection = Some(ConnectionObservation {
1142            reuse: ConnectionReuse::Reused,
1143            instance_id: "conn-7".into(),
1144            age_bucket: AgeBucket::of_ms(1_500),
1145            use_count: 2,
1146            multiplex: Multiplex::None,
1147            key_digest: A.into(),
1148            eviction: Some(EvictionReason::MaxUses),
1149        });
1150        round_trip(&rec);
1151        assert_eq!(
1152            rec.connection.as_ref().unwrap().age_bucket,
1153            AgeBucket::Under10s
1154        );
1155        for (ms, b) in [
1156            (0, AgeBucket::Under1s),
1157            (999, AgeBucket::Under1s),
1158            (1_000, AgeBucket::Under10s),
1159            (59_999, AgeBucket::Under60s),
1160            (60_000, AgeBucket::Under600s),
1161            (600_000, AgeBucket::Over600s),
1162        ] {
1163            assert_eq!(AgeBucket::of_ms(ms), b);
1164        }
1165    }
1166
1167    #[test]
1168    fn a_phase_expiry_is_omitted_when_absent_and_round_trips_when_present() {
1169        // Spec 020, 3.4: a record without a phase policy keeps its bytes.
1170        let plain = exchange();
1171        let bytes = String::from_utf8(plain.canonical().unwrap()).unwrap();
1172        assert!(!bytes.contains("phase_expiry"), "{bytes}");
1173        let mut timed = exchange();
1174        timed.phase_expiry = Some(PhaseExpiry {
1175            phase: TransportPhase::FirstResponseByte,
1176            ceiling_ms: 50,
1177            allowance_ms: 40,
1178            elapsed_ms: 40,
1179            limit: PhaseLimit::Budget,
1180            bytes_sent: true,
1181            code: RemoteCode::TransportInterrupted,
1182        });
1183        round_trip(&timed);
1184        let text = String::from_utf8(timed.canonical().unwrap()).unwrap();
1185        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}");
1186        assert!(
1187            ExchangeRecord::parse(
1188                text.replace(r#""limit":"budget""#, r#""limit":"budget","extra":1"#)
1189                    .as_bytes()
1190            )
1191            .is_err()
1192        );
1193    }
1194
1195    #[test]
1196    fn remote_documents_round_trip_through_canonical_bytes() {
1197        round_trip(&DescribeRequest {
1198            schema: schema::REMOTE_DESCRIBE.into(),
1199            versions: vec![1],
1200        });
1201        round_trip(&describe_response());
1202        round_trip(&infer_request());
1203        round_trip(&infer_response());
1204        let (req, resp) = cancel_docs();
1205        round_trip(&req);
1206        round_trip(&resp);
1207        round_trip(&exchange());
1208        // Documents without a fractional number have one form.
1209        let r = infer_request();
1210        assert_eq!(r.canonical().unwrap(), r.record_canonical().unwrap());
1211    }
1212
1213    #[test]
1214    fn remote_codes_are_closed_and_named_as_in_the_table() {
1215        for c in RemoteCode::ALL {
1216            let s = serde_json::to_string(&c).unwrap();
1217            assert_eq!(s, format!("\"{}\"", c.as_str()));
1218        }
1219        assert!(serde_json::from_str::<RemoteCode>("\"timeout\"").is_err());
1220    }
1221
1222    fn with_extra_field(doc: &impl Document, at: Option<&str>) -> Vec<u8> {
1223        let mut v: serde_json::Value = serde_json::to_value(doc).unwrap();
1224        let target = match at {
1225            None => &mut v,
1226            Some(p) => v.pointer_mut(p).unwrap(),
1227        };
1228        target
1229            .as_object_mut()
1230            .unwrap()
1231            .insert("zz_unknown".into(), serde_json::json!(1));
1232        serde_json::to_vec(&v).unwrap()
1233    }
1234
1235    fn schema_error<T: Document + std::fmt::Debug>(bytes: &[u8]) {
1236        match T::parse(bytes) {
1237            Err(DocumentError::Parse(ParseError::Schema { .. })) => {}
1238            other => panic!("{other:?}"),
1239        }
1240    }
1241
1242    #[test]
1243    fn remote_documents_refuse_unknown_fields() {
1244        schema_error::<DescribeResponse>(&with_extra_field(&describe_response(), None));
1245        schema_error::<DescribeResponse>(&with_extra_field(
1246            &describe_response(),
1247            Some("/backends/0/terms"),
1248        ));
1249        schema_error::<InferRequest>(&with_extra_field(&infer_request(), Some("/items/0")));
1250        schema_error::<InferResponse>(&with_extra_field(&infer_response(), Some("/served")));
1251        schema_error::<InferResponse>(&with_extra_field(
1252            &infer_response(),
1253            Some("/items/1/result/failure"),
1254        ));
1255        let (req, resp) = cancel_docs();
1256        schema_error::<CancelRequest>(&with_extra_field(&req, None));
1257        schema_error::<CancelResponse>(&with_extra_field(&resp, Some("/answers/0")));
1258        schema_error::<ExchangeRecord>(&with_extra_field(&exchange(), Some("/members/0")));
1259    }
1260
1261    #[test]
1262    fn remote_documents_refuse_a_wrong_schema_string() {
1263        let mut r = infer_request();
1264        r.schema = "rustev.remote-infer/2".into();
1265        let bytes = r.canonical().unwrap();
1266        assert!(matches!(
1267            InferRequest::parse(&bytes),
1268            Err(DocumentError::Schema { .. })
1269        ));
1270        // A cancel request is not an infer request, whatever its shape.
1271        let (c, _) = cancel_docs();
1272        assert!(InferRequest::parse(&c.canonical().unwrap()).is_err());
1273        let mut d = describe_response();
1274        d.schema = schema::REMOTE_INFER.into();
1275        assert!(matches!(
1276            DescribeResponse::parse(&d.canonical().unwrap()),
1277            Err(DocumentError::Schema { .. })
1278        ));
1279    }
1280
1281    #[test]
1282    fn remote_documents_enforce_their_parse_limits() {
1283        // Oversized: refused before reading content.
1284        let big = vec![b' '; limits::REMOTE_V1.max_bytes + 1];
1285        assert!(matches!(
1286            InferRequest::parse(&big),
1287            Err(DocumentError::Parse(ParseError::Bound(
1288                BoundError::TooLarge { .. }
1289            )))
1290        ));
1291        // Too deep.
1292        let deep = format!(
1293            "{}{}",
1294            "[".repeat(limits::REMOTE_V1.max_depth + 1),
1295            "]".repeat(limits::REMOTE_V1.max_depth + 1)
1296        );
1297        assert!(matches!(
1298            InferResponse::parse(deep.as_bytes()),
1299            Err(DocumentError::Parse(ParseError::Bound(
1300                BoundError::TooDeep { .. }
1301            )))
1302        ));
1303        // Too many items in one collection.
1304        let mut r = infer_request();
1305        r.items = (0..limits::REMOTE_V1.max_collection_len + 1)
1306            .map(|i| InferItem {
1307                attempt_id: format!("a/{i}"),
1308                projection: "{}".into(),
1309            })
1310            .collect();
1311        assert!(matches!(
1312            InferRequest::parse(&r.canonical().unwrap()),
1313            Err(DocumentError::Parse(ParseError::Bound(
1314                BoundError::CollectionTooLong { .. }
1315            )))
1316        ));
1317        // An exchange record parses under its own, smaller limits.
1318        let mut e = exchange();
1319        e.protocol = "p".repeat(limits::EXCHANGE_V1.max_string_bytes + 1);
1320        assert!(matches!(
1321            ExchangeRecord::parse(&e.canonical().unwrap()),
1322            Err(DocumentError::Parse(ParseError::Bound(
1323                BoundError::StringTooLong { .. }
1324            )))
1325        ));
1326    }
1327
1328    #[test]
1329    fn remote_exchange_records_are_cut_to_a_fixed_maximum_size() {
1330        // The worst case: every string at the cut and every byte escaped six
1331        // times over, every collection at its cap.
1332        let nasty = "\u{1}".repeat(4 * MAX_TEXT_BYTES);
1333        let s = || nasty.clone();
1334        let map = |n: usize| -> BTreeMap<String, String> {
1335            (0..n).map(|i| (format!("{i:03}{nasty}"), s())).collect()
1336        };
1337        let mut e = exchange();
1338        e.protocol = s();
1339        e.members = (0..MAX_ITEMS + 5)
1340            .map(|i| ExchangeMember {
1341                attempt_id: format!("{i:03}{nasty}"),
1342                outcome: MemberOutcome::Failure {
1343                    code: RemoteCode::TransportInterrupted,
1344                    detail: s(),
1345                },
1346                charge: Charge::Observed { units: u64::MAX },
1347            })
1348            .collect();
1349        e.request_digest = Some(s());
1350        e.response_digest = Some(s());
1351        e.retry_after = Some(s());
1352        e.requested_model = Some(s());
1353        e.served.identity = Some(s());
1354        e.route = vec![s(); MAX_ROUTE + 3];
1355        e.generation_id = Some(s());
1356        e.usage = (0..MAX_ENTRIES + 3)
1357            .map(|i| (format!("{i:03}{nasty}"), u64::MAX))
1358            .collect();
1359        e.reported_cost = Some(ReportedCost {
1360            amount: s(),
1361            currency: s(),
1362        });
1363        e.privacy_requested = map(MAX_ENTRIES + 3);
1364        e.provider_extras = map(MAX_ENTRIES + 3);
1365        assert!(!e.is_bounded());
1366        let b = e.bounded();
1367        assert!(b.truncated && b.is_bounded());
1368        assert_eq!(b.members.len(), MAX_ITEMS);
1369        assert!(
1370            b.members
1371                .iter()
1372                .all(|m| m.attempt_id.len() <= MAX_TEXT_BYTES)
1373        );
1374        let bytes = b.canonical().unwrap();
1375        assert!(
1376            bytes.len() <= limits::EXCHANGE_V1.max_bytes,
1377            "{} bytes",
1378            bytes.len()
1379        );
1380        assert_eq!(ExchangeRecord::parse_checked(&bytes).unwrap(), b);
1381        // A record beyond its bounds is refused even within the byte limit.
1382        let mut over = exchange();
1383        over.generation_id = Some("g".repeat(MAX_TEXT_BYTES + 1));
1384        assert!(ExchangeRecord::parse_checked(&over.canonical().unwrap()).is_err());
1385    }
1386
1387    #[test]
1388    fn remote_text_is_cut_on_a_character_boundary() {
1389        let s = "é".repeat(200);
1390        let c = cut_text(&s);
1391        assert!(c.len() <= MAX_TEXT_BYTES && c.len() >= MAX_TEXT_BYTES - 1);
1392        assert!(s.starts_with(&c));
1393        assert_eq!(cut_text("short"), "short");
1394    }
1395}
rustev_contract/remote.rs

rustev_contract/
remote.rs
