rig_core/serve/
recorder.rs

1//! Runtime-independent recording of handler dispatches.
2//!
3//! ```
4//! use rig_core::serve::DispatchContext;
5//!
6//! let origin = DispatchContext::default();
7//! assert!(origin.parent.is_none());
8//! ```
9
10use std::sync::Arc;
11
12use crate::{
13    effect::{Answered, EffectId, EffectKind, HandlerDescriptor, HandlerKey},
14    error::ErrorReport,
15    streaming::{Item, StreamEvent},
16    wasm_compat::{WasmCompatSend, WasmCompatSync},
17};
18
19/// Recorded parent dispatch, stable program scope identifier and purpose,
20/// without live runtime handles.
21#[derive(Clone, Debug, Default, PartialEq, Eq)]
22pub struct DispatchContext {
23    /// The dispatch this one was made from, if a handler made it.
24    pub parent: Option<EffectId>,
25    /// The scope of the program that made it, if its dispatcher was scoped.
26    pub scope: Option<std::sync::Arc<str>>,
27    /// What it was made for, such as `"turn"` or `"summary"`, if its
28    /// dispatcher said.
29    pub purpose: Option<std::borrow::Cow<'static, str>>,
30}
31
32/// What a driver tells about the dispatches it serves. A driver calls
33/// [`handlers`](Self::handlers) once when recording starts,
34/// [`begin`](Self::begin) as each dispatch is handed to its handler,
35/// [`event`](Self::event) for every streamed event when
36/// [`keep_events`](Self::keep_events) says so, and
37/// [`resolve`](Self::resolve) when the outcome is known. A recorder is
38/// shared between the driver and its owner, so every method takes `&self`;
39/// it rides in the dispatch observer and uses the platform compatibility bounds.
40pub trait Recorder: WasmCompatSend + WasmCompatSync + 'static {
41    /// Handlers the driver serves: those registered when recording started,
42    /// then each one installed later, as it is installed. A key described
43    /// again is the same handler re-registered; the latest description
44    /// stands.
45    fn handlers(&self, handlers: Vec<HandlerDescriptor>);
46    /// A dispatch begins: its id, the key it was routed to, the effect, and
47    /// where it came from (its parent and scope).
48    fn begin(&self, id: EffectId, key: HandlerKey, kind: EffectKind, origin: DispatchContext);
49    /// Removes a begun dispatch decided by a layer before handler execution.
50    /// Replay reruns layer decisions rather than recording them as handler outcomes.
51    fn discard(&self, id: EffectId);
52    /// Replaces the recorded request with a same-family layer patch, so it
53    /// reflects the request served by the innermost handler.
54    fn patch(&self, id: EffectId, kind: EffectKind);
55    /// Whether streamed items are wanted verbatim ([`Self::event`]).
56    fn keep_events(&self) -> bool;
57    /// One streamed item of `id`.
58    fn event(&self, id: EffectId, item: &Item<StreamEvent>);
59    /// An error item at its original position in a kept stream. Unlike the
60    /// folded outcome, this includes errors after an earlier terminal item.
61    fn stream_error(&self, _id: EffectId, _error: &ErrorReport) {}
62    /// Who the streamed reply of `id` is from, before its first item, when
63    /// [`keep_events`](Self::keep_events) says so.
64    fn origin(&self, id: EffectId, origin: &crate::message::Origin);
65    /// Explicitly published output, a tool's or a layer's facts, delivered
66    /// before `resolve`. A driver snapshots it without consuming the
67    /// caller's published context.
68    fn tool_output(&self, id: EffectId, output: crate::tool::ToolResultContext);
69    /// The outcome of `id`.
70    fn resolve(&self, id: EffectId, outcome: Result<Answered, ErrorReport>);
71}
72
73/// The [`Observe`](super::Observe) a recording driver installs on each
74/// dispatch ([`Dispatch::recorded_by`](super::Dispatch::recorded_by)): it
75/// tells `recorder` everything about the dispatch `id`, and the tool output
76/// its handler published, if any, before the outcome.
77pub(crate) struct RecordingObserver {
78    pub(crate) recorder: Arc<dyn Recorder + Send + Sync>,
79    pub(crate) id: EffectId,
80    pub(crate) published: Option<Arc<crate::tool::PublishedContext>>,
81}
82
83impl super::Observe for RecordingObserver {
84    fn outcome(&mut self, outcome: Result<Answered, ErrorReport>) {
85        if let Some(output) = self
86            .published
87            .as_ref()
88            .and_then(|published| published.result_context())
89        {
90            self.recorder.tool_output(self.id, output);
91        }
92        self.recorder.resolve(self.id, outcome);
93    }
94
95    fn keep_events(&self) -> bool {
96        self.recorder.keep_events()
97    }
98
99    fn event(&mut self, item: &Item<StreamEvent>) {
100        self.recorder.event(self.id, item);
101    }
102
103    fn stream_error(&mut self, error: &ErrorReport) {
104        self.recorder.stream_error(self.id, error);
105    }
106
107    fn origin(&mut self, origin: &crate::message::Origin) {
108        self.recorder.origin(self.id, origin);
109    }
110
111    /// A layer decided the dispatch before any handler saw it: the record
112    /// is dropped, since replay reruns the layers.
113    fn discard(&mut self, _layer: &str) {
114        self.recorder.discard(self.id);
115    }
116
117    fn patch(&mut self, kind: &EffectKind) {
118        self.recorder.patch(self.id, kind.clone());
119    }
120}
121
122/// `work`, the task that drives the dispatch `id`, with a panic in it (in
123/// the handler or in its reply's stream) turned into an internal error that
124/// `recorder`, if any, also records as the dispatch's outcome. Unwinding
125/// drops the dispatch's observer, which records a cancellation; this
126/// replaces it.
127pub fn catch_panics<T>(
128    recorder: Option<Arc<dyn Recorder + Send + Sync>>,
129    id: EffectId,
130    work: impl Future<Output = Result<T, ErrorReport>>,
131) -> impl Future<Output = Result<T, ErrorReport>> {
132    use futures::FutureExt;
133    async move {
134        std::panic::AssertUnwindSafe(work)
135            .catch_unwind()
136            .await
137            .unwrap_or_else(|panic| {
138                let message = panic
139                    .downcast_ref::<&str>()
140                    .copied()
141                    .or_else(|| panic.downcast_ref::<String>().map(String::as_str))
142                    .unwrap_or("no message");
143                let report = ErrorReport::new(
144                    crate::error::ErrorKind::Internal,
145                    format!("panicked: {message}"),
146                );
147                if let Some(recorder) = recorder {
148                    recorder.resolve(id, Err(report.clone()));
149                }
150                Err(report)
151            })
152    }
153}
rig_core/serve/recorder.rs

rig_core/serve/
recorder.rs
