agentmux_srv/backend/
narrated_events.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Remembers which events have already been narrated, so one background launch
5//! produces one ambient narration.
6//!
7//! WHY THIS IS PROCESS-WIDE AND NOT A LOCAL IN `register_handlers`.
8//! That function runs once per WebSocket CONNECTION, so a local set is reset by
9//! every reconnect — and a reconnect is exactly when duplicates arrive. An
10//! honored truncate clears the frontend's own `nodeIdSet` and re-parses the
11//! transcript, re-emitting every node as new, which re-invokes the narrate RPC
12//! for ids already narrated before the drop. Nothing else backstops it: the
13//! `AmbientGateway` singleton only guards calls that are concurrently in
14//! flight for a key, and releases its guard as soon as the first completes.
15//! Living on `AppState` alongside `dock_snapshots` and `pending_background_pids`
16//! is what makes the "narrate once" claim actually true. (reagent P1 on #3169.)
17//!
18//! Bounded, unlike a bare `HashSet`: this accumulates one entry per narrated
19//! event for the lifetime of the process, which for a long-lived instance
20//! running many background tasks is unbounded growth for a guard that only
21//! needs recent history. Oldest entries are evicted first — a reconnect replays
22//! recent nodes, not ancient ones, so forgetting the distant past costs nothing.
23
24use std::collections::{HashSet, VecDeque};
25
26use parking_lot::Mutex;
27
28/// How many narrated event ids to remember.
29///
30/// Generous relative to what the guard actually needs (a reconnect replays the
31/// current transcript, not the whole session) and still trivially small in
32/// memory — a few hundred short strings.
33const MAX_REMEMBERED: usize = 512;
34
35#[derive(Default)]
36struct Inner {
37    seen: HashSet<String>,
38    order: VecDeque<String>,
39}
40
41#[derive(Default)]
42pub struct NarratedEvents {
43    inner: Mutex<Inner>,
44}
45
46impl NarratedEvents {
47    pub fn new() -> Self {
48        Self::default()
49    }
50
51    /// Record `event_id` and report whether it is NEW.
52    ///
53    /// `true` means "not seen before — go ahead and narrate". `false` means it
54    /// has already been narrated and this call should do nothing.
55    pub fn mark_new(&self, event_id: &str) -> bool {
56        let mut g = self.inner.lock();
57        if !g.seen.insert(event_id.to_string()) {
58            return false;
59        }
60        g.order.push_back(event_id.to_string());
61        while g.order.len() > MAX_REMEMBERED {
62            if let Some(evicted) = g.order.pop_front() {
63                g.seen.remove(&evicted);
64            }
65        }
66        true
67    }
68}
69
70#[cfg(test)]
71mod tests {
72    use super::*;
73
74    #[test]
75    fn the_first_sighting_is_new_and_the_second_is_not() {
76        let n = NarratedEvents::new();
77        assert!(n.mark_new("toolu_1"));
78        assert!(!n.mark_new("toolu_1"));
79    }
80
81    #[test]
82    fn distinct_events_are_each_new() {
83        let n = NarratedEvents::new();
84        assert!(n.mark_new("toolu_1"));
85        assert!(n.mark_new("toolu_2"));
86    }
87
88    #[test]
89    fn it_stays_bounded_and_evicts_oldest_first() {
90        let n = NarratedEvents::new();
91        for i in 0..(MAX_REMEMBERED + 10) {
92            assert!(n.mark_new(&format!("id{i}")));
93        }
94        // The oldest have been forgotten — re-narrating one is the accepted
95        // cost of not growing without bound.
96        assert!(n.mark_new("id0"));
97        // Recent ones are still remembered, which is what the guard is for.
98        let newest = format!("id{}", MAX_REMEMBERED + 9);
99        assert!(!n.mark_new(&newest));
100    }
101}
agentmux_srv/backend/narrated_events.rs

agentmux_srv/backend/
narrated_events.rs
