taquba-workflow 0.13.0

Durable, at-least-once workflow runtime on top of the Taquba task queue. Particularly well-suited for AI agent runs.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;

use crate::Delivery;

/// Type-erased application state shared with every job handler.
#[derive(Default)]
pub(crate) struct State {
    map: HashMap<TypeId, Arc<dyn Any + Send + Sync>>,
}

impl State {
    pub(crate) fn insert<T: Any + Send + Sync>(&mut self, value: T) {
        self.map.insert(TypeId::of::<T>(), Arc::new(value));
    }

    pub(crate) fn get<T: Any + Send + Sync>(&self) -> Option<&T> {
        self.map
            .get(&TypeId::of::<T>())
            .and_then(|value| value.downcast_ref::<T>())
    }
}

/// The per-call context handed to [`Job::run`](crate::jobs::Job::run):
/// the application state registered on the
/// [`JobRunner`](crate::jobs::JobRunner) and the [`Delivery`] the job
/// runs under, which it dereferences to (identity, attempt count, lease,
/// cancellation token, memo, staged KV effects and committed KV reads).
/// It holds no domain-specific clients (HTTP, LLM, etc.); those belong
/// to the application's registered state or to layers built on top.
pub struct JobContext<'a> {
    state: &'a State,
    delivery: &'a Delivery,
}

impl<'a> JobContext<'a> {
    pub(crate) fn new(state: &'a State, delivery: &'a Delivery) -> Self {
        Self { state, delivery }
    }

    /// Borrow a piece of application state by type.
    ///
    /// State is registered on the runner via
    /// [`JobRunnerBuilder::state`](crate::jobs::JobRunnerBuilder::state).
    ///
    /// # Panics
    ///
    /// Panics if no value of type `T` was registered. Use
    /// [`try_state`](Self::try_state) for a non-panicking lookup.
    pub fn state<T: Any + Send + Sync>(&self) -> &T {
        self.try_state().unwrap_or_else(|| {
            panic!(
                "no application state of type `{}` registered on the JobRunner",
                std::any::type_name::<T>()
            )
        })
    }

    /// Borrow a piece of application state by type, returning `None` if no
    /// value of type `T` was registered.
    pub fn try_state<T: Any + Send + Sync>(&self) -> Option<&T> {
        self.state.get::<T>()
    }
}

impl Deref for JobContext<'_> {
    type Target = Delivery;

    fn deref(&self) -> &Delivery {
        self.delivery
    }
}