agent_core/core/
session_draft.rs

1//! The in-flight turn draft: `sessions/<id>.turn`.
2//!
3//! The session snapshot is saved at every round boundary (the engine's round
4//! checkpoints), so it always holds a VALID history. What it cannot hold is
5//! the response still being streamed — partial text is not a valid history
6//! entry until the turn stops. This small draft carries it instead:
7//!
8//! - written when a turn starts (the "turn is open" signal), then at most
9//!   once a second while text streams, and removed when the turn ends;
10//! - `base_len` = how many messages of the saved history the in-flight
11//!   response follows, so a draft that outlived its round is recognised as
12//!   stale;
13//! - `partial_text` = the text the model has streamed so far in that
14//!   response (text only: unsigned thinking and unfinished tool calls are
15//!   never replayable).
16//!
17//! Only the holder of the session lock writes a draft, and only the holder
18//! reads one back: a draft found when the holder loads the session means the
19//! previous holder died with a turn open, and it is folded into history as a
20//! real assistant message plus an interruption marker
21//! (`agent_engine::engine::interrupt`). The session actor's background
22//! writer (`agent_engine::session::persister`) applies draft writes and
23//! removals in order with the snapshot saves: the draft is removed only
24//! after the history that ends its turn is saved.
25//!
26//! Separate from the snapshot on purpose: O(partial text) bytes per write in
27//! every persistence mode, and the snapshot bytes stay exactly those of the
28//! last completed round. Same confined, private (0600), atomic writes as the
29//! snapshot. The `.turn` name keeps the session id as the file stem and is
30//! invisible to `*.json` listings; deleting a session (and retention)
31//! removes its draft with it.
32
33use std::path::Path;
34
35use serde::{Deserialize, Serialize};
36
37/// Upper bound on persisted partial text (a response is bounded by the
38/// model's max output tokens; this only guards pathological streams).
39pub const TURN_DRAFT_MAX_TEXT_BYTES: usize = 1024 * 1024;
40
41/// Upper bound on a draft file read at load: the capped text JSON-escaped
42/// in the worst case (`\u00XX` = 6 bytes per input byte) plus framing. A
43/// larger file was not written by Synaps and is rejected unread.
44pub const TURN_DRAFT_MAX_FILE_BYTES: u64 = 6 * TURN_DRAFT_MAX_TEXT_BYTES as u64 + 4096;
45
46#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
47pub struct TurnDraft {
48    /// Messages in the saved history the in-flight response continues from.
49    pub base_len: usize,
50    /// Text streamed so far in the in-flight response.
51    #[serde(default)]
52    pub partial_text: String,
53}
54
55fn artifact(id: &str) -> String {
56    format!("{id}.turn")
57}
58
59/// Read the draft for `id`. `Ok(None)` when absent (the normal case); an
60/// `InvalidData` error for an oversized (`TURN_DRAFT_MAX_FILE_BYTES`) or
61/// malformed file.
62pub fn read_turn_draft(dir: &Path, id: &str) -> std::io::Result<Option<TurnDraft>> {
63    let Some(bytes) = read_artifact(dir, &artifact(id))? else {
64        return Ok(None);
65    };
66    if bytes.len() as u64 > TURN_DRAFT_MAX_FILE_BYTES {
67        return Err(std::io::Error::new(
68            std::io::ErrorKind::InvalidData,
69            "turn draft exceeds its size limit",
70        ));
71    }
72    serde_json::from_slice(&bytes)
73        .map(Some)
74        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
75}
76
77/// Atomically (re)write the draft for `id` (0600, confined).
78pub fn write_turn_draft(dir: &Path, id: &str, draft: &TurnDraft) -> std::io::Result<()> {
79    let mut draft = draft.clone();
80    truncate_on_char_boundary(&mut draft.partial_text, TURN_DRAFT_MAX_TEXT_BYTES);
81    let bytes = serde_json::to_vec(&draft).map_err(std::io::Error::other)?;
82    write_artifact(dir, &artifact(id), &bytes)
83}
84
85/// Remove the draft for `id`. Idempotent.
86pub fn remove_turn_draft(dir: &Path, id: &str) -> std::io::Result<()> {
87    match remove_artifact(dir, &artifact(id)) {
88        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
89        other => other,
90    }
91}
92
93fn truncate_on_char_boundary(s: &mut String, max: usize) {
94    if s.len() > max {
95        let mut cut = max;
96        while !s.is_char_boundary(cut) {
97            cut -= 1;
98        }
99        s.truncate(cut);
100    }
101}
102
103// ── confined I/O (mirrors session_journal's handle-relative artifacts) ──────
104
105#[cfg(unix)]
106fn read_artifact(dir: &Path, name: &str) -> std::io::Result<Option<Vec<u8>>> {
107    use std::io::Read;
108    let handle = match crate::core::private_fs::ConfinedDir::open_absolute_no_symlinks(dir) {
109        Ok(h) => h,
110        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
111        Err(e) => return Err(e),
112    };
113    let file = match handle.open_file(&[name.to_string()]) {
114        Ok(f) => f,
115        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
116        Err(e) => return Err(e),
117    };
118    // Read at most one byte past the limit: enough to reject, never more.
119    let mut bytes = Vec::new();
120    file.take(TURN_DRAFT_MAX_FILE_BYTES + 1)
121        .read_to_end(&mut bytes)?;
122    Ok(Some(bytes))
123}
124
125#[cfg(unix)]
126fn write_artifact(dir: &Path, name: &str, data: &[u8]) -> std::io::Result<()> {
127    crate::core::private_fs::ConfinedDir::create_absolute_no_symlinks(dir)?.write_atomic(name, data)
128}
129
130#[cfg(unix)]
131fn remove_artifact(dir: &Path, name: &str) -> std::io::Result<()> {
132    match crate::core::private_fs::ConfinedDir::open_absolute_no_symlinks(dir) {
133        Ok(handle) => handle.remove_file(name),
134        Err(e) => Err(e),
135    }
136}
137
138#[cfg(not(unix))]
139fn read_artifact(dir: &Path, name: &str) -> std::io::Result<Option<Vec<u8>>> {
140    let path = dir.join(name);
141    match std::fs::symlink_metadata(&path) {
142        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
143        Ok(meta) if meta.file_type().is_symlink() => Err(std::io::Error::other(format!(
144            "refusing symlinked session artifact {name:?}"
145        ))),
146        Err(e) => Err(e),
147        Ok(_) => {
148            use std::io::Read;
149            let mut bytes = Vec::new();
150            std::fs::File::open(&path)?
151                .take(TURN_DRAFT_MAX_FILE_BYTES + 1)
152                .read_to_end(&mut bytes)?;
153            Ok(Some(bytes))
154        }
155    }
156}
157
158#[cfg(not(unix))]
159fn write_artifact(dir: &Path, name: &str, data: &[u8]) -> std::io::Result<()> {
160    crate::core::private_fs::ensure_private_dir(dir).map_err(std::io::Error::other)?;
161    crate::core::private_fs::write_atomic_private(&dir.join(name), data)
162        .map_err(std::io::Error::other)
163}
164
165#[cfg(not(unix))]
166fn remove_artifact(dir: &Path, name: &str) -> std::io::Result<()> {
167    std::fs::remove_file(dir.join(name))
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173
174    fn draft(base_len: usize, text: &str) -> TurnDraft {
175        TurnDraft {
176            base_len,
177            partial_text: text.into(),
178        }
179    }
180
181    #[test]
182    fn roundtrip_absent_and_idempotent_remove() {
183        let tmp = tempfile::tempdir().unwrap();
184        let dir = tmp.path().join("sessions");
185        assert_eq!(read_turn_draft(&dir, "s1").unwrap(), None, "no dir yet");
186        write_turn_draft(&dir, "s1", &draft(3, "partial")).unwrap();
187        assert_eq!(
188            read_turn_draft(&dir, "s1").unwrap(),
189            Some(draft(3, "partial"))
190        );
191        assert!(dir.join("s1.turn").exists());
192        remove_turn_draft(&dir, "s1").unwrap();
193        remove_turn_draft(&dir, "s1").unwrap();
194        assert_eq!(read_turn_draft(&dir, "s1").unwrap(), None);
195    }
196
197    #[cfg(unix)]
198    #[test]
199    fn draft_file_is_private() {
200        use std::os::unix::fs::PermissionsExt;
201        let tmp = tempfile::tempdir().unwrap();
202        let dir = tmp.path().join("sessions");
203        write_turn_draft(&dir, "s1", &draft(0, "x")).unwrap();
204        let mode = std::fs::metadata(dir.join("s1.turn"))
205            .unwrap()
206            .permissions()
207            .mode();
208        assert_eq!(mode & 0o777, 0o600);
209    }
210
211    #[test]
212    fn oversized_text_is_truncated_on_a_char_boundary() {
213        let tmp = tempfile::tempdir().unwrap();
214        let dir = tmp.path().join("sessions");
215        let big = "é".repeat(TURN_DRAFT_MAX_TEXT_BYTES); // 2 bytes each
216        write_turn_draft(&dir, "s1", &draft(1, &big)).unwrap();
217        let back = read_turn_draft(&dir, "s1").unwrap().unwrap();
218        assert!(back.partial_text.len() <= TURN_DRAFT_MAX_TEXT_BYTES);
219        assert!(back.partial_text.chars().all(|c| c == 'é'));
220    }
221
222    #[test]
223    fn corrupt_draft_is_an_error_not_a_panic() {
224        let tmp = tempfile::tempdir().unwrap();
225        let dir = tmp.path().join("sessions");
226        write_turn_draft(&dir, "s1", &draft(0, "")).unwrap();
227        std::fs::write(dir.join("s1.turn"), b"{not json").unwrap();
228        assert!(read_turn_draft(&dir, "s1").is_err());
229    }
230
231    /// The largest draft Synaps can write reads back; anything bigger is
232    /// rejected without reading it whole.
233    #[test]
234    fn draft_reads_are_bounded() {
235        let tmp = tempfile::tempdir().unwrap();
236        let dir = tmp.path().join("sessions");
237        // Worst-case escaping: every byte a control character.
238        let worst = "\u{1}".repeat(TURN_DRAFT_MAX_TEXT_BYTES);
239        write_turn_draft(&dir, "s1", &draft(0, &worst)).unwrap();
240        assert!(std::fs::metadata(dir.join("s1.turn")).unwrap().len() <= TURN_DRAFT_MAX_FILE_BYTES);
241        assert_eq!(read_turn_draft(&dir, "s1").unwrap().unwrap().partial_text, worst);
242
243        let huge = vec![b' '; TURN_DRAFT_MAX_FILE_BYTES as usize + 1];
244        std::fs::write(dir.join("s2.turn"), huge).unwrap();
245        let err = read_turn_draft(&dir, "s2").unwrap_err();
246        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
247    }
248}
agent_core/core/session_draft.rs

agent_core/core/
session_draft.rs
