agent_effects/
handler.rs

1//! Durable effect handlers: effects the runtime can finish without a caller.
2//!
3//! A closure effect can only be finished by a caller holding the closure. A
4//! handler is registered with the runtime under its effect name, and its
5//! input is stored with the record. So after a crash,
6//! [`Runtime::recover`](crate::Runtime::recover) can rebuild the call from
7//! the store and finish the effect with nobody calling: verify it, re-run it
8//! if that is safe, or escalate it.
9//!
10//! ```
11//! use agent_effects::handler::{EffectHandler, Handler, VerifiableEffect};
12//! use agent_effects::{EffectContext, EffectFailure, EffectKind, EffectOutcome, Runtime, Verification};
13//! use agent_effects_memory::MemoryStore;
14//!
15//! struct SendInvoice;
16//!
17//! impl EffectHandler for SendInvoice {
18//!     const NAME: &'static str = "invoice.send";
19//!     type Input = String; // the customer's email
20//!     type Output = String; // the provider's message id
21//!     type Error = EffectFailure;
22//!
23//!     fn kind(&self) -> EffectKind {
24//!         EffectKind::IrreversibleWrite
25//!     }
26//!
27//!     async fn execute(&self, ctx: &EffectContext, to: &String) -> Result<String, EffectFailure> {
28//!         // Send through the provider, forwarding ctx.idempotency_key().
29//!         Ok(format!("msg-for-{to}"))
30//!     }
31//! }
32//!
33//! impl VerifiableEffect for SendInvoice {
34//!     async fn verify(&self, _: &EffectContext, to: &String) -> Result<Verification<String>, EffectFailure> {
35//!         // Look the message up in the provider's outbox.
36//!         Ok(Verification::Confirmed(format!("msg-for-{to}")))
37//!     }
38//! }
39//!
40//! # #[tokio::main(flavor = "current_thread")]
41//! # async fn main() -> Result<(), agent_effects::RuntimeError> {
42//! let runtime = Runtime::builder(MemoryStore::new())
43//!     .register(Handler::new(SendInvoice).verifiable())
44//!     .build();
45//!
46//! let outcome = runtime
47//!     .submit::<SendInvoice>("invoice-1001", "ada@example.com".to_string())
48//!     .actor("agent:billing")
49//!     .await?;
50//! assert_eq!(outcome, EffectOutcome::Committed("msg-for-ada@example.com".to_string()));
51//! # Ok(())
52//! # }
53//! ```
54//!
55//! Optional capabilities are separate traits ([`VerifiableEffect`]; the
56//! compensation trait follows), so a handler only implements what it can
57//! actually do. The [`Handler`] builder only offers `.verifiable()` for
58//! handlers that implement it.
59
60use std::any::Any;
61use std::collections::HashMap;
62use std::fmt::Display;
63use std::future::{Future, IntoFuture};
64use std::pin::Pin;
65use std::sync::Arc;
66use std::time::Duration;
67
68use serde::Serialize;
69use serde::de::DeserializeOwned;
70use serde_json::Value;
71
72use crate::compensation::{
73    CompensationContext, CompensationOutcome, CompensationSpec, Compensator,
74};
75use crate::effect::{EffectContext, EffectFailure, EffectOutcome, EffectSpec, Precondition};
76use crate::error::RuntimeError;
77use crate::id::{EffectKey, EffectName, LogicalKey};
78use crate::kind::EffectKind;
79use crate::policy::{Capabilities, RiskLevel};
80use crate::retry::RetryPolicy;
81use crate::runtime::Runtime;
82use crate::state::EffectStatus;
83use crate::store::{EffectRecord, EffectStore, StoreError};
84use crate::verification::{NoVerification, Verification, VerificationMode, VerifyWith};
85
86/// An effect the runtime can execute, and re-execute after a crash, from
87/// its stored input.
88///
89/// Implement `execute`; override the other methods to describe the effect.
90/// The defaults are the most cautious choices.
91pub trait EffectHandler: Send + Sync + 'static {
92    /// The effect's name, e.g. `payment.charge`. One handler per name.
93    const NAME: &'static str;
94
95    /// What the effect acts on. Stored with the record so recovery can
96    /// rebuild the call: keep credentials in the handler, not here.
97    type Input: Serialize + DeserializeOwned + Send + Sync + 'static;
98
99    /// What the effect produces. Stored and replayed to later callers.
100    type Output: Serialize + DeserializeOwned + Send + 'static;
101
102    /// The handler's error, classified for the runtime.
103    type Error: Into<EffectFailure> + Send + 'static;
104
105    /// What re-executing does to the outside world. Defaults to
106    /// [`EffectKind::IrreversibleWrite`].
107    fn kind(&self) -> EffectKind {
108        EffectKind::IrreversibleWrite
109    }
110
111    /// Whether the remote system deduplicates on
112    /// [`EffectContext::idempotency_key`], which `execute` must send.
113    fn remote_idempotency(&self) -> bool {
114        false
115    }
116
117    /// Application-defined execution semantics version. Keep it stable for
118    /// compatible releases; changes refuse existing keys before execution.
119    #[allow(clippy::unnecessary_literal_bound)] // Implementations may borrow configuration.
120    fn contract_version(&self) -> &str {
121        "1"
122    }
123
124    /// Optional trusted adapter/provider identity (never credentials).
125    fn adapter_identity(&self) -> Option<&str> {
126        None
127    }
128
129    /// Schema written for new durable inputs and migrations used by recovery.
130    fn input_schema(&self) -> crate::PayloadSchema {
131        crate::PayloadSchema::default()
132    }
133
134    /// Schema written for outputs and migrations used on committed replay.
135    fn output_schema(&self) -> crate::PayloadSchema {
136        crate::PayloadSchema::default()
137    }
138
139    /// The retry policy; `None` uses the runtime's.
140    fn retry_policy(&self) -> Option<RetryPolicy> {
141        None
142    }
143
144    /// How long to wait for one attempt; `None` waits indefinitely.
145    fn attempt_timeout(&self) -> Option<Duration> {
146        None
147    }
148
149    /// How much damage the effect could do; see
150    /// [`RiskPolicy`](crate::RiskPolicy). Defaults to [`RiskLevel::Low`].
151    fn risk(&self) -> RiskLevel {
152        RiskLevel::Low
153    }
154
155    /// Whether a human must approve the effect before its first attempt;
156    /// see [`approval`](crate::approval).
157    fn requires_approval(&self) -> bool {
158        false
159    }
160
161    /// Checked before the first attempt; see
162    /// [`EffectBuilder::precondition`](crate::EffectBuilder::precondition).
163    fn precondition(
164        &self,
165        ctx: &EffectContext,
166        input: &Self::Input,
167    ) -> impl Future<Output = Precondition> + Send {
168        let _ = (ctx, input);
169        async { Precondition::Satisfied }
170    }
171
172    /// Performs the effect once.
173    fn execute(
174        &self,
175        ctx: &EffectContext,
176        input: &Self::Input,
177    ) -> impl Future<Output = Result<Self::Output, Self::Error>> + Send;
178}
179
180/// A handler whose effect can be looked up in the remote system.
181///
182/// Register it with [`Handler::verifiable`] to verify after every success and
183/// to resolve unknown outcomes; see
184/// [`EffectBuilder::verify`](crate::EffectBuilder::verify).
185pub trait VerifiableEffect: EffectHandler {
186    /// How far a lookup can be trusted. Defaults to
187    /// [`VerificationMode::Authoritative`]; use
188    /// [`VerificationMode::EventuallyConsistent`] for lookups that lag.
189    fn verification_mode(&self) -> VerificationMode {
190        VerificationMode::Authoritative
191    }
192
193    /// Asks the remote system whether the effect applied.
194    fn verify(
195        &self,
196        ctx: &EffectContext,
197        input: &Self::Input,
198    ) -> impl Future<Output = Result<Verification<Self::Output>, Self::Error>> + Send;
199}
200
201/// A handler whose effect can be undone.
202///
203/// Register it with [`Handler::compensable`]; undo an effect with
204/// [`Runtime::compensate`](crate::Runtime::compensate). The compensation must
205/// be idempotent: a crashed or ambiguous attempt is run again. Send
206/// [`CompensationContext::idempotency_key`] to the remote system.
207pub trait CompensableEffect: EffectHandler {
208    /// Undoes the effect. `output` is what `execute` returned, if it was
209    /// stored.
210    fn compensate(
211        &self,
212        ctx: &CompensationContext,
213        input: &Self::Input,
214        output: Option<&Self::Output>,
215    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
216}
217
218type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
219
220type VerifyFn<H> = Arc<
221    dyn Fn(
222            Arc<H>,
223            EffectContext,
224            Arc<<H as EffectHandler>::Input>,
225        ) -> BoxFuture<Result<Verification<<H as EffectHandler>::Output>, EffectFailure>>
226        + Send
227        + Sync,
228>;
229
230/// A handler and the capabilities it is registered with. Pass it to
231/// [`RuntimeBuilder::register`](crate::RuntimeBuilder::register).
232pub struct Handler<H: EffectHandler> {
233    effect: Arc<H>,
234    verify: Option<(VerificationMode, VerifyFn<H>)>,
235    compensate: Option<Compensator>,
236}
237
238impl<H: EffectHandler> Clone for Handler<H> {
239    fn clone(&self) -> Self {
240        Self {
241            effect: Arc::clone(&self.effect),
242            verify: self.verify.clone(),
243            compensate: self.compensate.clone(),
244        }
245    }
246}
247
248impl<H: EffectHandler> Handler<H> {
249    fn contract(&self, retry: RetryPolicy) -> crate::store::ExecutionContract {
250        use crate::store::{ExecutionContract, VerificationContract};
251        let effect = &self.effect;
252        let mut contract = ExecutionContract::new(effect.kind(), retry.max_attempts);
253        contract.remote_idempotency = effect.remote_idempotency();
254        contract.handler_version = effect.contract_version().into();
255        contract.adapter = effect.adapter_identity().map(str::to_owned);
256        contract.require_approval = effect.requires_approval();
257        contract.verification = match self.verify.as_ref().map(|(mode, _)| mode) {
258            None | Some(VerificationMode::None) => VerificationContract::None,
259            Some(VerificationMode::Authoritative) => VerificationContract::Authoritative,
260            Some(VerificationMode::EventuallyConsistent { settle }) => {
261                VerificationContract::EventuallyConsistent { settle: *settle }
262            }
263        };
264        contract
265    }
266
267    /// Registers `handler` with no optional capabilities.
268    pub fn new(handler: H) -> Self {
269        Self {
270            effect: Arc::new(handler),
271            verify: None,
272            compensate: None,
273        }
274    }
275}
276
277impl<H: CompensableEffect> Handler<H> {
278    /// Lets the effect be undone with
279    /// [`Runtime::compensate`](crate::Runtime::compensate), using
280    /// [`CompensableEffect::compensate`], and lets recovery finish an
281    /// interrupted compensation.
282    #[must_use]
283    pub fn compensable(mut self) -> Self {
284        let handler = Arc::clone(&self.effect);
285        self.compensate = Some(Arc::new(move |ctx, input, output| {
286            let handler = Arc::clone(&handler);
287            Box::pin(async move {
288                let input = handler
289                    .input_schema()
290                    .decode(ctx.input_schema_version(), input.unwrap_or(Value::Null))
291                    .map_err(|e| EffectFailure::permanent(e.to_string()))?;
292                let input: H::Input = serde_json::from_value(input).map_err(|e| {
293                    EffectFailure::permanent(format!("stored input does not match: {e}"))
294                })?;
295                let output: Option<H::Output> = output
296                    .map(|output| {
297                        let migrated = handler
298                            .output_schema()
299                            .decode(ctx.output_schema_version(), output)
300                            .map_err(|e| EffectFailure::permanent(e.to_string()))?;
301                        serde_json::from_value(migrated)
302                            .map_err(|e| EffectFailure::permanent(e.to_string()))
303                    })
304                    .transpose()
305                    .map_err(|e| {
306                        EffectFailure::permanent(format!("stored output does not match: {e}"))
307                    })?;
308                handler
309                    .compensate(&ctx, &input, output.as_ref())
310                    .await
311                    .map_err(Into::into)
312            })
313        }));
314        self
315    }
316}
317
318impl<H: VerifiableEffect> Handler<H> {
319    /// Verifies the effect after every success and to resolve unknown
320    /// outcomes, using [`VerifiableEffect::verify`].
321    #[must_use]
322    pub fn verifiable(mut self) -> Self {
323        let mode = self.effect.verification_mode();
324        let verify: VerifyFn<H> = Arc::new(|handler, ctx, input| {
325            Box::pin(async move { handler.verify(&ctx, &input).await.map_err(Into::into) })
326        });
327        self.verify = Some((mode, verify));
328        self
329    }
330}
331
332/// A registered handler, erased so the runtime can hold handlers of many
333/// types and resume their effects by name.
334pub(crate) struct Registered<S> {
335    /// The `Handler<H>`, for typed submission.
336    typed: Arc<dyn Any + Send + Sync>,
337    /// Finishes an effect of this handler from its stored record.
338    pub(crate) resume: Resume<S>,
339}
340
341pub(crate) type Resume<S> = Arc<
342    dyn Fn(Runtime<S>, EffectRecord) -> BoxFuture<Result<EffectStatus, RuntimeError>> + Send + Sync,
343>;
344
345pub(crate) type Registry<S> = HashMap<&'static str, Registered<S>>;
346
347impl<S: EffectStore> Registered<S> {
348    pub(crate) fn new<H: EffectHandler>(handler: Handler<H>) -> Self {
349        let typed: Arc<dyn Any + Send + Sync> = Arc::new(handler.clone());
350        let resume: Resume<S> = Arc::new(move |runtime: Runtime<S>, record: EffectRecord| {
351            let handler = handler.clone();
352            Box::pin(async move { resume(&runtime, &handler, record).await })
353        });
354        Self { typed, resume }
355    }
356
357    pub(crate) fn typed<H: EffectHandler>(&self) -> Option<Handler<H>> {
358        self.typed.downcast_ref::<Handler<H>>().cloned()
359    }
360}
361
362/// A submission of input to a registered handler. Await it to run the
363/// effect. Created by [`Runtime::submit`].
364#[must_use = "a submission does nothing until it is awaited"]
365pub struct Submission<'a, S, H: EffectHandler> {
366    runtime: &'a Runtime<S>,
367    key: String,
368    input: H::Input,
369    actor: Option<String>,
370}
371
372impl<'a, S, H: EffectHandler> Submission<'a, S, H> {
373    pub(crate) fn new(runtime: &'a Runtime<S>, key: impl Display, input: H::Input) -> Self {
374        Self {
375            runtime,
376            key: key.to_string(),
377            input,
378            actor: None,
379        }
380    }
381
382    /// Who is asking for the effect, e.g. `agent:billing`.
383    pub fn actor(mut self, actor: impl Into<String>) -> Self {
384        self.actor = Some(actor.into());
385        self
386    }
387}
388
389impl<'a, S: EffectStore, H: EffectHandler> IntoFuture for Submission<'a, S, H> {
390    type Output = Result<EffectOutcome<H::Output>, RuntimeError>;
391    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + 'a>>;
392
393    fn into_future(self) -> Self::IntoFuture {
394        Box::pin(async move {
395            let handler = self
396                .runtime
397                .handler::<H>()
398                .ok_or(RuntimeError::NotRegistered { name: H::NAME })?;
399            let key = EffectKey::new(EffectName::new(H::NAME)?, LogicalKey::new(self.key)?);
400            let json = serde_json::to_value(&self.input).map_err(RuntimeError::Input)?;
401            let stored = Stored {
402                schema_version: handler.effect.input_schema().version(),
403                // Computed by the runtime, after redaction.
404                fingerprint: None,
405                json,
406                actor: self.actor,
407                from_record: false,
408            };
409            run(self.runtime, &handler, key, Arc::new(self.input), stored).await
410        })
411    }
412}
413
414/// Finishes an effect of `handler` from its record, with no caller: its
415/// compensation if one is under way, else the effect itself.
416async fn resume<S: EffectStore, H: EffectHandler>(
417    runtime: &Runtime<S>,
418    handler: &Handler<H>,
419    record: EffectRecord,
420) -> Result<EffectStatus, RuntimeError> {
421    let id = record.id;
422    runtime.enforce_contract(
423        &record,
424        handler.contract(
425            handler
426                .effect
427                .retry_policy()
428                .unwrap_or_else(|| runtime.default_retry()),
429        ),
430        handler.effect.risk(),
431    )?;
432    if record.status == EffectStatus::Compensating {
433        let compensate = handler
434            .compensate
435            .clone()
436            .ok_or(RuntimeError::NotCompensable { name: H::NAME })?;
437        let spec = CompensationSpec {
438            key: record.key.clone(),
439            reason: None,
440            actor: Some(format!("recovery:{}", runtime.worker_id())),
441            retry: handler.effect.retry_policy(),
442            attempt_timeout: handler.effect.attempt_timeout(),
443        };
444        runtime.compensate_effect(spec, compensate).await?;
445        let settled = runtime
446            .store()
447            .get(id)
448            .await?
449            .ok_or(StoreError::NotFound(id))?;
450        return Ok(settled.status);
451    }
452    let json = record.input.clone().unwrap_or(Value::Null);
453    let migrated = handler
454        .effect
455        .input_schema()
456        .decode(record.input_schema_version, json.clone())
457        .map_err(|source| RuntimeError::InputPayload { id, source })?;
458    let input: H::Input = serde_json::from_value(migrated)
459        .map_err(|source| RuntimeError::StoredInput { id, source })?;
460    // Rebuilt from the record itself, so it is the same effect by
461    // definition: reuse its fingerprint rather than recompute it, which
462    // would refuse every older record if canonicalization ever changed.
463    let stored = Stored {
464        schema_version: record.input_schema_version,
465        json,
466        fingerprint: record.input_fingerprint.clone(),
467        actor: record.created_by.clone(),
468        from_record: true,
469    };
470    run(
471        runtime,
472        handler,
473        record.key.clone(),
474        Arc::new(input),
475        stored,
476    )
477    .await?;
478    let settled = runtime
479        .store()
480        .get(id)
481        .await?
482        .ok_or(StoreError::NotFound(id))?;
483    Ok(settled.status)
484}
485
486/// What is recorded about an effect besides its key.
487struct Stored {
488    schema_version: u32,
489    json: Value,
490    fingerprint: Option<String>,
491    actor: Option<String>,
492    /// From an existing record: already redacted, fingerprint kept as is.
493    from_record: bool,
494}
495
496/// Runs `handler`'s effect through the same machinery as a closure effect.
497async fn run<S: EffectStore, H: EffectHandler>(
498    runtime: &Runtime<S>,
499    handler: &Handler<H>,
500    key: EffectKey,
501    input: Arc<H::Input>,
502    stored: Stored,
503) -> Result<EffectOutcome<H::Output>, RuntimeError> {
504    let effect = Arc::clone(&handler.effect);
505    let precondition = {
506        let (effect, input) = (Arc::clone(&effect), Arc::clone(&input));
507        Arc::new(move |ctx: EffectContext| -> BoxFuture<Precondition> {
508            let (effect, input) = (Arc::clone(&effect), Arc::clone(&input));
509            Box::pin(async move { effect.precondition(&ctx, &input).await })
510        })
511    };
512    let spec = EffectSpec {
513        input_schema_version: stored.schema_version,
514        output_schema: effect.output_schema(),
515        handler_version: effect.contract_version().into(),
516        adapter: effect.adapter_identity().map(str::to_owned),
517        fingerprint: stored.fingerprint,
518        input: Some(stored.json),
519        key,
520        capabilities: Capabilities {
521            kind: effect.kind(),
522            remote_idempotency: effect.remote_idempotency(),
523            verification: handler
524                .verify
525                .as_ref()
526                .map_or(VerificationMode::None, |(mode, _)| *mode),
527        },
528        actor: stored.actor,
529        retry: effect
530            .retry_policy()
531            .unwrap_or_else(|| runtime.default_retry()),
532        attempt_timeout: effect.attempt_timeout(),
533        precondition: Some(precondition),
534        require_approval: effect.requires_approval(),
535        risk: effect.risk(),
536        automatic_retry: true,
537        input_stored: stored.from_record,
538    };
539    let action = {
540        let (effect, input) = (Arc::clone(&effect), Arc::clone(&input));
541        move |ctx: EffectContext| {
542            let (effect, input) = (Arc::clone(&effect), Arc::clone(&input));
543            async move { effect.execute(&ctx, &input).await.map_err(Into::into) }
544        }
545    };
546    match &handler.verify {
547        Some((_, verify)) => {
548            let verify = Arc::clone(verify);
549            let checker =
550                VerifyWith(move |ctx| verify(Arc::clone(&effect), ctx, Arc::clone(&input)));
551            runtime.execute(spec, action, checker).await
552        }
553        None => runtime.execute(spec, action, NoVerification).await,
554    }
555}
556
557/// A request to undo a registered handler's effect. Await it. Created by
558/// [`Runtime::compensate`](crate::Runtime::compensate).
559#[must_use = "a compensation does nothing until it is awaited"]
560pub struct CompensationSubmission<'a, S, H: EffectHandler> {
561    runtime: &'a Runtime<S>,
562    key: String,
563    reason: Option<String>,
564    actor: Option<String>,
565    _handler: std::marker::PhantomData<fn() -> H>,
566}
567
568impl<'a, S, H: EffectHandler> CompensationSubmission<'a, S, H> {
569    pub(crate) fn new(runtime: &'a Runtime<S>, key: impl Display) -> Self {
570        Self {
571            runtime,
572            key: key.to_string(),
573            reason: None,
574            actor: None,
575            _handler: std::marker::PhantomData,
576        }
577    }
578
579    /// Why the effect is being undone; recorded in the audit trail.
580    pub fn reason(mut self, reason: impl Into<String>) -> Self {
581        self.reason = Some(reason.into());
582        self
583    }
584
585    /// Who is undoing it.
586    pub fn actor(mut self, actor: impl Into<String>) -> Self {
587        self.actor = Some(actor.into());
588        self
589    }
590}
591
592impl<'a, S: EffectStore, H: EffectHandler> IntoFuture for CompensationSubmission<'a, S, H> {
593    type Output = Result<CompensationOutcome, RuntimeError>;
594    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + 'a>>;
595
596    fn into_future(self) -> Self::IntoFuture {
597        Box::pin(async move {
598            let handler = self
599                .runtime
600                .handler::<H>()
601                .ok_or(RuntimeError::NotRegistered { name: H::NAME })?;
602            let compensate = handler
603                .compensate
604                .clone()
605                .ok_or(RuntimeError::NotCompensable { name: H::NAME })?;
606            let spec = CompensationSpec {
607                key: EffectKey::new(EffectName::new(H::NAME)?, LogicalKey::new(self.key)?),
608                reason: self.reason,
609                actor: self.actor,
610                retry: handler.effect.retry_policy(),
611                attempt_timeout: handler.effect.attempt_timeout(),
612            };
613            let record = self
614                .runtime
615                .store()
616                .get_by_key(&spec.key)
617                .await?
618                .ok_or_else(|| RuntimeError::NoSuchEffect {
619                    key: spec.key.to_string(),
620                })?;
621            if record.execution_contract.is_none() {
622                return Err(RuntimeError::LegacyContract { id: record.id });
623            }
624            self.runtime.enforce_contract(
625                &record,
626                handler.contract(
627                    handler
628                        .effect
629                        .retry_policy()
630                        .unwrap_or_else(|| self.runtime.default_retry()),
631                ),
632                handler.effect.risk(),
633            )?;
634            self.runtime.compensate_effect(spec, compensate).await
635        })
636    }
637}
agent_effects/handler.rs

agent_effects/
handler.rs
