artifactize_core/agent/
session.rs

1//! Saved Agent conversations: one append-only JSON Lines file per review session, which
2//! `session show` reads and `session send` continues.
3//!
4//! `$STATE/agent-sessions/` (owner-only) holds `<sessionId>.jsonl`, an owner-only file of
5//! one JSON event per line, and `<sessionId>.lock`, which serializes the sends to it:
6//!
7//! - `review`, first: the session, Run, request and eval ids, the producer and state id,
8//!   the backend, model, reasoning, provider parameters, budgets and tool definitions;
9//! - `message`: one rig [`Message`] exactly as the provider received or sent it, with the
10//!   `turn` (the provider request) that carried it. The system prompt and instructions come
11//!   first, then each answer with its reasoning (encrypted reasoning items and thinking
12//!   signatures included), tool calls and their results. A message of tool results has
13//!   `isError`, whether each result failed, in order. The repair prompt has `repair`, and
14//!   the answer it replaced is the message before it;
15//! - `attempt`: one provider attempt of a `turn`, its `usage` counters and any error;
16//! - `end`: the review's `result`, or its `errorCode` and `error`;
17//! - `send` and `answer` around each follow-up, whose events carry its `send` number.
18//!
19//! Every event has its time in `at`. Writing never fails a review: an error is reported on
20//! stderr and the review goes on without its conversation.
21
22pub mod document;
23mod events;
24mod gc;
25pub mod live;
26mod outcome;
27mod pages;
28#[cfg(test)]
29mod pages_tests;
30pub mod transcript;
31#[cfg(test)]
32mod transcript_tests;
33pub use events::{
34    Answer, Budgets, Delivery, DeliveryKind, DeliveryState, End, Event, Header, Kind, MessageEvent,
35    Send,
36};
37#[cfg(test)]
38mod recorder_tests;
39mod summary;
40
41pub use gc::{Collection, Usage, collect, usage};
42pub use summary::{Summary, tokens_text};
43
44use std::{
45    fmt,
46    fs::File,
47    io::{BufRead, BufReader, Write},
48    path::{Path, PathBuf},
49};
50
51use rig_core::message::{AssistantContent, Message, ToolResultContent, UserContent};
52use serde::{Deserialize, Serialize};
53use serde_json::Value;
54#[cfg(test)]
55use serde_json::json;
56
57use crate::{platform, store::Request};
58
59/// The session store below the state directory.
60pub const DIRECTORY: &str = "agent-sessions";
61/// The version of the session file's events.
62const VERSION: u32 = 1;
63
64/// Where an Agent review's conversation lives: the producing `user@host`, the state (by its
65/// stable id), and the Run, request and session ids. Its text form is
66/// `user@host/<state>/<runId>/<requestId>/<sessionId>`.
67#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
68#[serde(rename_all = "camelCase")]
69pub struct SessionRef {
70    pub producer: crate::types::ProducerId,
71    pub state: crate::types::StateId,
72    pub run_id: crate::types::RunId,
73    pub request_id: crate::types::RequestId,
74    pub session_id: crate::types::SessionId,
75}
76
77impl fmt::Display for SessionRef {
78    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
79        write!(
80            formatter,
81            "{}/{}/{}/{}/{}",
82            self.producer, self.state, self.run_id, self.request_id, self.session_id
83        )
84    }
85}
86
87impl SessionRef {
88    /// The text form; `None` for anything else, such as a bare request or session id.
89    pub fn parse(text: &str) -> Option<Self> {
90        let mut parts = text.rsplitn(5, '/');
91        let (session_id, request_id, run_id, state, producer) = (
92            parts.next()?,
93            parts.next()?,
94            parts.next()?,
95            parts.next()?,
96            parts.next()?,
97        );
98        let reference = Self {
99            producer: producer.parse().ok()?,
100            state: state.parse().ok()?,
101            run_id: run_id.parse().ok()?,
102            request_id: request_id.parse().ok()?,
103            session_id: session_id.parse().ok()?,
104        };
105        reference.valid().then_some(reference)
106    }
107
108    /// Bounded, printable fields; the ids are single path segments. A reference from a remote
109    /// record is checked before it is kept.
110    pub fn valid(&self) -> bool {
111        let printable = |value: &str| {
112            (1..=crate::types::MAX_ID_BYTES).contains(&value.len())
113                && !value.chars().any(char::is_control)
114        };
115        [
116            self.state.as_str(),
117            self.run_id.as_str(),
118            self.request_id.as_str(),
119            self.session_id.as_str(),
120        ]
121        .into_iter()
122        .all(|id| printable(id) && valid_id(id))
123    }
124}
125
126/// A session or state id: letters, digits, `-`, `_` and `.`, never only dots.
127fn valid_id(id: &str) -> bool {
128    id.parse::<crate::types::SessionId>().is_ok()
129}
130
131/// The session store of a state.
132pub fn directory(state: &Path) -> PathBuf {
133    state.join(DIRECTORY)
134}
135
136/// The conversation file of a session id.
137pub fn path(state: &Path, id: &str) -> Result<PathBuf, String> {
138    let id: crate::types::SessionId = id.parse()?;
139    Ok(directory(state).join(format!("{id}.jsonl")))
140}
141
142/// Where new conversations are saved, and as which producer and state; `None` when saving is
143/// off (`agentSessions.enabled: false` in `limits.json`).
144#[derive(Debug, Clone)]
145pub struct Saving {
146    pub state: PathBuf,
147    pub state_id: crate::types::StateId,
148    pub producer: crate::types::ProducerId,
149}
150
151fn warn(message: &str) {
152    let _ = writeln!(std::io::stderr().lock(), "{message}");
153}
154
155/// Appends one session's events. It writes nothing when saving is off, and after a failed
156/// write it reports the error once and writes nothing more.
157pub struct Recorder {
158    id: crate::types::SessionId,
159    state: Recording,
160    publisher: crate::changes::Publisher,
161}
162
163/// Only a successfully flushed review owns a saved reference. Follow-ups have a send number
164/// instead; a later write failure retains the review's reference but drops its file.
165enum Recording {
166    Off,
167    Pending(Box<Pending>),
168    Review { file: File, reference: SessionRef },
169    FollowUp { file: File, send: usize },
170    Failed,
171    SavedFailure(SessionRef),
172}
173struct Pending {
174    path: PathBuf,
175    reference: SessionRef,
176    eval_id: Option<crate::types::EvalId>,
177    target: Option<crate::types::ArtifactName>,
178}
179
180impl Recorder {
181    /// A session that is not saved.
182    pub fn off(id: &crate::types::SessionId) -> Self {
183        Self {
184            id: id.clone(),
185            state: Recording::Off,
186            publisher: crate::changes::Publisher::default(),
187        }
188    }
189
190    /// The recorder of a request's new review session `id`. No file is created until start.
191    pub fn new(saving: Option<&Saving>, request: &Request, id: &crate::types::SessionId) -> Self {
192        let Some(saving) = saving else {
193            return Self::off(id);
194        };
195        let path = match path(&saving.state, id) {
196            Ok(path) => path,
197            Err(error) => {
198                warn(&error);
199                return Self::off(id);
200            }
201        };
202        Self {
203            id: id.clone(),
204            state: Recording::Pending(Box::new(Pending {
205                path,
206                reference: SessionRef {
207                    producer: saving.producer.clone(),
208                    state: saving.state_id.clone(),
209                    run_id: request.run_id.clone(),
210                    request_id: request.id.clone(),
211                    session_id: id.clone(),
212                },
213                eval_id: Some(request.eval_id.clone()),
214                target: Some(request.target.clone()),
215            })),
216            publisher: crate::changes::Publisher::new(&saving.state),
217        }
218    }
219
220    /// A session saved below `state` for no Run.
221    #[cfg(test)]
222    pub(crate) fn test(state: &Path, id: &str) -> Self {
223        let id: crate::types::SessionId = id.parse().unwrap();
224        Self {
225            state: Recording::Pending(Box::new(Pending {
226                path: path(state, &id).unwrap(),
227                reference: SessionRef {
228                    producer: "tester@host".parse().unwrap(),
229                    state: "state".parse().unwrap(),
230                    run_id: "run".parse().unwrap(),
231                    request_id: "request".parse().unwrap(),
232                    session_id: id.clone(),
233                },
234                eval_id: None,
235                target: None,
236            })),
237            ..Self::off(&id)
238        }
239    }
240
241    /// Append follow-up `send` to the saved conversation at `path` without a new review header.
242    pub(crate) fn append(
243        state: &Path,
244        path: &Path,
245        id: &crate::types::SessionId,
246        send: usize,
247    ) -> Result<Self, String> {
248        let file = platform::open_no_follow(File::options().append(true), path)
249            .map_err(|error| format!("Cannot open Agent session {id}: {error}"))?;
250        Ok(Self {
251            id: id.clone(),
252            state: Recording::FollowUp { file, send },
253            publisher: crate::changes::Publisher::new(state),
254        })
255    }
256
257    pub fn id(&self) -> &crate::types::SessionId {
258        &self.id
259    }
260
261    /// The saved review's reference, after its header flush, even if a later append fails.
262    pub fn reference(&self) -> Option<&SessionRef> {
263        match &self.state {
264            Recording::Review { reference, .. } | Recording::SavedFailure(reference) => {
265                Some(reference)
266            }
267            Recording::Off
268            | Recording::Pending(_)
269            | Recording::FollowUp { .. }
270            | Recording::Failed => None,
271        }
272    }
273
274    /// Create the session file with its `review` event: the identity, then `details`.
275    pub fn start(&mut self, mut details: Header) {
276        let state = std::mem::replace(&mut self.state, Recording::Off);
277        let pending = match state {
278            Recording::Pending(pending) => pending,
279            other => {
280                self.state = other;
281                return;
282            }
283        };
284        let reference = pending.reference;
285        details.version = Some(VERSION);
286        details.producer = Some(reference.producer.clone());
287        details.state = Some(reference.state.clone());
288        details.run_id = Some(reference.run_id.clone());
289        details.request_id = Some(reference.request_id.clone());
290        details.session_id = Some(reference.session_id.clone());
291        details.eval_id = pending.eval_id;
292        details.target = pending.target;
293        match create(&pending.path) {
294            Ok(file) => self.start_file(file, reference, details),
295            Err(error) => self.fail(&error),
296        }
297    }
298
299    /// Promote the reference only after the initial header has been flushed.
300    fn start_file(&mut self, mut file: File, reference: SessionRef, details: Header) {
301        match write_event(&mut file, None, Kind::Review(Box::new(details))) {
302            Ok(()) => {
303                self.state = Recording::Review { file, reference };
304                self.notify();
305            }
306            Err(error) => self.fail(&error.to_string()),
307        }
308    }
309
310    fn active(&self) -> bool {
311        matches!(
312            self.state,
313            Recording::Review { .. } | Recording::FollowUp { .. }
314        )
315    }
316
317    /// One message of `turn`; `repair` marks the prompt that asked for the verdict again.
318    pub(crate) fn message(&mut self, turn: usize, message: &Message, repair: bool) {
319        if !self.active() {
320            return;
321        }
322        self.event(Kind::Message(MessageEvent {
323            turn,
324            message: message.clone(),
325            repair,
326            is_error: Vec::new(),
327            question: None,
328        }));
329    }
330
331    /// The message of tool results that `turn` sends, with whether each failed.
332    pub(crate) fn tool_results(&mut self, turn: usize, message: &Message, failed: &[bool]) {
333        if !self.active() {
334            return;
335        }
336        self.event(Kind::Message(MessageEvent {
337            turn,
338            message: message.clone(),
339            repair: false,
340            is_error: failed.to_vec(),
341            question: None,
342        }));
343    }
344
345    /// Append an event, stamped with its time and the follow-up it belongs to. End events do
346    /// not close the writer; the caller owns its lifetime and session GC scheduling.
347    pub fn event(&mut self, kind: Kind) {
348        let result = match &mut self.state {
349            Recording::Review { file, .. } => write_event(file, None, kind),
350            Recording::FollowUp { file, send } => write_event(file, Some(*send), kind),
351            Recording::Off
352            | Recording::Pending(_)
353            | Recording::Failed
354            | Recording::SavedFailure(_) => return,
355        };
356        match result {
357            Ok(()) => self.notify(),
358            Err(error) => self.fail(&error.to_string()),
359        }
360    }
361
362    fn notify(&self) {
363        self.publisher
364            .notify(crate::changes::Change::SessionInvalidated(self.id.clone()));
365    }
366
367    fn fail(&mut self, error: &str) {
368        self.state = match std::mem::replace(&mut self.state, Recording::Failed) {
369            Recording::Review { reference, .. } | Recording::SavedFailure(reference) => {
370                Recording::SavedFailure(reference)
371            }
372            Recording::Off
373            | Recording::Pending(_)
374            | Recording::FollowUp { .. }
375            | Recording::Failed => Recording::Failed,
376        };
377        warn(&format!(
378            "Cannot save Agent session {}: {error}; the review goes on without saving it.",
379            self.id
380        ));
381    }
382}
383
384/// A successful flush is the single write boundary for both headers and later events.
385fn write_event(file: &mut File, send: Option<usize>, kind: Kind) -> std::io::Result<()> {
386    let event = Event {
387        kind,
388        send,
389        at: Some(crate::broker::now()),
390    };
391    let mut line = serde_json::to_vec(&event).expect("session events are JSON");
392    line.push(b'\n');
393    file.write_all(&line).and_then(|()| file.flush())
394}
395
396/// A new owner-only session file in an owner-only session store.
397fn create(path: &Path) -> Result<File, String> {
398    let directory = path.parent().expect("a session file has a directory");
399    platform::create_private_dir_all(directory).map_err(|e| e.to_string())?;
400    if !platform::is_private_dir(directory).map_err(|e| e.to_string())? {
401        return Err(format!(
402            "{} must have owner-only permissions.",
403            crate::platform::path_text(directory)
404        ));
405    }
406    let file = platform::open_no_follow(
407        platform::private_options().append(true).create_new(true),
408        path,
409    )
410    .map_err(|e| e.to_string())?;
411    platform::restrict_file(&file).map_err(|e| e.to_string())?;
412    Ok(file)
413}
414
415/// A saved conversation, read whole.
416#[derive(Debug)]
417pub struct Conversation {
418    pub path: PathBuf,
419    pub events: Vec<Event>,
420    /// Original events for the lossless --json output; computations use events.
421    pub wire_events: Vec<Value>,
422}
423
424impl Conversation {
425    /// Read a session file; `None` when it does not exist. A last line cut short by a crash
426    /// is left out.
427    pub fn load(path: &Path) -> Result<Option<Self>, String> {
428        let file = match platform::open_no_follow(File::options().read(true), path) {
429            Ok(file) => file,
430            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
431            Err(error) => {
432                return Err(format!(
433                    "Cannot read {}: {error}",
434                    crate::platform::path_text(path)
435                ));
436            }
437        };
438        if !file.metadata().map_err(|e| e.to_string())?.is_file() {
439            return Err(format!(
440                "{} is not a regular file.",
441                crate::platform::path_text(path)
442            ));
443        }
444        let lines: Vec<String> = BufReader::new(file)
445            .lines()
446            .collect::<Result<_, _>>()
447            .map_err(|e| format!("Cannot read {}: {e}", crate::platform::path_text(path)))?;
448        let count = lines.len();
449        let mut events = Vec::new();
450        let mut wire_events = Vec::new();
451        for (index, line) in lines.into_iter().enumerate() {
452            match serde_json::from_str::<Value>(&line) {
453                Ok(value) => {
454                    let event: Event = serde_json::from_value(value.clone()).map_err(|error| {
455                        format!(
456                            "{} line {} is not a session event: {error}",
457                            crate::platform::path_text(path),
458                            index + 1
459                        )
460                    })?;
461                    wire_events.push(value);
462                    events.push(event);
463                }
464                Err(error) if error.is_eof() && index + 1 == count => {}
465                Err(_) => {
466                    return Err(format!(
467                        "{} line {} is not a session event.",
468                        crate::platform::path_text(path),
469                        index + 1
470                    ));
471                }
472            }
473        }
474        if events
475            .first()
476            .is_none_or(|event| !matches!(event.kind, Kind::Review(_)))
477        {
478            return Err(format!(
479                "{} has no review event.",
480                crate::platform::path_text(path)
481            ));
482        }
483        Ok(Some(Self {
484            path: path.into(),
485            events,
486            wire_events,
487        }))
488    }
489
490    /// The `review` event.
491    pub fn header(&self) -> &Header {
492        let Kind::Review(header) = &self.events[0].kind else {
493            unreachable!("load requires a review header")
494        };
495        header
496    }
497
498    /// How many follow-ups were sent.
499    pub fn sends(&self) -> usize {
500        self.events
501            .iter()
502            .filter(|event| matches!(event.kind, Kind::Send(_)))
503            .count()
504    }
505
506    /// Every message in order, the review's and its follow-ups', ready to send again. Tool
507    /// calls a stopped review never answered get an error result next to the results of
508    /// those that ran, since a provider expects every call answered.
509    pub fn history(&self) -> Result<Vec<Message>, String> {
510        let mut messages = self
511            .events
512            .iter()
513            .filter_map(|event| match &event.kind {
514                Kind::Message(event) => Some(event.message.clone()),
515                _ => None,
516            })
517            .collect::<Vec<_>>();
518        let Some(last) = messages
519            .iter()
520            .rposition(|message| matches!(message, Message::Assistant { .. }))
521        else {
522            return Ok(messages);
523        };
524        let answered: Vec<_> = messages[last + 1..]
525            .iter()
526            .flat_map(|message| match message {
527                Message::User { content } => content.as_slice(),
528                _ => &[],
529            })
530            .filter_map(|part| match part {
531                UserContent::ToolResult(result) => Some(result.call.wire()),
532                _ => None,
533            })
534            .collect();
535        let Message::Assistant { content, .. } = &messages[last] else {
536            unreachable!("the last answer")
537        };
538        let missing: Vec<_> = content
539            .iter()
540            .filter_map(|part| match part {
541                AssistantContent::ToolCall(call) if !answered.contains(&call.id.wire()) => {
542                    Some(UserContent::ToolResult(call.result(vec![
543                        ToolResultContent::text(
544                            "Not run: the review stopped before this tool call completed.",
545                        ),
546                    ])))
547                }
548                _ => None,
549            })
550            .collect();
551        if missing.is_empty() {
552            return Ok(messages);
553        }
554        let ran = last + 1 < messages.len();
555        match messages.last_mut() {
556            Some(Message::User { content }) if ran => content.extend(missing),
557            _ => messages.push(Message::User { content: missing }),
558        }
559        Ok(messages)
560    }
561}
562
563/// Hold `<sessionId>.lock` until the returned file drops: sends to one session take turns.
564pub async fn lock(state: &Path, id: &str) -> Result<File, String> {
565    let path = path(state, id)?.with_extension("lock");
566    let file = platform::open_no_follow(
567        platform::private_options()
568            .read(true)
569            .write(true)
570            .create(true)
571            .truncate(false),
572        &path,
573    )
574    .map_err(|e| format!("Cannot lock Agent session {id}: {e}"))?;
575    loop {
576        match file.try_lock() {
577            Ok(()) => return Ok(file),
578            Err(std::fs::TryLockError::WouldBlock) => {
579                tokio::time::sleep(platform::FILE_LOCK_RETRY_INTERVAL).await;
580            }
581            Err(std::fs::TryLockError::Error(error)) => {
582                return Err(format!("Cannot lock Agent session {id}: {error}"));
583            }
584        }
585    }
586}
587
588#[cfg(test)]
589pub(crate) mod live_tests;
590#[cfg(test)]
591mod tests {
592    use super::*;
593    use std::time::Duration;
594
595    fn reference() -> SessionRef {
596        SessionRef {
597            producer: "alice@laptop".parse().unwrap(),
598            state: "6f1c0a52-8d1e-4c43-9a55-31f6a7f2d0e4".parse().unwrap(),
599            run_id: "run-Hq2b9X".parse().unwrap(),
600            request_id: "run-Hq2b9X-3".parse().unwrap(),
601            session_id: "0e8c7c2e-2a49-4b8e-9f7a-5d7a0c3b1f20".parse().unwrap(),
602        }
603    }
604
605    #[test]
606    fn producer_identity_validates_at_the_session_wire_boundary() {
607        let valid = serde_json::to_value(reference()).unwrap();
608        for producer in [
609            String::new(),
610            "bad\nproducer".into(),
611            "x".repeat(201),
612            "é".repeat(101),
613        ] {
614            let mut wire = valid.clone();
615            wire["producer"] = json!(producer);
616            assert!(serde_json::from_value::<SessionRef>(wire).is_err());
617        }
618        for producer in [
619            "name/with/slash@host".to_owned(),
620            "x".repeat(200),
621            "é".repeat(100),
622        ] {
623            let mut wire = valid.clone();
624            wire["producer"] = json!(producer);
625            let reference: SessionRef = serde_json::from_value(wire.clone()).unwrap();
626            assert_eq!(serde_json::to_value(reference).unwrap(), wire);
627        }
628    }
629
630    #[test]
631    fn references_round_trip_through_their_text_form() {
632        let reference = reference();
633        let text = reference.to_string();
634        assert_eq!(
635            text,
636            "alice@laptop/6f1c0a52-8d1e-4c43-9a55-31f6a7f2d0e4/run-Hq2b9X/run-Hq2b9X-3/0e8c7c2e-2a49-4b8e-9f7a-5d7a0c3b1f20"
637        );
638        assert_eq!(SessionRef::parse(&text), Some(reference));
639        // A producer may hold a slash; the ids never do.
640        let odd = SessionRef {
641            producer: "a/b@host".parse().unwrap(),
642            ..self::reference()
643        };
644        assert_eq!(SessionRef::parse(&odd.to_string()), Some(odd));
645        for text in [
646            "run-Hq2b9X-3",
647            "0e8c7c2e-2a49-4b8e-9f7a-5d7a0c3b1f20",
648            "a/b/c/d",
649            "alice@laptop/state/run/../session",
650            "alice@laptop/state/run/request/",
651            "alice@laptop/st ate/run/request/session",
652        ] {
653            assert_eq!(SessionRef::parse(text), None, "{text}");
654        }
655        let root = crate::test_os::tempdir();
656        assert!(path(root.path(), "..").is_err());
657        assert!(path(root.path(), "a/b").is_err());
658    }
659
660    #[test]
661    fn calls_a_stopped_review_never_answered_are_answered_on_replay() {
662        use rig_core::message::{ToolCall, ToolFunction, ToolName};
663        let call = |id: &str| {
664            ToolCall::from_wire(
665                id,
666                ToolFunction::new(ToolName::new("read").unwrap(), json!({})),
667            )
668        };
669        let answer = Message::Assistant {
670            id: None,
671            content: vec![
672                AssistantContent::ToolCall(call("c1")),
673                AssistantContent::ToolCall(call("c2")),
674            ],
675        };
676        let ran = Message::tool_results(vec![
677            call("c1").result(vec![ToolResultContent::text("ran")]),
678        ]);
679        let replay = |messages: &[&Message]| {
680            let events: Vec<Value> = std::iter::once(json!({"kind":"review"}))
681                .chain(
682                    messages
683                        .iter()
684                        .map(|message| json!({"kind":"message","message":message})),
685                )
686                .collect();
687            Conversation {
688                path: "session.jsonl".into(),
689                wire_events: Vec::new(),
690                events: events
691                    .into_iter()
692                    .map(|event| serde_json::from_value(event).unwrap())
693                    .collect(),
694            }
695            .history()
696            .unwrap()
697        };
698        let results = |message: &Message| match message {
699            Message::User { content } => content
700                .iter()
701                .map(|part| match part {
702                    UserContent::ToolResult(result) => {
703                        (result.call.wire().into_owned(), json!(result.content))
704                    }
705                    _ => panic!("only tool results"),
706                })
707                .collect::<Vec<_>>(),
708            _ => panic!("a user message"),
709        };
710        let not_run = json!([
711            {"type":"text","text":"Not run: the review stopped before this tool call completed."},
712        ]);
713        // The budget stopped the second call: it is answered next to the first one's result.
714        let history = replay(&[&answer, &ran]);
715        assert_eq!(history.len(), 2);
716        assert_eq!(
717            results(&history[1]),
718            [
719                ("c1".to_owned(), json!([{"type":"text","text":"ran"}])),
720                ("c2".to_owned(), not_run.clone())
721            ]
722        );
723        // No call ran: every call is answered in a new message.
724        let history = replay(&[&answer]);
725        assert_eq!(history.len(), 2);
726        assert_eq!(
727            results(&history[1]),
728            [
729                ("c1".to_owned(), not_run.clone()),
730                ("c2".to_owned(), not_run)
731            ]
732        );
733    }
734
735    #[test]
736    fn a_session_that_cannot_be_written_is_skipped() {
737        let root = crate::test_os::tempdir();
738        // The store's place is taken by a file, so the session cannot be created.
739        let state = root.path().join("state");
740        std::fs::create_dir(&state).unwrap();
741        std::fs::write(directory(&state), "not a directory").unwrap();
742        let reference = reference();
743        let request = crate::monitor::tests::request(
744            "app/check",
745            "RUNNING",
746            json!({"profile":{"kind":"agent","backend":"openai","model":"fixture"}}),
747        );
748        let saving = Saving {
749            state: state.clone(),
750            state_id: reference.state,
751            producer: reference.producer,
752        };
753        let mut recorder = Recorder::new(Some(&saving), &request.request, &reference.session_id);
754        recorder.start(Header {
755            backend: Some(crate::config::Backend::Openai),
756            ..Header::default()
757        });
758        recorder.message(1, &Message::user("still recorded nowhere"), false);
759        recorder.event(Kind::End(End::default()));
760        assert!(recorder.reference().is_none());
761        assert!(std::fs::metadata(directory(&state)).unwrap().is_file());
762    }
763    #[test]
764    fn jsonl_edges_keep_wire_events_and_reject_invalid_complete_records() {
765        let root = crate::test_os::tempdir();
766        let file = root.path().join("legacy.jsonl");
767        let header = r#"{"kind":"review","version":1,"sessionId":"legacy.session","runId":"run-old","requestId":"run-old-1","backend":"openai","model":"fixture","reasoning":null,"parameters":{"vendor":true},"budgets":{"timeoutMs":1000,"maxToolCalls":null,"maxTokens":null},"tools":[],"unknownSavedField":7,"at":"2026-01-01T00:00:00Z"}"#;
768        let message = r#"{"kind":"message","turn":1,"message":{"role":"user","content":[{"type":"text","text":"hello"}]},"at":"2026-01-01T00:00:01Z"}"#;
769        std::fs::write(&file, format!("{header}\n{message}\n{{\"kind\":")).unwrap();
770        let conversation = Conversation::load(&file).unwrap().unwrap();
771        assert_eq!(conversation.events.len(), 2);
772        assert_eq!(
773            conversation.header().session_id.as_ref().unwrap().as_str(),
774            "legacy.session"
775        );
776        assert_eq!(
777            conversation.header().budgets.as_ref().unwrap().timeout_ms,
778            Some(Duration::from_millis(1000))
779        );
780        assert_eq!(
781            conversation.wire_events[0],
782            serde_json::from_str::<Value>(header).unwrap()
783        );
784        assert_eq!(
785            conversation.wire_events[1],
786            serde_json::from_str::<Value>(message).unwrap()
787        );
788        assert_eq!(conversation.history().unwrap().len(), 1);
789        for invalid in [
790            r#"{"kind":"message","message":17}"#,
791            r#"{"kind":"attempt","turn":"bad","attempt":1,"usage":{}}"#,
792            r#"{"kind":"unknown"}"#,
793        ] {
794            std::fs::write(&file, format!("{header}\n{invalid}\n")).unwrap();
795            assert!(Conversation::load(&file).is_err(), "{invalid}");
796        }
797        std::fs::write(&file, header.replace("legacy.session", "../escape")).unwrap();
798        assert!(Conversation::load(&file).is_err());
799    }
800}
artifactize_core/agent/session.rs

artifactize_core/agent/
session.rs
