rig_core/observe/
adapter.rs

1//! Execution-local correlation and facts from provider request boundaries.
2//!
3//! ```
4//! use rig_core::observe::scrub_diagnostic;
5//!
6//! assert_eq!(scrub_diagnostic("Bearer secret", &[]), "[redacted]");
7//! ```
8
9use std::sync::{Arc, Mutex};
10
11use serde::{Deserialize, Serialize};
12
13use super::{Action, Emitter, Observation, Stage, Subject, Witness};
14
15/// Bounds diagnostic text and removes control characters. Known credentials
16/// or credential markers redact the whole message; oversized messages are
17/// replaced rather than truncated to a potentially sensitive prefix.
18pub fn scrub_diagnostic(value: &str, secrets: &[String]) -> String {
19    scrub::text(value, secrets)
20}
21
22/// The secrets a URL carries for diagnostic redaction: userinfo and known
23/// credential query parameters, origin-form request URIs included. The
24/// returned values are secrets: keep them runtime-only and never write
25/// them into a diagnostic or an artifact.
26pub fn diagnostic_url_secrets(url: &str) -> Vec<String> {
27    scrub::url_secrets(url)
28}
29
30/// One fact about a provider operation, independently of its effect record.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32pub struct AdapterObservation {
33    /// Caller-owned logical operation identity within this execution.
34    pub operation: String,
35    /// One-based HTTP send ordinal; absent only when correlation is exhausted.
36    pub attempt: Option<u64>,
37    /// One-based host dispatch attempt, when explicitly supplied by the host.
38    /// Independent of the HTTP send ordinal: a dispatch may send more than once.
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub host_attempt: Option<std::num::NonZeroU64>,
41    /// What the owning boundary observed.
42    pub event: AdapterEvent,
43    /// Scrubbed volatile diagnostics, excluded from semantic comparison.
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub analysis: Option<AdapterAnalysis>,
46}
47
48/// Metadata from the provider boundary. Request/response bodies are not included.
49#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(tag = "event", rename_all = "snake_case")]
51pub enum AdapterEvent {
52    /// Fields actually present in a provider payload; absent fields are not updates.
53    Provider {
54        /// The provider's verdict and selected model, independent of transport closure.
55        verdict: AdapterVerdict,
56    },
57    /// A provider error envelope, including envelopes carried under HTTP 200.
58    ErrorEnvelope {
59        /// Scrubbed original envelope fields, not an inferred HTTP status.
60        error: AdapterErrorEnvelope,
61    },
62    /// The attempt's normalized usage, observed once when its reply ends.
63    Usage {
64        /// The counters of the response's usage.
65        usage: AdapterUsage,
66    },
67    /// A request is about to be sent.
68    Started {
69        /// HTTP method.
70        method: String,
71        /// Provider-declared route template, never a credential-bearing URI.
72        route: String,
73    },
74    /// The transport returned response headers.
75    Response {
76        /// Actual HTTP response status, including successful responses.
77        status: u16,
78    },
79    /// The body reached EOF, independently of any provider terminal verdict.
80    TransportEof {
81        /// Number of complete frames, excluding provider-recognized analysis-only frames.
82        after: usize,
83        /// Undelimited SSE bytes remaining at EOF; zero means no partial frame.
84        partial_bytes: usize,
85    },
86    /// The owning request boundary closed this attempt.
87    Finished {
88        /// Whether the attempt completed, failed, or was dropped.
89        ending: AdapterEnding,
90    },
91    /// A complete transport frame could not be decoded; the driver may continue.
92    Corrupt {
93        /// One-based ordinal counting known, unknown and corrupt frames,
94        /// excluding provider-recognized analysis-only frames.
95        frame: usize,
96    },
97    /// The context cannot assign another unique send ordinal.
98    IdentityExhausted,
99}
100
101/// Sparse provider metadata: a present field supersedes its previous value.
102#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
103pub struct AdapterVerdict {
104    /// Provider finish code, including unknown codes after scrubbing.
105    pub finish_reason: Option<String>,
106    /// Provider prompt-block/refusal reason, when reported.
107    pub block_reason: Option<String>,
108    /// Provider detail accompanying a finish/block reason.
109    pub detail: Option<String>,
110    /// Actual provider model/version, not the requested alias.
111    pub model: Option<String>,
112}
113
114/// Analysis-only response metadata. Never use identifiers as operation join keys.
115#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
116pub struct AdapterAnalysis {
117    /// Provider response identifier, scrubbed and bounded.
118    pub response_id: Option<String>,
119    /// Allowlisted response headers. None means the transport supplied no map;
120    /// an empty map means it supplied no allowlisted values.
121    pub headers: Option<std::collections::BTreeMap<String, String>>,
122}
123
124/// Original provider error-envelope fields, separate from Rig retry classification.
125#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
126pub struct AdapterErrorEnvelope {
127    /// Provider code in its scalar wire spelling; unknown codes stay observable.
128    pub code: Option<String>,
129    /// Provider status name, when present.
130    pub status: Option<String>,
131    /// Bounded, scrubbed provider message, when present.
132    pub message: Option<String>,
133}
134
135/// Error-envelope projection with optional code, message, and `type`/`status`.
136/// Codes remain JSON values until emission validates their scalar shape.
137#[derive(Default)]
138pub struct ObservedError {
139    pub code: Option<serde_json::Value>,
140    pub kind: Option<String>,
141    pub message: Option<String>,
142}
143
144impl ObservedError {
145    /// The error envelope a provider document carries, located as its
146    /// [`crate::provider_response::ProviderResponseError`] reads it: an
147    /// error event's own `type` is its tag, not a kind, and an `error`
148    /// string is the message alone.
149    pub fn of(doc: &serde_json::Value) -> Option<Self> {
150        let (error, nested) = crate::provider_response::located_error(doc)?;
151        let text = |key: &str| error.get(key).and_then(serde_json::Value::as_str);
152        let kind = nested.then(|| text("type").or_else(|| text("status")));
153        Some(Self {
154            code: error.get("code").filter(|code| !code.is_null()).cloned(),
155            kind: kind.flatten().map(str::to_owned),
156            message: error
157                .as_str()
158                .or_else(|| text("message"))
159                .map(str::to_owned),
160        })
161    }
162
163    /// Emit this envelope as the attempt's error-envelope fact, scrubbed.
164    pub fn emit(self, sink: &mut ObservationSink<'_>) {
165        let code = self.code.map(|code| match code {
166            serde_json::Value::String(code) => sink.scrub(&code),
167            serde_json::Value::Number(code) => code.to_string(),
168            _ => "[invalid]".to_owned(),
169        });
170        sink.error(AdapterErrorEnvelope {
171            code,
172            status: self.kind.map(|value| sink.scrub(&value)),
173            message: self.message.map(|value| sink.scrub(&value)),
174        });
175    }
176}
177
178/// The counters of the [`Usage`](crate::completion::Usage) the decoder
179/// normalized into the reply's end, observed once when the reply ends; an
180/// attempt that fails before its end observes none. Unreported is `None`.
181#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
182pub struct AdapterUsage {
183    /// Every input token: uncached, cached, cache writes and hosted-tool prompt.
184    pub input_tokens: Option<u64>,
185    /// Every output token, reasoning included.
186    pub output_tokens: Option<u64>,
187    /// Always `None`; kept for serde compatibility. Read the response's
188    /// [`Usage::total_tokens`](crate::completion::Usage::total_tokens).
189    pub total_tokens: Option<u64>,
190    /// The part of the input read from a provider-managed cache.
191    pub cached_input_tokens: Option<u64>,
192    /// The part of the output spent on internal reasoning.
193    pub reasoning_tokens: Option<u64>,
194    /// The part of the input a provider's hosted tools added to the prompt.
195    pub tool_input_tokens: Option<u64>,
196}
197
198impl From<&crate::completion::Usage> for AdapterUsage {
199    fn from(usage: &crate::completion::Usage) -> Self {
200        // Exhaustive: a new `Usage` counter must be mapped or ignored here.
201        let crate::completion::Usage {
202            input_tokens,
203            output_tokens,
204            total_tokens: _,
205            cached_input_tokens,
206            cache_creation_input_tokens: _,
207            tool_use_prompt_tokens,
208            reasoning_tokens,
209            cost: _,
210        } = *usage;
211        Self {
212            input_tokens,
213            output_tokens,
214            total_tokens: None,
215            cached_input_tokens,
216            reasoning_tokens,
217            tool_input_tokens: tool_use_prompt_tokens,
218        }
219    }
220}
221
222/// Boundary identified by the adapter's original typed error, not its message.
223#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
224#[serde(rename_all = "snake_case")]
225pub enum AdapterErrorBoundary {
226    /// Request construction or validation failed.
227    Request,
228    /// The provider returned a failure response or error envelope.
229    ProviderResponse,
230    /// Response decoding or validation failed.
231    Decode,
232    /// A typed transport termination was reported.
233    Transport,
234    /// An erased client error does not establish a more specific boundary.
235    #[default]
236    Unknown,
237}
238
239impl AdapterErrorBoundary {
240    pub(crate) fn from_http(error: &crate::http_client::Error) -> Self {
241        use crate::http_client::Error as H;
242        match error {
243            H::Protocol(_) | H::InvalidHeaderValue(_) | H::NoHeaders | H::Unbuilt(_) => {
244                Self::Request
245            }
246            H::InvalidContentType(_) => Self::Decode,
247            H::StreamEnded => Self::Transport,
248            H::InvalidStatusCodeWithDetails { .. } => Self::ProviderResponse,
249            _ => Self::Unknown,
250        }
251    }
252}
253
254/// How a provider attempt closed, distinct from the run's ending.
255#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
256#[serde(tag = "ending", rename_all = "snake_case")]
257pub enum AdapterEnding {
258    /// The request produced a decoded response.
259    Decoded,
260    /// The request failed; provider payloads remain in the existing error report.
261    Error {
262        /// Known boundary before error conversion erases transport subtypes.
263        #[serde(default)]
264        boundary: AdapterErrorBoundary,
265        /// Stable Rig error classification.
266        kind: String,
267        /// HTTP status reported by the failure, when available.
268        status: Option<u16>,
269        /// Retryability under the existing policy, not a decision to retry.
270        retryable: bool,
271    },
272    /// A provider terminal record was decoded from the stream.
273    Terminal,
274    /// Transport EOF without a provider terminal record.
275    Eof {
276        /// Number of complete frames, excluding provider-recognized analysis-only frames.
277        after: usize,
278    },
279    /// EOF left an undelimited SSE event; bytes are never retained in the fact.
280    PartialFrame {
281        /// Raw bytes since the last blank event delimiter, saturated at usize::MAX.
282        byte_count: usize,
283        /// Complete frames previously delivered, excluding analysis-only frames.
284        after: usize,
285    },
286    /// The future or stream was dropped before its boundary closed.
287    Dropped,
288}
289
290/// A caller-supplied witness and logical operation identity.
291///
292/// Clone this context for retries of the same operation. Create a new context
293/// with a distinct identity for a different call, including identical parallel
294/// requests. Identity is execution-local; it does not assert replay equivalence.
295/// This handle is runtime-only and must never be serialized into provider data.
296#[derive(Clone)]
297pub struct AdapterContext {
298    inner: Arc<AdapterContextInner>,
299}
300
301struct AdapterContextInner {
302    sink: Arc<dyn Witness + Send + Sync>,
303    subject: Subject,
304    operation: String,
305    next: Arc<Mutex<Option<u64>>>,
306    host_attempt: Option<std::num::NonZeroU64>,
307}
308
309impl std::fmt::Debug for AdapterContext {
310    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
311        // Caller-supplied identity can contain sensitive data; do not log it.
312        f.debug_struct("AdapterContext").finish_non_exhaustive()
313    }
314}
315
316impl AdapterContext {
317    /// Caller-owned identity of this logical operation within the execution.
318    pub fn operation(&self) -> &str {
319        &self.inner.operation
320    }
321
322    /// Bind a logical operation to its witness and existing subject.
323    /// The caller must use a non-sensitive, execution-unique operation identity.
324    pub fn new(
325        sink: Arc<dyn Witness + Send + Sync>,
326        subject: Subject,
327        operation: impl Into<String>,
328    ) -> Self {
329        Self {
330            inner: Arc::new(AdapterContextInner {
331                sink,
332                subject,
333                operation: operation.into(),
334                next: Arc::new(Mutex::new(Some(1))),
335                host_attempt: None,
336            }),
337        }
338    }
339
340    /// Bind an explicit host attempt to its current dispatch subject.
341    ///
342    /// The logical operation, witness and HTTP send counter remain shared.
343    /// Existing clones retain their own subject and host ordinal, so an older
344    /// in-flight attempt cannot be relabeled by a concurrent retry. The host
345    /// owns ordinal allocation and must start a new context for a changed
346    /// logical request rather than reusing this operation identity.
347    pub fn for_host_attempt(&self, subject: Subject, attempt: std::num::NonZeroU64) -> Self {
348        Self {
349            inner: Arc::new(AdapterContextInner {
350                sink: self.inner.sink.clone(),
351                subject,
352                operation: self.inner.operation.clone(),
353                next: self.inner.next.clone(),
354                host_attempt: Some(attempt),
355            }),
356        }
357    }
358
359    /// Begins a send and captures request credentials for diagnostic scrubbing.
360    /// `route` must be a credential-free provider template without base-URL
361    /// prefixes or query data. `request_id_header` is the reply header the
362    /// wire reads its request id from. Returns `None` when attempt IDs are
363    /// exhausted.
364    pub(crate) fn attempt_for<B>(
365        &self,
366        request: &http::Request<B>,
367        route: &str,
368        request_id_header: Option<&'static str>,
369    ) -> Option<AdapterAttempt> {
370        let mut attempt = self.begin(request.method(), route)?;
371        attempt.secrets = scrub::request_secrets(request);
372        attempt.request_id_header = request_id_header;
373        Some(attempt)
374    }
375
376    fn emit(&self, attempt: Option<u64>, event: AdapterEvent) {
377        self.emit_with_analysis(attempt, event, None);
378    }
379
380    fn emit_with_analysis(
381        &self,
382        attempt: Option<u64>,
383        event: AdapterEvent,
384        analysis: Option<AdapterAnalysis>,
385    ) {
386        self.inner.sink.observe(Observation::new(
387            self.inner.subject.clone(),
388            Stage::Handler,
389            Emitter::named("rig-core/adapter"),
390            Action::Adapter {
391                observation: AdapterObservation {
392                    operation: self.inner.operation.clone(),
393                    attempt,
394                    host_attempt: self.inner.host_attempt,
395                    event,
396                    analysis,
397                },
398            },
399        ));
400    }
401
402    /// Allocates a send ordinal and emits its start under a credential-free
403    /// route template. Emits exhaustion at the last ordinal; later calls return
404    /// `None` without reusing identities or affecting provider execution.
405    pub(crate) fn begin(&self, method: &http::Method, route: &str) -> Option<AdapterAttempt> {
406        let number = {
407            let mut next = self
408                .inner
409                .next
410                .lock()
411                .unwrap_or_else(std::sync::PoisonError::into_inner);
412            let number = (*next)?;
413            *next = number.checked_add(1);
414            number
415        };
416        self.emit(
417            Some(number),
418            AdapterEvent::Started {
419                method: method.to_string(),
420                route: route.to_owned(),
421            },
422        );
423        if number == u64::MAX {
424            self.emit(None, AdapterEvent::IdentityExhausted);
425        }
426        Some(AdapterAttempt {
427            context: self.clone(),
428            number,
429            closed: false,
430            response_seen: false,
431            unframed: 0,
432            secrets: Vec::new(),
433            request_id_header: None,
434            pending_response_id: None,
435            error_boundary: None,
436        })
437    }
438}
439
440/// The request owns this guard until completion or cancellation.
441pub(crate) struct AdapterAttempt {
442    context: AdapterContext,
443    number: u64,
444    closed: bool,
445    response_seen: bool,
446    /// Bytes the transport's framer had not framed when the body ended.
447    unframed: usize,
448    secrets: Vec<String>,
449    /// The reply header the wire reads its request id from.
450    request_id_header: Option<&'static str>,
451    pending_response_id: Option<String>,
452    error_boundary: Option<AdapterErrorBoundary>,
453}
454
455impl AdapterAttempt {
456    pub(crate) fn text(&self, text: &str) -> String {
457        scrub::text(text, &self.secrets)
458    }
459
460    fn emit_with_analysis(&self, event: AdapterEvent, analysis: AdapterAnalysis) {
461        let analysis = (analysis != AdapterAnalysis::default()).then_some(analysis);
462        self.context
463            .emit_with_analysis(Some(self.number), event, analysis);
464    }
465
466    fn emit(&self, event: AdapterEvent) {
467        self.context.emit(Some(self.number), event);
468    }
469
470    /// Project a payload's facts through the projector that understands it.
471    pub(crate) fn project(&mut self, project: impl FnOnce(&mut ObservationSink<'_>)) {
472        project(&mut ObservationSink { attempt: self });
473    }
474
475    pub(crate) fn provider(&mut self, verdict: AdapterVerdict, response_id: Option<String>) {
476        if response_id.is_some() {
477            self.pending_response_id = response_id;
478        }
479        // An ID-only payload must not manufacture a semantic provider event.
480        // Retain at most one ID until the next verdict or the attempt closure.
481        if verdict != AdapterVerdict::default() {
482            let response_id = self.pending_response_id.take();
483            self.emit_with_analysis(
484                AdapterEvent::Provider { verdict },
485                AdapterAnalysis {
486                    response_id,
487                    ..AdapterAnalysis::default()
488                },
489            );
490        }
491    }
492
493    pub(crate) fn response_with_headers(
494        &mut self,
495        status: http::StatusCode,
496        headers: Option<&http::HeaderMap>,
497    ) {
498        if !self.response_seen {
499            self.response_seen = true;
500            self.emit_with_analysis(
501                AdapterEvent::Response {
502                    status: status.as_u16(),
503                },
504                AdapterAnalysis {
505                    headers: headers
506                        .map(|h| scrub::headers(h, &self.secrets, self.request_id_header)),
507                    ..AdapterAnalysis::default()
508                },
509            );
510        }
511    }
512
513    pub(crate) fn finish(&mut self, ending: AdapterEnding) {
514        if !self.closed {
515            self.closed = true;
516            let response_id = self.pending_response_id.take();
517            self.emit_with_analysis(
518                AdapterEvent::Finished { ending },
519                AdapterAnalysis {
520                    response_id,
521                    ..AdapterAnalysis::default()
522                },
523            );
524        }
525    }
526}
527
528impl Drop for AdapterAttempt {
529    fn drop(&mut self) {
530        self.finish(AdapterEnding::Dropped);
531    }
532}
533
534/// Where a payload projector writes its observation facts.
535///
536/// A projector reads verdicts, ids and error envelopes off a raw payload
537/// before normalization discards them. Text it forwards must go through
538/// [`Self::scrub`]: a payload can echo credentials.
539///
540/// Usage is not a projector's to report: the driver observes it once, from
541/// the usage the decoder normalized into the reply's end.
542///
543/// ```compile_fail
544/// use rig_core::wire::{AdapterEvent, AdapterUsage, ObservationSink};
545///
546/// // A projector cannot parse usage a second time off the raw payload.
547/// fn project(sink: &mut ObservationSink<'_>) {
548///     sink.emit(AdapterEvent::Usage {
549///         usage: AdapterUsage::default(),
550///     });
551/// }
552/// ```
553pub struct ObservationSink<'a> {
554    attempt: &'a mut AdapterAttempt,
555}
556
557impl ObservationSink<'_> {
558    /// Record the provider's error envelope.
559    pub fn error(&mut self, error: AdapterErrorEnvelope) {
560        self.attempt.emit(AdapterEvent::ErrorEnvelope { error });
561    }
562
563    /// Record the provider's verdict, and the response id it named.
564    pub fn provider(&mut self, verdict: AdapterVerdict, response_id: Option<String>) {
565        self.attempt.provider(verdict, response_id);
566    }
567
568    /// Bound and redact diagnostic text from the payload.
569    pub fn scrub(&self, value: &str) -> String {
570        self.attempt.text(value)
571    }
572}
573
574/// Shared by the SSE transport and frame driver, which own different boundaries.
575#[derive(Clone, Default)]
576pub(crate) struct AdapterSlot(Arc<Mutex<Option<AdapterAttempt>>>);
577
578impl AdapterSlot {
579    pub(crate) fn transport_eof(&self, after: usize) {
580        if let Some(attempt) = self
581            .0
582            .lock()
583            .unwrap_or_else(std::sync::PoisonError::into_inner)
584            .as_ref()
585        {
586            attempt.emit(AdapterEvent::TransportEof {
587                after,
588                partial_bytes: attempt.unframed,
589            });
590        }
591    }
592
593    /// Project a reply payload's facts through the projector that
594    /// understands it.
595    pub(crate) fn project(&self, project: impl FnOnce(&mut ObservationSink<'_>)) {
596        if let Some(attempt) = self
597            .0
598            .lock()
599            .unwrap_or_else(std::sync::PoisonError::into_inner)
600            .as_mut()
601        {
602            attempt.project(project);
603        }
604    }
605
606    /// Close the attempt as its reply ended: the one place usage is
607    /// observed, once, from the decoder's end, just before `Terminal`.
608    pub(crate) fn terminal(&self, usage: Option<&crate::completion::Usage>) {
609        let mut attempt = self
610            .0
611            .lock()
612            .unwrap_or_else(std::sync::PoisonError::into_inner);
613        if let Some(attempt) = attempt.as_mut().filter(|attempt| !attempt.closed) {
614            let usage = usage.map(AdapterUsage::from);
615            if let Some(usage) = usage.filter(|usage| *usage != AdapterUsage::default()) {
616                attempt.emit(AdapterEvent::Usage { usage });
617            }
618            attempt.finish(AdapterEnding::Terminal);
619        }
620    }
621
622    /// Install the attempt this send's facts belong to.
623    pub(crate) fn install(&self, attempt: Option<AdapterAttempt>) {
624        *self
625            .0
626            .lock()
627            .unwrap_or_else(std::sync::PoisonError::into_inner) = attempt;
628    }
629
630    pub(crate) fn response(&self, status: http::StatusCode) {
631        self.response_with_headers(status, None);
632    }
633
634    pub(crate) fn response_with_headers(
635        &self,
636        status: http::StatusCode,
637        headers: Option<&http::HeaderMap>,
638    ) {
639        if let Some(attempt) = self
640            .0
641            .lock()
642            .unwrap_or_else(std::sync::PoisonError::into_inner)
643            .as_mut()
644        {
645            attempt.response_with_headers(status, headers);
646        }
647    }
648
649    fn finish(&self, ending: AdapterEnding) {
650        if let Some(attempt) = self
651            .0
652            .lock()
653            .unwrap_or_else(std::sync::PoisonError::into_inner)
654            .as_mut()
655        {
656            attempt.finish(ending);
657        }
658    }
659
660    /// Preserve information before the SSE transport converts the owned error.
661    pub(crate) fn error_boundary(&self, boundary: AdapterErrorBoundary) {
662        if let Some(attempt) = self
663            .0
664            .lock()
665            .unwrap_or_else(std::sync::PoisonError::into_inner)
666            .as_mut()
667        {
668            attempt.error_boundary = Some(boundary);
669        }
670    }
671
672    pub(crate) fn fail(&self, error: &crate::error::ProviderError) {
673        if let Some(status) = error.provider_response_status() {
674            self.response(status);
675        }
676        let report = error.report();
677        let boundary = self
678            .0
679            .lock()
680            .unwrap_or_else(std::sync::PoisonError::into_inner)
681            .as_ref()
682            .and_then(|attempt| attempt.error_boundary)
683            .unwrap_or_else(|| error.boundary());
684        self.finish(AdapterEnding::Error {
685            boundary,
686            kind: report.kind.code().to_owned(),
687            status: report.http_status,
688            retryable: report.is_retryable(),
689        });
690    }
691
692    /// Record the bytes the transport's framer had not framed when the
693    /// reply's body ended.
694    pub(crate) fn unframed(&self, bytes: usize) {
695        if let Some(attempt) = self
696            .0
697            .lock()
698            .unwrap_or_else(std::sync::PoisonError::into_inner)
699            .as_mut()
700        {
701            attempt.unframed = bytes;
702        }
703    }
704
705    pub(crate) fn eof(&self, after: usize) {
706        if let Some(attempt) = self
707            .0
708            .lock()
709            .unwrap_or_else(std::sync::PoisonError::into_inner)
710            .as_mut()
711        {
712            let byte_count = attempt.unframed;
713            let ending = if byte_count == 0 {
714                AdapterEnding::Eof { after }
715            } else {
716                AdapterEnding::PartialFrame { byte_count, after }
717            };
718            attempt.finish(ending);
719        }
720    }
721
722    pub(crate) fn corrupt(&self, frame: usize) {
723        if let Some(attempt) = self
724            .0
725            .lock()
726            .unwrap_or_else(std::sync::PoisonError::into_inner)
727            .as_ref()
728        {
729            attempt
730                .context
731                .emit(Some(attempt.number), AdapterEvent::Corrupt { frame });
732        }
733    }
734}
735
736mod scrub;
737#[cfg(test)]
738mod tests;
rig_core/observe/adapter.rs

rig_core/observe/
adapter.rs
