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    normalize_identity_headers_with_change(headers).map(|(identity, _changed)| identity)
128}
129
130/// Normalize reserved identity headers and report whether their stored bytes
131/// changed. Read-side projections use this to preserve the defense-in-depth
132/// normalization contract without cloning every event header map first.
133pub(crate) fn normalize_identity_headers_with_change(
134    headers: &mut BTreeMap<String, String>,
135) -> Result<(EventIdentity, bool), EventIdentityError> {
136    let identity = EventIdentity::from_headers(headers)?;
137    let changed = EventIdentityField::ALL
138        .into_iter()
139        .any(|field| headers.get(field.header_name()).map(String::as_str) != identity.get(field));
140    if changed {
141        identity.apply_to_headers(headers)?;
142    }
143    Ok((identity, changed))
144}
145
146/// Validate and normalize reserved identity headers without materializing an
147/// [`EventIdentity`]. Read paths without a redaction hook only need the
148/// normalized stored projection, so keeping the common already-normalized
149/// case allocation-free avoids rebuilding producer identity for every event.
150pub(crate) fn normalize_identity_headers_in_place(
151    headers: &mut BTreeMap<String, String>,
152) -> Result<bool, EventIdentityError> {
153    let mut changed = false;
154    for field in EventIdentityField::ALL {
155        let Some(value) = headers.get(field.header_name()) else {
156            continue;
157        };
158        let normalized = value.trim();
159        validate_normalized_value(field, normalized)?;
160        let replacement = (normalized.len() != value.len()).then(|| normalized.to_string());
161        if let Some(replacement) = replacement {
162            headers.insert(field.header_name().to_string(), replacement);
163            changed = true;
164        }
165    }
166    Ok(changed)
167}
168
169#[derive(Clone, Debug, PartialEq, Eq)]
170pub enum EventIdentityError {
171    Invalid {
172        field: EventIdentityField,
173        reason: &'static str,
174    },
175    Conflict {
176        field: EventIdentityField,
177        existing: String,
178        incoming: String,
179    },
180}
181
182impl std::fmt::Display for EventIdentityError {
183    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
184        match self {
185            Self::Invalid { field, reason } => {
186                write!(formatter, "invalid {}: {reason}", field.header_name())
187            }
188            Self::Conflict {
189                field,
190                existing,
191                incoming,
192            } => write!(
193                formatter,
194                "conflicting {} values '{existing}' and '{incoming}'",
195                field.header_name()
196            ),
197        }
198    }
199}
200
201impl std::error::Error for EventIdentityError {}
202
203fn normalize_value(field: EventIdentityField, value: String) -> Result<String, EventIdentityError> {
204    let normalized = value.trim();
205    validate_normalized_value(field, normalized)?;
206    Ok(normalized.to_string())
207}
208
209fn validate_normalized_value(
210    field: EventIdentityField,
211    normalized: &str,
212) -> Result<(), EventIdentityError> {
213    if normalized.is_empty() {
214        return Err(EventIdentityError::Invalid {
215            field,
216            reason: "value must not be blank",
217        });
218    }
219    if normalized.chars().any(char::is_control) {
220        return Err(EventIdentityError::Invalid {
221            field,
222            reason: "value must not contain control characters",
223        });
224    }
225    Ok(())
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231
232    #[test]
233    fn identity_normalizes_and_round_trips_headers() {
234        let identity = EventIdentity::new()
235            .with(EventIdentityField::RunId, " run-1 ")
236            .unwrap()
237            .with(EventIdentityField::TurnId, "turn-1")
238            .unwrap();
239        let mut headers = BTreeMap::from([("traceparent".to_string(), "trace-1".to_string())]);
240
241        identity.apply_to_headers(&mut headers).unwrap();
242
243        assert_eq!(headers["run_id"], "run-1");
244        assert_eq!(EventIdentity::from_headers(&headers).unwrap(), identity);
245        assert_eq!(headers["traceparent"], "trace-1");
246    }
247
248    #[test]
249    fn identity_rejects_conflicting_producer_values() {
250        let identity = EventIdentity::new()
251            .with(EventIdentityField::RunId, "run-2")
252            .unwrap();
253        let mut headers = BTreeMap::from([("run_id".to_string(), "run-1".to_string())]);
254
255        let error = identity.apply_to_headers(&mut headers).unwrap_err();
256
257        assert!(matches!(error, EventIdentityError::Conflict { .. }));
258    }
259
260    #[test]
261    fn in_place_normalization_does_not_rewrite_valid_headers() {
262        let mut headers = BTreeMap::from([
263            ("run_id".to_string(), "run-1".to_string()),
264            ("turn_id".to_string(), " turn-1 ".to_string()),
265        ]);
266
267        assert!(normalize_identity_headers_in_place(&mut headers).unwrap());
268        assert_eq!(headers["run_id"], "run-1");
269        assert_eq!(headers["turn_id"], "turn-1");
270        assert!(!normalize_identity_headers_in_place(&mut headers).unwrap());
271    }
272}
harn_session_store/identity.rs

harn_session_store/
identity.rs
