harn_session_store/
identity.rs

1//! Typed producer identity projected into canonical event headers.
2
3use std::collections::BTreeMap;
4
5/// Reserved canonical header keys used to correlate an event with its producer.
6#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
7pub enum EventIdentityField {
8    RunId,
9    TurnId,
10    SourceEventId,
11    MessageId,
12    ToolCallId,
13}
14
15impl EventIdentityField {
16    pub const ALL: [Self; 5] = [
17        Self::RunId,
18        Self::TurnId,
19        Self::SourceEventId,
20        Self::MessageId,
21        Self::ToolCallId,
22    ];
23
24    pub const fn header_name(self) -> &'static str {
25        match self {
26            Self::RunId => "run_id",
27            Self::TurnId => "turn_id",
28            Self::SourceEventId => "source_event_id",
29            Self::MessageId => "message_id",
30            Self::ToolCallId => "tool_call_id",
31        }
32    }
33}
34
35/// Producer-stamped run, turn, message, and tool correlation values.
36///
37/// The store persists these values in the existing signed `headers` member of
38/// `harn.session.event.v1`; adding identity therefore needs no schema fork.
39#[derive(Clone, Debug, Default, PartialEq, Eq)]
40pub struct EventIdentity {
41    values: BTreeMap<EventIdentityField, String>,
42}
43
44impl EventIdentity {
45    pub fn new() -> Self {
46        Self::default()
47    }
48
49    /// Add a field, rejecting blank/control-bearing or conflicting values.
50    pub fn with(
51        mut self,
52        field: EventIdentityField,
53        value: impl Into<String>,
54    ) -> Result<Self, EventIdentityError> {
55        self.insert(field, value)?;
56        Ok(self)
57    }
58
59    pub fn insert(
60        &mut self,
61        field: EventIdentityField,
62        value: impl Into<String>,
63    ) -> Result<(), EventIdentityError> {
64        let value = normalize_value(field, value.into())?;
65        if let Some(existing) = self.values.get(&field) {
66            if existing != &value {
67                return Err(EventIdentityError::Conflict {
68                    field,
69                    existing: existing.clone(),
70                    incoming: value,
71                });
72            }
73            return Ok(());
74        }
75        self.values.insert(field, value);
76        Ok(())
77    }
78
79    pub fn get(&self, field: EventIdentityField) -> Option<&str> {
80        self.values.get(&field).map(String::as_str)
81    }
82
83    pub fn is_empty(&self) -> bool {
84        self.values.is_empty()
85    }
86
87    /// Parse and normalize only the reserved identity keys in `headers`.
88    pub fn from_headers(headers: &BTreeMap<String, String>) -> Result<Self, EventIdentityError> {
89        let mut identity = Self::new();
90        for field in EventIdentityField::ALL {
91            if let Some(value) = headers.get(field.header_name()) {
92                identity.insert(field, value.clone())?;
93            }
94        }
95        Ok(identity)
96    }
97
98    /// Project identity into headers without silently replacing a producer ID.
99    pub fn apply_to_headers(
100        &self,
101        headers: &mut BTreeMap<String, String>,
102    ) -> Result<(), EventIdentityError> {
103        for (&field, value) in &self.values {
104            let key = field.header_name();
105            if let Some(existing) = headers.get(key) {
106                let existing = normalize_value(field, existing.clone())?;
107                if existing != *value {
108                    return Err(EventIdentityError::Conflict {
109                        field,
110                        existing,
111                        incoming: value.clone(),
112                    });
113                }
114            }
115        }
116        for (&field, value) in &self.values {
117            headers.insert(field.header_name().to_string(), value.clone());
118        }
119        Ok(())
120    }
121}
122
123/// Normalize reserved identity headers in place at the store boundary.
124pub(crate) fn normalize_identity_headers(
125    headers: &mut BTreeMap<String, String>,
126) -> Result<EventIdentity, EventIdentityError> {
127    let identity = EventIdentity::from_headers(headers)?;
128    identity.apply_to_headers(headers)?;
129    Ok(identity)
130}
131
132#[derive(Clone, Debug, PartialEq, Eq)]
133pub enum EventIdentityError {
134    Invalid {
135        field: EventIdentityField,
136        reason: &'static str,
137    },
138    Conflict {
139        field: EventIdentityField,
140        existing: String,
141        incoming: String,
142    },
143}
144
145impl std::fmt::Display for EventIdentityError {
146    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
147        match self {
148            Self::Invalid { field, reason } => {
149                write!(formatter, "invalid {}: {reason}", field.header_name())
150            }
151            Self::Conflict {
152                field,
153                existing,
154                incoming,
155            } => write!(
156                formatter,
157                "conflicting {} values '{existing}' and '{incoming}'",
158                field.header_name()
159            ),
160        }
161    }
162}
163
164impl std::error::Error for EventIdentityError {}
165
166fn normalize_value(field: EventIdentityField, value: String) -> Result<String, EventIdentityError> {
167    let normalized = value.trim();
168    if normalized.is_empty() {
169        return Err(EventIdentityError::Invalid {
170            field,
171            reason: "value must not be blank",
172        });
173    }
174    if normalized.chars().any(char::is_control) {
175        return Err(EventIdentityError::Invalid {
176            field,
177            reason: "value must not contain control characters",
178        });
179    }
180    Ok(normalized.to_string())
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186
187    #[test]
188    fn identity_normalizes_and_round_trips_headers() {
189        let identity = EventIdentity::new()
190            .with(EventIdentityField::RunId, " run-1 ")
191            .unwrap()
192            .with(EventIdentityField::TurnId, "turn-1")
193            .unwrap();
194        let mut headers = BTreeMap::from([("traceparent".to_string(), "trace-1".to_string())]);
195
196        identity.apply_to_headers(&mut headers).unwrap();
197
198        assert_eq!(headers["run_id"], "run-1");
199        assert_eq!(EventIdentity::from_headers(&headers).unwrap(), identity);
200        assert_eq!(headers["traceparent"], "trace-1");
201    }
202
203    #[test]
204    fn identity_rejects_conflicting_producer_values() {
205        let identity = EventIdentity::new()
206            .with(EventIdentityField::RunId, "run-2")
207            .unwrap();
208        let mut headers = BTreeMap::from([("run_id".to_string(), "run-1".to_string())]);
209
210        let error = identity.apply_to_headers(&mut headers).unwrap_err();
211
212        assert!(matches!(error, EventIdentityError::Conflict { .. }));
213    }
214}
harn_session_store/identity.rs

harn_session_store/
identity.rs
