agentmux_srv/backend/agent_session/
history_summary.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! What an agent was last asked, read from its GLOBAL transcript zone.
5//!
6//! "My Agents" needs a preview, a last-activity time and an honest empty state
7//! for every agent, whether or not it was launched in this channel and whether
8//! or not a pane is open. The per-block `output.state.json` it used to read is
9//! no longer written; the history lives under `agent:<defId>:current` (or, once
10//! archived, `agent:<defId>:archive:<ms>`) in the shared transcript store, keyed
11//! by agent id and written by every channel.
12//!
13//! The `output` file there is the provider's raw NDJSON stream and can be
14//! gigabytes (most of it `stream_event` deltas), so it is never loaded whole:
15//! the newest user prompt is found by reading backwards in fixed chunks, up to
16//! a cap, and the result is remembered until the file changes.
17//! Spec: SPEC_MY_AGENTS_TILES_AUTH_AND_HISTORY_2026_10_03.md §5.3.
18
19use std::collections::HashMap;
20use std::sync::Mutex;
21
22use crate::backend::storage::filestore::FileStore;
23
24use super::archive::collapse_preview;
25use super::session_io::OUTPUT_FILE;
26use super::zone_naming::{agent_current_zone, is_valid_definition_id};
27
28/// Bytes read per backwards step.
29const CHUNK: i64 = 1024 * 1024;
30/// The most of a transcript's tail one lookup reads. Real transcripts put the
31/// newest user prompt well inside the last two chunks; past this the answer is
32/// "none recent", not a longer read.
33const MAX_SCAN: i64 = 4 * CHUNK;
34/// A single line longer than this is skipped rather than carried across chunks.
35const MAX_LINE: usize = 2 * 1024 * 1024;
36/// Remembered results kept before the memo is reset.
37const MEMO_CAP: usize = 512;
38
39/// What the transcript store knows about one agent.
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub enum HistoryRead {
42    /// No transcript for this agent: it has never run, or left nothing behind.
43    NoTranscript,
44    /// A transcript exists.
45    Transcript(AgentHistory),
46    /// The store could not be read; the real state is unknown.
47    Failed,
48}
49
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct AgentHistory {
52    /// The newest real user prompt in the scanned tail, collapsed to one line.
53    /// Empty when the tail holds none (a long run of tool work).
54    pub last_user_message: String,
55    /// When the transcript last changed, in the store's own unit (ms).
56    pub last_activity_ms: i64,
57}
58
59/// The store to read: the process-global transcript store, or in tests the one
60/// the current thread installed.
61fn history_store() -> Option<std::sync::Arc<FileStore>> {
62    #[cfg(test)]
63    {
64        if let Some(s) = TEST_STORE.with(|t| t.borrow().clone()) {
65            return Some(s);
66        }
67    }
68    super::global_store::global_transcript_store().cloned()
69}
70
71#[cfg(test)]
72thread_local! {
73    static TEST_STORE: std::cell::RefCell<Option<std::sync::Arc<FileStore>>> =
74        const { std::cell::RefCell::new(None) };
75}
76
77/// Install a transcript store for the current thread only (tests).
78#[cfg(test)]
79pub(crate) fn set_test_store(store: Option<std::sync::Arc<FileStore>>) {
80    TEST_STORE.with(|t| *t.borrow_mut() = store);
81}
82
83/// History for each distinct agent id, read on the blocking pool. Empty when no
84/// transcript store is installed, which callers treat as "use the old source".
85pub async fn read_histories(definition_ids: Vec<String>) -> HashMap<String, HistoryRead> {
86    let Some(store) = history_store() else {
87        return HashMap::new();
88    };
89    let mut ids = definition_ids;
90    ids.sort();
91    ids.dedup();
92    tokio::task::spawn_blocking(move || {
93        ids.into_iter()
94            .map(|id| {
95                let r = read_agent_history(&store, &id);
96                (id, r)
97            })
98            .collect()
99    })
100    .await
101    .unwrap_or_default()
102}
103
104type MemoKey = (usize, String);
105type MemoVal = (i64, i64, AgentHistory);
106
107fn memo() -> &'static Mutex<HashMap<MemoKey, MemoVal>> {
108    static MEMO: std::sync::OnceLock<Mutex<HashMap<MemoKey, MemoVal>>> = std::sync::OnceLock::new();
109    MEMO.get_or_init(|| Mutex::new(HashMap::new()))
110}
111
112/// Read what `definition_id`'s transcript says, falling back to its newest
113/// archive when the current zone is empty.
114pub fn read_agent_history(store: &FileStore, definition_id: &str) -> HistoryRead {
115    if !is_valid_definition_id(definition_id) {
116        return HistoryRead::NoTranscript;
117    }
118    let zone = match newest_zone_with_output(store, definition_id) {
119        Ok(Some(z)) => z,
120        Ok(None) => return HistoryRead::NoTranscript,
121        Err(e) => {
122            tracing::warn!(error = %e, definition_id, "agent history: transcript lookup failed");
123            return HistoryRead::Failed;
124        }
125    };
126    let (zone, size, modts) = zone;
127    let key = (store as *const FileStore as usize, zone.clone());
128    if let Some((s, m, h)) = memo().lock().unwrap().get(&key) {
129        if *s == size && *m == modts {
130            return HistoryRead::Transcript(h.clone());
131        }
132    }
133    let last_user_message = match scan_tail_for_prompt(store, &zone, size) {
134        Ok(m) => m,
135        Err(e) => {
136            tracing::warn!(error = %e, zone = %zone, "agent history: transcript tail read failed");
137            return HistoryRead::Failed;
138        }
139    };
140    let history = AgentHistory { last_user_message, last_activity_ms: modts };
141    let mut memo = memo().lock().unwrap();
142    if memo.len() >= MEMO_CAP {
143        memo.clear();
144    }
145    memo.insert(key, (size, modts, history.clone()));
146    HistoryRead::Transcript(history)
147}
148
149/// `(zone, size, modts)` of the agent's current zone if it has output, else of
150/// its newest archive that does.
151fn newest_zone_with_output(
152    store: &FileStore,
153    definition_id: &str,
154) -> Result<Option<(String, i64, i64)>, String> {
155    let current = agent_current_zone(definition_id);
156    if let Some(f) = store.stat(&current, OUTPUT_FILE).map_err(|e| e.to_string())? {
157        if f.size > 0 {
158            return Ok(Some((current, f.size, f.modts)));
159        }
160    }
161    let prefix = format!("agent:{definition_id}:archive:");
162    let mut archives: Vec<(u64, String)> = store
163        .zone_ids_with_prefix(&prefix)
164        .map_err(|e| e.to_string())?
165        .into_iter()
166        .filter_map(|z| z.strip_prefix(&prefix).and_then(|t| t.parse::<u64>().ok()).map(|ts| (ts, z)))
167        .collect();
168    archives.sort_by(|a, b| b.0.cmp(&a.0));
169    for (_, zone) in archives {
170        if let Some(f) = store.stat(&zone, OUTPUT_FILE).map_err(|e| e.to_string())? {
171            if f.size > 0 {
172                return Ok(Some((zone, f.size, f.modts)));
173            }
174        }
175    }
176    Ok(None)
177}
178
179/// The newest real user prompt within the last [`MAX_SCAN`] bytes of `output`,
180/// or an empty string if there is none.
181fn scan_tail_for_prompt(store: &FileStore, zone: &str, size: i64) -> Result<String, String> {
182    let floor = (size - MAX_SCAN).max(0);
183    let mut end = size;
184    // The start of the line that straddles the boundary with the chunk after
185    // this one; prepended to nothing, appended to this chunk's bytes.
186    let mut carry: Vec<u8> = Vec::new();
187    while end > floor {
188        let start = (end - CHUNK).max(floor);
189        let (_, mut buf) = store
190            .read_at(zone, OUTPUT_FILE, start, end - start)
191            .map_err(|e| e.to_string())?;
192        buf.extend_from_slice(&carry);
193        carry.clear();
194        // Unless this chunk begins the file, its first line may be a fragment
195        // of one that began in the chunk before; hold it back.
196        let body_from = if start > 0 {
197            match buf.iter().position(|b| *b == b'\n') {
198                Some(i) => {
199                    if i <= MAX_LINE {
200                        carry = buf[..i].to_vec();
201                    }
202                    i + 1
203                }
204                None => {
205                    if buf.len() <= MAX_LINE {
206                        carry = std::mem::take(&mut buf);
207                    }
208                    buf.len()
209                }
210            }
211        } else {
212            0
213        };
214        for line in buf[body_from.min(buf.len())..].split(|b| *b == b'\n').rev() {
215            if let Some(msg) = user_prompt_in_line(line) {
216                return Ok(collapse_preview(&msg));
217            }
218        }
219        end = start;
220    }
221    Ok(String::new())
222}
223
224/// The text of a real user prompt if `line` is one: a `user` record that is not
225/// a tool result, not from a sub-agent, and not startup or injected scaffolding.
226fn user_prompt_in_line(line: &[u8]) -> Option<String> {
227    // Cheap reject before parsing: nearly every line is a stream_event.
228    if !contains(line, br#""type":"user""#) {
229        return None;
230    }
231    let v: serde_json::Value = serde_json::from_slice(line).ok()?;
232    if v.get("type")?.as_str()? != "user" {
233        return None;
234    }
235    if v.get("tool_use_result").is_some() {
236        return None;
237    }
238    if v.get("parent_tool_use_id").map_or(false, |p| !p.is_null()) {
239        return None;
240    }
241    let content = v.get("message")?.get("content")?;
242    let text = match content {
243        serde_json::Value::String(s) => s.clone(),
244        serde_json::Value::Array(blocks) => {
245            if blocks.iter().any(|b| b.get("type").and_then(|t| t.as_str()) == Some("tool_result")) {
246                return None;
247            }
248            blocks
249                .iter()
250                .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
251                .collect::<Vec<_>>()
252                .join("\n")
253        }
254        _ => return None,
255    };
256    let text = text.trim();
257    if text.is_empty() || is_scaffolding(text) {
258        return None;
259    }
260    Some(text.to_string())
261}
262
263/// Text the harness puts in the user turn that the user did not type.
264fn is_scaffolding(text: &str) -> bool {
265    const PREFIXES: &[&str] = &[
266        "# Session Context",
267        "[JEKT:",
268        "[BROADCAST:",
269        "[AgentMux]",
270        "<system-reminder>",
271        "<command-name>",
272        "<command-message>",
273        "<local-command",
274        "Caveat:",
275        "[Request interrupted",
276    ];
277    PREFIXES.iter().any(|p| text.starts_with(p))
278}
279
280fn contains(haystack: &[u8], needle: &[u8]) -> bool {
281    haystack.windows(needle.len()).any(|w| w == needle)
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287    use crate::backend::storage::filestore::{FileMeta, FileOpts};
288    use std::collections::HashMap as Map;
289
290    fn put(fs: &FileStore, zone: &str, body: &[u8]) {
291        let meta: FileMeta = Map::new();
292        fs.make_file(zone, OUTPUT_FILE, meta, FileOpts::default()).unwrap();
293        fs.write_file(zone, OUTPUT_FILE, body).unwrap();
294    }
295
296    fn user(text: &str) -> String {
297        serde_json::json!({"type":"user","message":{"role":"user","content":text}}).to_string()
298    }
299
300    fn delta() -> String {
301        r#"{"type":"stream_event","event":{"type":"content_block_delta"}}"#.to_string()
302    }
303
304    fn read(fs: &FileStore, id: &str) -> HistoryRead {
305        read_agent_history(fs, id)
306    }
307
308    #[test]
309    fn no_zone_is_no_transcript_and_a_bad_id_is_too() {
310        let fs = FileStore::open_in_memory().unwrap();
311        assert_eq!(read(&fs, "agent-1"), HistoryRead::NoTranscript);
312        assert_eq!(read(&fs, "../evil"), HistoryRead::NoTranscript);
313    }
314
315    #[test]
316    fn finds_the_newest_real_prompt_and_skips_tool_results_and_scaffolding() {
317        let fs = FileStore::open_in_memory().unwrap();
318        let body = [
319            user("# Session Context\nboilerplate"),
320            user("first real question"),
321            delta(),
322            user("fix the login race"),
323            r#"{"type":"user","tool_use_result":{},"message":{"role":"user","content":[{"type":"tool_result","content":"x"}]}}"#.to_string(),
324            r#"{"type":"user","parent_tool_use_id":"toolu_1","message":{"role":"user","content":"sub-agent prompt"}}"#.to_string(),
325            user("[JEKT:FROM=someone TIER=coord]\nbody"),
326            delta(),
327        ]
328        .join("\n");
329        put(&fs, "agent:agent-1:current", body.as_bytes());
330        match read(&fs, "agent-1") {
331            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "fix the login race"),
332            other => panic!("expected a transcript, got {other:?}"),
333        }
334    }
335
336    #[test]
337    fn reads_a_prompt_given_as_text_blocks() {
338        let fs = FileStore::open_in_memory().unwrap();
339        let body = serde_json::json!({"type":"user","message":{"role":"user","content":[
340            {"type":"text","text":"line one"},{"type":"text","text":"line two"}]}})
341        .to_string();
342        put(&fs, "agent:agent-2:current", body.as_bytes());
343        match read(&fs, "agent-2") {
344            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "line one line two"),
345            other => panic!("expected a transcript, got {other:?}"),
346        }
347    }
348
349    #[test]
350    fn a_transcript_with_no_prompt_is_still_a_transcript() {
351        let fs = FileStore::open_in_memory().unwrap();
352        put(&fs, "agent:agent-3:current", delta().as_bytes());
353        match read(&fs, "agent-3") {
354            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, ""),
355            other => panic!("expected a transcript, got {other:?}"),
356        }
357    }
358
359    #[test]
360    fn an_archived_only_agent_reads_its_newest_archive() {
361        let fs = FileStore::open_in_memory().unwrap();
362        put(&fs, "agent:agent-4:archive:1000", user("old question").as_bytes());
363        put(&fs, "agent:agent-4:archive:2000", user("newer question").as_bytes());
364        match read(&fs, "agent-4") {
365            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "newer question"),
366            other => panic!("expected a transcript, got {other:?}"),
367        }
368    }
369
370    #[test]
371    fn an_empty_current_zone_falls_through_to_the_archive() {
372        let fs = FileStore::open_in_memory().unwrap();
373        put(&fs, "agent:agent-5:current", b"");
374        put(&fs, "agent:agent-5:archive:5", user("before the archive").as_bytes());
375        match read(&fs, "agent-5") {
376            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "before the archive"),
377            other => panic!("expected a transcript, got {other:?}"),
378        }
379    }
380
381    /// The point of the tail read: a transcript far larger than the scan cap is
382    /// answered from its end without reading the rest, and a prompt buried
383    /// deeper than the cap is reported as "none recent", not searched for.
384    #[test]
385    fn only_the_tail_is_read_and_a_prompt_beyond_the_cap_is_not_found() {
386        let fs = FileStore::open_in_memory_with_cap(64 * 1024 * 1024).unwrap();
387        let filler = format!("{}\n", delta());
388        let mut body = Vec::new();
389        body.extend_from_slice(user("buried far back").as_bytes());
390        body.push(b'\n');
391        while (body.len() as i64) < MAX_SCAN + 3 * CHUNK {
392            body.extend_from_slice(filler.as_bytes());
393        }
394        put(&fs, "agent:agent-6:current", &body);
395        match read(&fs, "agent-6") {
396            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, ""),
397            other => panic!("expected a transcript, got {other:?}"),
398        }
399        // A recent prompt at the very end of the same file is found.
400        let mut body2 = body.clone();
401        body2.extend_from_slice(user("asked just now").as_bytes());
402        put(&fs, "agent:agent-7:current", &body2);
403        match read(&fs, "agent-7") {
404            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "asked just now"),
405            other => panic!("expected a transcript, got {other:?}"),
406        }
407    }
408
409    /// One filler line of exactly `n` bytes, newline included.
410    fn filler_of(n: usize) -> Vec<u8> {
411        let head = br#"{"type":"stream_event","p":""#;
412        let tail = b"\"}\n";
413        let mut v = head.to_vec();
414        v.extend(std::iter::repeat(b'x').take(n - head.len() - tail.len()));
415        v.extend_from_slice(tail);
416        v
417    }
418
419    /// A prompt line that straddles a chunk boundary is reassembled, not lost.
420    #[test]
421    fn a_prompt_straddling_a_chunk_boundary_is_found() {
422        let fs = FileStore::open_in_memory_with_cap(64 * 1024 * 1024).unwrap();
423        let prompt = user("split across two chunks");
424        let mut body = Vec::new();
425        body.extend(filler_of(CHUNK as usize));
426        body.extend_from_slice(prompt.as_bytes());
427        body.push(b'\n');
428        // The last chunk starts CHUNK bytes from the end, so this much after the
429        // prompt puts that boundary 20 bytes into the prompt's own line.
430        let after = CHUNK as usize - 20;
431        body.extend(filler_of(after));
432        put(&fs, "agent:agent-8:current", &body);
433        match read(&fs, "agent-8") {
434            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "split across two chunks"),
435            other => panic!("expected a transcript, got {other:?}"),
436        }
437    }
438
439    /// Manual check against a real transcript store (read-only): times a cold
440    /// read of every agent in it.
441    /// `AGENTMUX_HISTORY_REAL_STORE=<path to filestore.db> cargo test -p agentmux-srv --bins -- --ignored --nocapture history_real_store`
442    #[test]
443    #[ignore]
444    fn history_real_store_timing() {
445        let Ok(path) = std::env::var("AGENTMUX_HISTORY_REAL_STORE") else { return };
446        let fs = FileStore::open_read_only(std::path::Path::new(&path)).unwrap();
447        let ids: Vec<String> = fs
448            .zone_ids_with_prefix("agent:")
449            .unwrap()
450            .into_iter()
451            .filter_map(|z| z.strip_suffix(":current").and_then(|z| z.strip_prefix("agent:")).map(String::from))
452            .collect();
453        let t = std::time::Instant::now();
454        let (mut found, mut empty, mut none, mut failed) = (0, 0, 0, 0);
455        for id in &ids {
456            match read_agent_history(&fs, id) {
457                HistoryRead::Transcript(h) if h.last_user_message.is_empty() => empty += 1,
458                HistoryRead::Transcript(_) => found += 1,
459                HistoryRead::NoTranscript => none += 1,
460                HistoryRead::Failed => failed += 1,
461            }
462        }
463        println!("{} agents: {found} with a prompt, {empty} without, {none} no transcript, {failed} failed, cold in {:?}", ids.len(), t.elapsed());
464        let t = std::time::Instant::now();
465        for id in &ids {
466            let _ = read_agent_history(&fs, id);
467        }
468        println!("warm in {:?}", t.elapsed());
469    }
470
471    #[test]
472    fn a_changed_transcript_is_read_again() {
473        let fs = FileStore::open_in_memory().unwrap();
474        put(&fs, "agent:agent-9:current", user("one").as_bytes());
475        let first = read(&fs, "agent-9");
476        fs.append_data("agent:agent-9:current", OUTPUT_FILE, format!("\n{}", user("two")).as_bytes())
477            .unwrap();
478        let second = read(&fs, "agent-9");
479        assert_ne!(first, second);
480        match second {
481            HistoryRead::Transcript(h) => assert_eq!(h.last_user_message, "two"),
482            other => panic!("expected a transcript, got {other:?}"),
483        }
484    }
485}
agentmux_srv/backend/agent_session/history_summary.rs

agentmux_srv/backend/agent_session/
history_summary.rs
