theway-core 0.1.21

theway core — stateful agent runtime + harness (Agent loop, skills, prompt templates, sessions, compaction) on top of theway-llm-provider.
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

use std::sync::Arc;

use async_trait::async_trait;
use theway_contract::extension::ExtensionDurableEntry;
use theway_contract::session::{SessionError, SessionStore, StoredSessionEntry};
use thiserror::Error;

#[async_trait]
pub trait SessionExtensionStatePort: Send + Sync {
    async fn append_durable_entries(
        &self,
        extension_id: &str,
        entries: Vec<ExtensionDurableEntry>,
    ) -> Result<Vec<String>, SessionExtensionStateError>;

    async fn replay_durable_entries(
        &self,
        extension_id: &str,
        leaf_id: Option<&str>,
    ) -> Result<Vec<ExtensionDurableEntry>, SessionExtensionStateError>;
}

pub struct PersistentSessionExtensionStatePort {
    store: Arc<dyn SessionStore>,
}

impl PersistentSessionExtensionStatePort {
    pub fn new(store: Arc<dyn SessionStore>) -> Self {
        Self { store }
    }

    pub fn store(&self) -> &Arc<dyn SessionStore> {
        &self.store
    }
}

#[async_trait]
impl SessionExtensionStatePort for PersistentSessionExtensionStatePort {
    async fn append_durable_entries(
        &self,
        extension_id: &str,
        entries: Vec<ExtensionDurableEntry>,
    ) -> Result<Vec<String>, SessionExtensionStateError> {
        if entries.is_empty() {
            return Ok(Vec::new());
        }
        for entry in &entries {
            entry
                .validate()
                .map_err(|error| SessionExtensionStateError::InvalidEntry(error.to_string()))?;
            if entry.extension_id != extension_id {
                return Err(SessionExtensionStateError::OwnerMismatch {
                    expected: extension_id.to_string(),
                    actual: entry.extension_id.clone(),
                });
            }
        }

        let mut parent_id = self.store.get_leaf_id().await?;
        let timestamp = chrono::Utc::now().to_rfc3339();
        let mut ids = Vec::with_capacity(entries.len());
        let mut stored = Vec::with_capacity(entries.len());
        for entry in entries {
            let id = self.store.create_entry_id().await?;
            stored.push(StoredSessionEntry::extension(
                id.clone(),
                parent_id,
                timestamp.clone(),
                entry,
            )?);
            parent_id = Some(id.clone());
            ids.push(id);
        }
        self.store.append_entries(stored).await?;
        Ok(ids)
    }

    async fn replay_durable_entries(
        &self,
        extension_id: &str,
        leaf_id: Option<&str>,
    ) -> Result<Vec<ExtensionDurableEntry>, SessionExtensionStateError> {
        self.store
            .get_extension_entries(extension_id, leaf_id)
            .await?
            .into_iter()
            .map(|entry| {
                entry.extension_payload()?.ok_or_else(|| {
                    SessionExtensionStateError::InvalidEntry(
                        "extension query returned a non-extension entry".into(),
                    )
                })
            })
            .collect()
    }
}

#[derive(Clone, Copy, Debug, Default)]
pub struct NoopSessionExtensionStatePort;

#[async_trait]
impl SessionExtensionStatePort for NoopSessionExtensionStatePort {
    async fn append_durable_entries(
        &self,
        _extension_id: &str,
        entries: Vec<ExtensionDurableEntry>,
    ) -> Result<Vec<String>, SessionExtensionStateError> {
        if entries.is_empty() {
            Ok(Vec::new())
        } else {
            Err(SessionExtensionStateError::Unavailable)
        }
    }

    async fn replay_durable_entries(
        &self,
        _extension_id: &str,
        _leaf_id: Option<&str>,
    ) -> Result<Vec<ExtensionDurableEntry>, SessionExtensionStateError> {
        Ok(Vec::new())
    }
}

#[derive(Debug, Error)]
pub enum SessionExtensionStateError {
    #[error("session extension state persistence is unavailable")]
    Unavailable,
    #[error("extension durable entry is invalid: {0}")]
    InvalidEntry(String),
    #[error("extension durable entry owner mismatch: expected {expected}, got {actual}")]
    OwnerMismatch { expected: String, actual: String },
    #[error(transparent)]
    Session(#[from] SessionError),
}