use async_trait::async_trait;
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value;
use crate::core::{
EffectError, EffectKey, Phase, ProtectedField, Provenance, RetryPolicy, Sensitivity, SourceId,
Spend, StepId, Trust, canon,
};
impl EffectKey {
#[doc(hidden)]
#[must_use]
pub fn for_effect(
step: StepId,
phase: Phase,
ordinal: u32,
attempt: u32,
d: &EffectDescriptor,
) -> Self {
Self::derive(
step,
phase,
ordinal,
attempt,
&d.kind,
&canon::value_bytes(&d.args),
)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, serde::Deserialize)]
pub struct EffectDescriptor {
pub kind: String,
pub args: Value,
}
impl EffectDescriptor {
pub fn new(kind: impl Into<String>, args: Value) -> Self {
Self {
kind: kind.into(),
args,
}
}
pub fn nullary(kind: impl Into<String>) -> Self {
Self {
kind: kind.into(),
args: Value::Null,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, serde::Deserialize)]
pub struct DeclaredOutput {
pub trust: Trust,
pub sensitivity: Sensitivity,
}
impl DeclaredOutput {
#[must_use]
pub fn of<E: Effect + ?Sized>(effect: &E) -> Self {
Self {
trust: effect.trust(),
sensitivity: effect.output_sensitivity(),
}
}
#[must_use]
pub const fn untrusted() -> Self {
Self {
trust: Trust::Untrusted,
sensitivity: Sensitivity::Internal,
}
}
#[must_use]
pub const fn trusted() -> Self {
Self {
trust: Trust::Trusted,
sensitivity: Sensitivity::Public,
}
}
}
#[derive(Debug)]
pub enum Reconciliation<T> {
Landed(T),
DidNotHappen,
Inconclusive,
}
impl<T> Reconciliation<T> {
#[must_use]
pub fn disposition(&self) -> crate::core::Disposition {
use crate::core::Disposition;
match self {
Self::Landed(_) => Disposition::Landed,
Self::DidNotHappen => Disposition::DidNotHappen,
Self::Inconclusive => Disposition::InDoubt,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case", tag = "mode")]
pub enum Recovery {
Retry,
Idempotent { key: String },
Reconcile,
#[default]
RequiresOperator,
}
#[async_trait]
pub trait Effect: Send + Sync {
type Output: Serialize + DeserializeOwned + Send;
fn descriptor(&self) -> EffectDescriptor;
fn attach(&mut self, _provenance: &Provenance) {}
fn gen_ai_operation(&self) -> Option<&'static str> {
None
}
fn mutates(&self) -> bool {
true
}
fn recovery(&self) -> Recovery {
if self.mutates() {
Recovery::RequiresOperator
} else {
Recovery::Retry
}
}
fn retry(&self) -> RetryPolicy {
RetryPolicy::default()
}
fn max_sensitivity(&self) -> Sensitivity {
Sensitivity::Public
}
fn sink_arguments(&self) -> Option<&Value> {
None
}
fn protected_fields(&self) -> &[ProtectedField] {
&[]
}
fn delegation_depth(&self) -> Option<usize> {
None
}
fn source(&self) -> SourceId {
SourceId::new(format!("effect:{}", self.descriptor().kind))
}
fn trust(&self) -> Trust {
Trust::Untrusted
}
fn output_sensitivity(&self) -> Sensitivity {
Sensitivity::Public
}
fn spend(&self, _output: &Self::Output) -> Spend {
Spend::default()
}
async fn perform(&self) -> Result<Self::Output, EffectError>;
async fn reconcile(&self) -> Result<Reconciliation<Self::Output>, EffectError> {
Ok(Reconciliation::Inconclusive)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GroupOutcome {
Committed,
Aborted,
Quarantined,
}
impl GroupOutcome {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Committed => "committed",
Self::Aborted => "aborted",
Self::Quarantined => "quarantined",
}
}
}
#[async_trait]
pub trait AnyEffect: Send + Sync {
fn descriptor(&self) -> EffectDescriptor;
fn attach_erased(&mut self, provenance: &Provenance);
fn gen_ai_operation(&self) -> Option<&'static str>;
fn mutates(&self) -> bool;
fn recovery(&self) -> Recovery;
fn retry(&self) -> RetryPolicy;
fn max_sensitivity(&self) -> Sensitivity;
fn sink_arguments(&self) -> Option<&Value>;
fn protected_fields(&self) -> &[ProtectedField];
fn delegation_depth(&self) -> Option<usize>;
fn source(&self) -> SourceId;
fn trust(&self) -> Trust;
fn output_sensitivity(&self) -> Sensitivity;
fn spend_erased(&self, output: &Value) -> Spend;
async fn perform_erased(&self) -> Result<Value, EffectError>;
async fn reconcile_erased(&self) -> Result<Reconciliation<Value>, EffectError>;
}
#[async_trait]
impl<E> AnyEffect for E
where
E: Effect,
{
fn descriptor(&self) -> EffectDescriptor {
Effect::descriptor(self)
}
fn attach_erased(&mut self, provenance: &Provenance) {
Effect::attach(self, provenance);
}
fn gen_ai_operation(&self) -> Option<&'static str> {
Effect::gen_ai_operation(self)
}
fn mutates(&self) -> bool {
Effect::mutates(self)
}
fn recovery(&self) -> Recovery {
Effect::recovery(self)
}
fn retry(&self) -> RetryPolicy {
Effect::retry(self)
}
fn max_sensitivity(&self) -> Sensitivity {
Effect::max_sensitivity(self)
}
fn sink_arguments(&self) -> Option<&Value> {
Effect::sink_arguments(self)
}
fn protected_fields(&self) -> &[ProtectedField] {
Effect::protected_fields(self)
}
fn delegation_depth(&self) -> Option<usize> {
Effect::delegation_depth(self)
}
fn source(&self) -> SourceId {
Effect::source(self)
}
fn trust(&self) -> Trust {
Effect::trust(self)
}
fn output_sensitivity(&self) -> Sensitivity {
Effect::output_sensitivity(self)
}
fn spend_erased(&self, output: &Value) -> Spend {
serde_json::from_value::<E::Output>(output.clone())
.map(|o| Effect::spend(self, &o))
.unwrap_or_default()
}
async fn perform_erased(&self) -> Result<Value, EffectError> {
let out = Effect::perform(self).await?;
serde_json::to_value(out).map_err(EffectError::OutputShape)
}
async fn reconcile_erased(&self) -> Result<Reconciliation<Value>, EffectError> {
Ok(match Effect::reconcile(self).await? {
Reconciliation::Landed(o) => {
Reconciliation::Landed(serde_json::to_value(o).map_err(EffectError::OutputShape)?)
}
Reconciliation::DidNotHappen => Reconciliation::DidNotHappen,
Reconciliation::Inconclusive => Reconciliation::Inconclusive,
})
}
}
#[async_trait]
impl Effect for Box<dyn AnyEffect + '_> {
type Output = Value;
fn descriptor(&self) -> EffectDescriptor {
(**self).descriptor()
}
fn attach(&mut self, provenance: &Provenance) {
(**self).attach_erased(provenance);
}
fn gen_ai_operation(&self) -> Option<&'static str> {
(**self).gen_ai_operation()
}
fn mutates(&self) -> bool {
(**self).mutates()
}
fn recovery(&self) -> Recovery {
(**self).recovery()
}
fn retry(&self) -> RetryPolicy {
(**self).retry()
}
fn max_sensitivity(&self) -> Sensitivity {
(**self).max_sensitivity()
}
fn sink_arguments(&self) -> Option<&Value> {
(**self).sink_arguments()
}
fn protected_fields(&self) -> &[ProtectedField] {
(**self).protected_fields()
}
fn delegation_depth(&self) -> Option<usize> {
(**self).delegation_depth()
}
fn source(&self) -> SourceId {
(**self).source()
}
fn trust(&self) -> Trust {
(**self).trust()
}
fn output_sensitivity(&self) -> Sensitivity {
(**self).output_sensitivity()
}
fn spend(&self, output: &Value) -> Spend {
(**self).spend_erased(output)
}
async fn perform(&self) -> Result<Value, EffectError> {
(**self).perform_erased().await
}
async fn reconcile(&self) -> Result<Reconciliation<Value>, EffectError> {
(**self).reconcile_erased().await
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn descriptors_with_reordered_args_are_equal_after_canonicalization() {
let a = EffectDescriptor::new("tool.call", json!({"b": 2, "a": 1}));
let b = EffectDescriptor::new("tool.call", json!({"a": 1, "b": 2}));
assert_eq!(
crate::core::canon::value_bytes(&a.args),
crate::core::canon::value_bytes(&b.args),
"argument order must not change an effect's identity"
);
}
#[test]
fn mutating_effects_default_to_operator_recovery() {
struct Mutating;
#[async_trait]
impl Effect for Mutating {
type Output = ();
fn descriptor(&self) -> EffectDescriptor {
EffectDescriptor::nullary("test.mutate")
}
async fn perform(&self) -> Result<(), EffectError> {
Ok(())
}
}
assert!(matches!(
Effect::recovery(&Mutating),
Recovery::RequiresOperator
));
}
#[test]
fn read_only_effects_default_to_retry() {
struct ReadOnly;
#[async_trait]
impl Effect for ReadOnly {
type Output = ();
fn descriptor(&self) -> EffectDescriptor {
EffectDescriptor::nullary("test.read")
}
fn mutates(&self) -> bool {
false
}
async fn perform(&self) -> Result<(), EffectError> {
Ok(())
}
}
assert!(matches!(Effect::recovery(&ReadOnly), Recovery::Retry));
}
}