agentmux_srv/backend/
continued_note.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The pick-up note: when a fresh session continues a pane's conversation from
5//! AgentMux's record (after an account switch, typically), the agent says
6//! nothing until the user types, because the record only rides on the first
7//! message (`continuity::prefix_user_message`). So once the session's "Given to
8//! the agent" card is written, the pane gets one ambient narration line: where
9//! things stand, then "How can I help?".
10//!
11//! View-only, like every ambient narration: it never enters the CLI transcript
12//! and the record still goes with the first message. Best-effort: a slow,
13//! capped or failed model call means no note, never a held-up session.
14
15use std::collections::HashMap;
16use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
17use std::sync::{Arc, LazyLock, Mutex};
18use std::time::Duration;
19
20use crate::backend::mps;
21use crate::backend::storage::store::Store;
22
23/// The narration kind the pane receives.
24pub(crate) const KIND: &str = "session_continued";
25
26/// How long to wait for the new session's card before writing the note anyway
27/// (a provider without the startup hook never writes one).
28const CARD_WAIT: Duration = Duration::from_secs(20);
29const CARD_POLL: Duration = Duration::from_millis(250);
30
31/// A card composed this long before the spawn was recorded still counts: the
32/// hook can run before the spawn path reaches its bookkeeping.
33const CARD_SLACK_MS: i64 = 5_000;
34
35/// One note per pane in this window: a resume the CLI rejects respawns within
36/// the same launch, and the retry carries the record again.
37const REPEAT_WINDOW_MS: i64 = 2 * 60_000;
38
39/// The latest spawn of each pane that carries the record, and when each pane
40/// last showed a note.
41#[derive(Default)]
42struct Notes {
43    /// block → (this spawn's token, its "record not yet sent" flag). A newer
44    /// spawn replaces the entry, so an older spawn's task stands down: its
45    /// flag belongs to a process that is gone, and the user may already have
46    /// typed into the new one (#4707).
47    current: HashMap<String, (u64, Arc<AtomicBool>)>,
48    /// block → when its last note was shown.
49    shown: HashMap<String, i64>,
50}
51
52static NOTES: LazyLock<Mutex<Notes>> = LazyLock::new(|| Mutex::new(Notes::default()));
53static NEXT_TOKEN: AtomicU64 = AtomicU64::new(1);
54
55fn notes() -> std::sync::MutexGuard<'static, Notes> {
56    NOTES.lock().unwrap_or_else(|e| e.into_inner())
57}
58
59impl Notes {
60    fn shown_recently(&self, block_id: &str, now: i64) -> bool {
61        self.shown.get(block_id).is_some_and(|at| now - *at < REPEAT_WINDOW_MS)
62    }
63
64    /// Records `token` as `block_id`'s latest spawn. False when a note was
65    /// shown inside [`REPEAT_WINDOW_MS`]: no second note for the same switch.
66    fn register(&mut self, block_id: &str, token: u64, pending: Arc<AtomicBool>, now: i64) -> bool {
67        if self.shown_recently(block_id, now) {
68            return false;
69        }
70        self.current.insert(block_id.to_string(), (token, pending));
71        true
72    }
73
74    /// Whether `token` is still `block_id`'s latest spawn and its record
75    /// hasn't gone out with a user message.
76    fn still_wanted(&self, block_id: &str, token: u64) -> bool {
77        self.current
78            .get(block_id)
79            .is_some_and(|(t, pending)| *t == token && pending.load(Ordering::Acquire))
80    }
81
82    /// Takes the right to show the note, once: `token` is still wanted and no
83    /// note was shown inside the window.
84    fn take_show(&mut self, block_id: &str, token: u64, now: i64) -> bool {
85        if !self.still_wanted(block_id, token) || self.shown_recently(block_id, now) {
86            return false;
87        }
88        self.current.remove(block_id);
89        self.shown.retain(|_, at| now - *at < REPEAT_WINDOW_MS);
90        self.shown.insert(block_id.to_string(), now);
91        true
92    }
93}
94
95/// Whether a stdin line is a message the user sent: a `type: user` frame
96/// that isn't AgentMux's hidden memory re-delivery (`<system-reminder>` +
97/// the reinjection signature). Control responses and hidden turns don't
98/// cancel the note.
99pub(crate) fn is_user_message(line: &str) -> bool {
100    let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
101        return false;
102    };
103    if v.get("type").and_then(serde_json::Value::as_str) != Some("user") {
104        return false;
105    }
106    let first_text = match v.get("message").and_then(|m| m.get("content")) {
107        Some(serde_json::Value::String(s)) => s.as_str(),
108        Some(serde_json::Value::Array(blocks)) => blocks
109            .iter()
110            .find_map(|b| b.get("text").and_then(serde_json::Value::as_str))
111            .unwrap_or(""),
112        _ => "",
113    };
114    !crate::ambient::digest::is_hidden_reinjection_text(first_text)
115}
116
117/// Schedules the note for a session spawned at `spawned_ms` carrying `record`.
118/// `record_pending` turns false when the record goes out with the user's first
119/// message: by then the user has spoken, and the note is dropped.
120pub(crate) fn schedule(
121    mstore: Arc<Store>,
122    broker: Arc<mps::Broker>,
123    block_id: String,
124    record: String,
125    record_pending: Arc<AtomicBool>,
126    spawned_ms: i64,
127) {
128    let token = NEXT_TOKEN.fetch_add(1, Ordering::Relaxed);
129    if !notes().register(&block_id, token, record_pending, spawned_ms) {
130        return;
131    }
132    tokio::spawn(async move {
133        let since = spawned_ms - CARD_SLACK_MS;
134        let deadline = tokio::time::Instant::now() + CARD_WAIT;
135        while !crate::server::memory_delivery_handlers::startup_card_written_since(&block_id, since)
136            && tokio::time::Instant::now() < deadline
137            && notes().still_wanted(&block_id, token)
138        {
139            tokio::time::sleep(CARD_POLL).await;
140        }
141        if !notes().still_wanted(&block_id, token) {
142            return;
143        }
144        let Some(text) =
145            crate::ambient::tasks::generate_continued_note(&mstore, &block_id, spawned_ms as u64, &record).await
146        else {
147            return;
148        };
149        // The user may have typed, or a newer spawn taken over, while the
150        // model was writing.
151        if !notes().take_show(&block_id, token, agentmux_common::time::now_ms()) {
152            return;
153        }
154        tracing::info!(block_id = %block_id, "continuity: pick-up note written");
155        let narration = crate::backend::mps_events::AmbientNarrationEvent { block_id: block_id.to_string(), kind: KIND.to_string(), text: text.to_string() };
156        broker.publish_event::<crate::backend::mps_events::AmbientNarration>(&narration, vec![format!("block:{block_id}")], 0);
157    });
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    fn flag(v: bool) -> Arc<AtomicBool> {
165        Arc::new(AtomicBool::new(v))
166    }
167
168    #[test]
169    fn a_newer_spawn_takes_over_and_the_older_task_stands_down() {
170        let mut n = Notes::default();
171        let (old, new) = (flag(true), flag(true));
172        assert!(n.register("b", 1, old.clone(), 1_000));
173        assert!(n.register("b", 2, new.clone(), 2_000), "a respawn before any note is shown registers");
174        assert!(!n.still_wanted("b", 1), "the older spawn's task stands down");
175        assert!(n.still_wanted("b", 2));
176        // The user types into the new process: its flag clears, no note.
177        new.store(false, Ordering::Release);
178        assert!(!n.take_show("b", 2, 3_000));
179        assert!(!n.take_show("b", 1, 3_000), "the old flag (still true) can't publish either");
180    }
181
182    #[test]
183    fn only_a_message_the_user_sent_cancels_the_note() {
184        let user = |content: serde_json::Value| {
185            serde_json::json!({"type": "user", "message": {"role": "user", "content": content}}).to_string()
186        };
187        assert!(is_user_message(&user(serde_json::json!("fix the login bug"))));
188        assert!(is_user_message(&user(serde_json::json!([{"type": "text", "text": "fix it"}]))));
189        let hidden = format!(
190            "<system-reminder>
191{} AgentMux could not resume this agent's prior session.
192</system-reminder>
193",
194            crate::backend::memory_delivery::REINJECTION_SIGNATURE
195        );
196        assert!(!is_user_message(&user(serde_json::json!(hidden))), "the hidden memory re-delivery");
197        assert!(!is_user_message(r#"{"type":"control_response","response":{}}"#));
198        assert!(!is_user_message("not json"));
199    }
200
201    #[test]
202    fn one_note_per_pane_inside_the_window() {
203        let mut n = Notes::default();
204        assert!(n.register("b", 1, flag(true), 1_000));
205        assert!(n.take_show("b", 1, 1_500));
206        assert!(!n.take_show("b", 1, 1_600), "shown once");
207        assert!(!n.register("b", 2, flag(true), 1_000 + REPEAT_WINDOW_MS - 1), "a retry's respawn writes no second note");
208        assert!(n.register("c", 3, flag(true), 2_000), "another pane has its own note");
209        assert!(n.register("b", 4, flag(true), 1_500 + REPEAT_WINDOW_MS), "a later switch gets a note again");
210    }
211}
agentmux_srv/backend/continued_note.rs

agentmux_srv/backend/
continued_note.rs
