perspt_sdk/model/
projection.rs

1//! Ledger-folded provider-neutral model context.
2
3use std::collections::BTreeSet;
4
5use serde::{Deserialize, Serialize};
6
7use super::{Conversation, Message};
8use crate::error::{Result, SdkError};
9use crate::ledger::content_hash;
10
11pub const CONVERSATION_EVENT_SCHEMA_VERSION: u32 = 1;
12
13#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
14pub struct ConversationSeeded {
15    pub event_schema_version: u32,
16    pub conversation: Conversation,
17    pub digest: String,
18}
19
20impl ConversationSeeded {
21    pub fn new(conversation: Conversation) -> Result<Self> {
22        let payload = serde_json::to_vec(&conversation)
23            .map_err(|error| SdkError::Domain(error.to_string()))?;
24        let digest = content_hash(
25            [b"conversation-seeded:v1:".as_slice(), payload.as_slice()]
26                .concat()
27                .as_slice(),
28        );
29        Ok(Self {
30            event_schema_version: CONVERSATION_EVENT_SCHEMA_VERSION,
31            conversation,
32            digest,
33        })
34    }
35}
36
37#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
38#[serde(tag = "delta", rename_all = "snake_case")]
39pub enum ConversationDelta {
40    Message { message: Message },
41    ToolActivated { name: String },
42    Compacted { control: String },
43}
44
45#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
46pub struct ConversationDeltaRecord {
47    pub event_schema_version: u32,
48    pub prior_digest: String,
49    pub delta: ConversationDelta,
50    pub digest: String,
51}
52
53/// The sole fold owner for model context, activated tools, unresolved calls,
54/// and the canonical conversation digest.
55#[derive(Debug, Clone, PartialEq)]
56pub struct ConversationProjection {
57    conversation: Conversation,
58    activated_tools: BTreeSet<String>,
59    digest: String,
60}
61
62impl ConversationProjection {
63    pub fn from_seed(seed: &ConversationSeeded) -> Result<Self> {
64        anyhow_seed(seed)?;
65        Ok(Self {
66            conversation: seed.conversation.clone(),
67            activated_tools: BTreeSet::new(),
68            digest: seed.digest.clone(),
69        })
70    }
71
72    /// Prepare a delta for durable append. The caller persists this record
73    /// before passing it to `apply`.
74    pub fn prepare(&self, delta: ConversationDelta) -> Result<ConversationDeltaRecord> {
75        let digest = delta_digest(&self.digest, &delta)?;
76        Ok(ConversationDeltaRecord {
77            event_schema_version: CONVERSATION_EVENT_SCHEMA_VERSION,
78            prior_digest: self.digest.clone(),
79            delta,
80            digest,
81        })
82    }
83
84    /// Validate a persisted delta completely, then apply it.
85    pub fn apply(&mut self, record: &ConversationDeltaRecord) -> Result<()> {
86        if record.event_schema_version != CONVERSATION_EVENT_SCHEMA_VERSION {
87            return Err(SdkError::Domain(
88                "unsupported conversation event schema".into(),
89            ));
90        }
91        if record.prior_digest != self.digest {
92            return Err(SdkError::Domain(
93                "conversation delta has a stale parent".into(),
94            ));
95        }
96        if delta_digest(&record.prior_digest, &record.delta)? != record.digest {
97            return Err(SdkError::Domain(
98                "conversation delta digest mismatch".into(),
99            ));
100        }
101        match &record.delta {
102            ConversationDelta::Message { message } => self.conversation.push(message.clone()),
103            ConversationDelta::ToolActivated { name } => {
104                self.activated_tools.insert(name.clone());
105            }
106            ConversationDelta::Compacted { control } => {
107                self.conversation.compact_with_control(control.clone());
108            }
109        }
110        self.digest = record.digest.clone();
111        Ok(())
112    }
113
114    pub fn refold(seed: &ConversationSeeded, records: &[ConversationDeltaRecord]) -> Result<Self> {
115        let mut projection = Self::from_seed(seed)?;
116        for record in records {
117            projection.apply(record)?;
118        }
119        Ok(projection)
120    }
121
122    pub fn conversation(&self) -> &Conversation {
123        &self.conversation
124    }
125
126    pub fn activated_tools(&self) -> &BTreeSet<String> {
127        &self.activated_tools
128    }
129
130    pub fn unresolved_call_ids(&self) -> Vec<String> {
131        self.conversation.unresolved_call_ids()
132    }
133
134    pub fn digest(&self) -> &str {
135        &self.digest
136    }
137}
138
139fn anyhow_seed(seed: &ConversationSeeded) -> Result<()> {
140    if seed.event_schema_version != CONVERSATION_EVENT_SCHEMA_VERSION {
141        return Err(SdkError::Domain(
142            "unsupported conversation seed schema".into(),
143        ));
144    }
145    let expected = ConversationSeeded::new(seed.conversation.clone())?;
146    if expected.digest != seed.digest {
147        return Err(SdkError::Domain("conversation seed digest mismatch".into()));
148    }
149    Ok(())
150}
151
152fn delta_digest(parent: &str, delta: &ConversationDelta) -> Result<String> {
153    let payload = serde_json::to_vec(delta).map_err(|error| SdkError::Domain(error.to_string()))?;
154    Ok(content_hash(
155        [parent.as_bytes(), b":", payload.as_slice()]
156            .concat()
157            .as_slice(),
158    ))
159}
160
161#[cfg(test)]
162mod tests {
163    use super::*;
164
165    #[test]
166    fn refold_matches_incremental_application() {
167        let seed = ConversationSeeded::new(Conversation::with_system("governed")).unwrap();
168        let mut live = ConversationProjection::from_seed(&seed).unwrap();
169        let first = live
170            .prepare(ConversationDelta::Message {
171                message: Message::User {
172                    content: "hello".into(),
173                },
174            })
175            .unwrap();
176        live.apply(&first).unwrap();
177        let second = live
178            .prepare(ConversationDelta::ToolActivated {
179                name: "lookup".into(),
180            })
181            .unwrap();
182        live.apply(&second).unwrap();
183        let folded = ConversationProjection::refold(&seed, &[first, second]).unwrap();
184        assert_eq!(live, folded);
185    }
186
187    #[test]
188    fn tampered_or_reordered_deltas_fail_closed() {
189        let seed = ConversationSeeded::new(Conversation::with_system("s")).unwrap();
190        let projection = ConversationProjection::from_seed(&seed).unwrap();
191        let mut record = projection
192            .prepare(ConversationDelta::ToolActivated { name: "a".into() })
193            .unwrap();
194        record.delta = ConversationDelta::ToolActivated { name: "b".into() };
195        assert!(ConversationProjection::refold(&seed, &[record]).is_err());
196    }
197}
perspt_sdk/model/projection.rs

perspt_sdk/model/
projection.rs
