agent_engine/runtime/
message_time.rs

1//! When each message happened: the host-only `_synaps_at` key.
2//!
3//! API history (role/content JSON) has no per-message time, so a transcript
4//! rebuilt from it (resume, attach, `/resync`, compaction) could only show
5//! "now". The host stamps every NEW message with its own time, in memory,
6//! under [`AT_KEY`] (RFC 3339 UTC). Like every `_synaps_*` key it never
7//! reaches a provider (`continuation::WireMessages`), and persistence keeps
8//! it beside the message, never inside (`session_journal::MESSAGE_AT_KEY`).
9//!
10//! Stamping never re-stamps and never touches settled history: a
11//! [`Stamper`] only looks at messages appended since its last pass (its
12//! floor), from the tail backward, and stops at the first message that
13//! already has a time. Messages loaded without a time (legacy sessions)
14//! stay without one and render with an empty time, never "now".
15
16use std::sync::Arc;
17
18use chrono::{DateTime, SecondsFormat, Utc};
19
20use crate::SharedMessage;
21
22/// The in-memory message time key.
23pub const AT_KEY: &str = agent_core::core::session_journal::MESSAGE_AT_KEY;
24
25/// Where message times come from. Defaults to the wall clock; tests inject
26/// their own to "advance time" between a turn and a rebuild.
27/// One thin pointer: it rides in `SessionConfig` (an `Attach` payload).
28#[derive(Clone)]
29pub struct MessageClock(Arc<ClockFn>);
30
31type ClockFn = Box<dyn Fn() -> DateTime<Utc> + Send + Sync>;
32
33impl MessageClock {
34    pub fn new(now: impl Fn() -> DateTime<Utc> + Send + Sync + 'static) -> Self {
35        Self(Arc::new(Box::new(now)))
36    }
37
38    pub fn now(&self) -> DateTime<Utc> {
39        (self.0)()
40    }
41}
42
43impl Default for MessageClock {
44    fn default() -> Self {
45        Self::new(Utc::now)
46    }
47}
48
49impl std::fmt::Debug for MessageClock {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        f.write_str("MessageClock")
52    }
53}
54
55/// The stored form of a time: RFC 3339, UTC, whole seconds.
56pub fn format_time(at: DateTime<Utc>) -> String {
57    at.to_rfc3339_opts(SecondsFormat::Secs, true)
58}
59
60/// The recorded time of a message, if any.
61pub fn message_time(message: &serde_json::Value) -> Option<&str> {
62    agent_core::core::session_journal::message_time(message)
63}
64
65/// Stamp `messages[floor..]` that lack a time with `at`, walking back from
66/// the tail and stopping at the first message that has one. Returns how many
67/// were stamped. O(new messages).
68pub fn stamp_new(messages: &mut [SharedMessage], floor: usize, at: DateTime<Utc>) -> usize {
69    let mut stamp = None;
70    let mut stamped = 0;
71    let floor = floor.min(messages.len());
72    for message in messages[floor..].iter_mut().rev() {
73        if message.get(AT_KEY).is_some() || !message.is_object() {
74            break;
75        }
76        let ts = stamp.get_or_insert_with(|| format_time(at)).clone();
77        if let Some(object) = Arc::make_mut(message).as_object_mut() {
78            object.insert(AT_KEY.into(), serde_json::Value::String(ts));
79            stamped += 1;
80        }
81    }
82    stamped
83}
84
85/// Stamps what was appended to one history since the previous pass.
86///
87/// The floor starts at the history's length when the stamper is created
88/// (everything already there is settled) and moves to the end after every
89/// pass. A history replaced by a shorter one (rollover, repair) clamps it.
90#[derive(Debug, Clone, Default)]
91pub struct Stamper {
92    floor: usize,
93}
94
95impl Stamper {
96    pub fn new(settled: usize) -> Self {
97        Self { floor: settled }
98    }
99
100    /// Stamp every new message of `messages` with `clock`'s time; returns
101    /// how many.
102    pub fn stamp(&mut self, messages: &mut [SharedMessage], clock: &MessageClock) -> usize {
103        let floor = self.floor.min(messages.len());
104        self.floor = messages.len();
105        if floor == messages.len() {
106            return 0;
107        }
108        stamp_new(messages, floor, clock.now())
109    }
110
111    /// Treat `settled` messages as already handled (a history loaded or
112    /// replaced wholesale).
113    pub fn settle(&mut self, settled: usize) {
114        self.floor = settled;
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use super::*;
121    use serde_json::json;
122
123    fn at(min: u32) -> DateTime<Utc> {
124        chrono::TimeZone::with_ymd_and_hms(&Utc, 2026, 10, 25, 5, min, 0).unwrap()
125    }
126
127    fn msgs(n: usize) -> Vec<SharedMessage> {
128        (0..n)
129            .map(|i| Arc::new(json!({"role": "user", "content": format!("m{i}")})))
130            .collect()
131    }
132
133    fn times(m: &[SharedMessage]) -> Vec<Option<&str>> {
134        m.iter().map(|m| message_time(m)).collect()
135    }
136
137    #[test]
138    fn stamping_is_idempotent_and_only_touches_new_messages() {
139        // Three legacy messages: settled, never stamped.
140        let mut m = msgs(3);
141        let clock_min = Arc::new(std::sync::atomic::AtomicU32::new(47));
142        let c = clock_min.clone();
143        let clock = MessageClock::new(move || at(c.load(std::sync::atomic::Ordering::SeqCst)));
144        let mut stamper = Stamper::new(m.len());
145        assert_eq!(stamper.stamp(&mut m, &clock), 0);
146        assert_eq!(times(&m), vec![None, None, None]);
147
148        m.extend(msgs(2));
149        assert_eq!(stamper.stamp(&mut m, &clock), 2);
150        let first = "2026-10-25T05:47:00Z";
151        assert_eq!(times(&m), vec![None, None, None, Some(first), Some(first)]);
152        let ptrs: Vec<_> = m.iter().map(Arc::as_ptr).collect();
153
154        // Later passes never re-stamp, even with the clock moved on.
155        clock_min.store(58, std::sync::atomic::Ordering::SeqCst);
156        assert_eq!(stamper.stamp(&mut m, &clock), 0);
157        assert_eq!(stamp_new(&mut m, 0, at(58)), 0, "stops at the stamped tail");
158        assert_eq!(m.iter().map(Arc::as_ptr).collect::<Vec<_>>(), ptrs, "untouched");
159
160        m.extend(msgs(1));
161        assert_eq!(stamper.stamp(&mut m, &clock), 1);
162        assert_eq!(message_time(&m[5]), Some("2026-10-25T05:58:00Z"));
163        assert_eq!(message_time(&m[3]), Some(first));
164    }
165
166    #[test]
167    fn a_shorter_replacement_history_clamps_the_floor() {
168        let clock = MessageClock::new(|| at(1));
169        let mut stamper = Stamper::new(4);
170        // Rollover: a fresh, shorter head; its survivors keep their times.
171        let mut head = msgs(2);
172        assert_eq!(stamper.stamp(&mut head, &clock), 0);
173        head.extend(msgs(1));
174        assert_eq!(stamper.stamp(&mut head, &clock), 1);
175        assert_eq!(times(&head), vec![None, None, Some("2026-10-25T05:01:00Z")]);
176    }
177}
agent_engine/runtime/message_time.rs

agent_engine/runtime/
message_time.rs
