Skip to main content

supercode_harness/
session_journal.rs

1//! BP-8 (catalog domain 5): the **append-only session journal** — one
2//! flush-per-record log of everything that happens to a session while it is
3//! live, written beside the transcript as `<name>.journal.jsonl`.
4//!
5//! The problem it solves. supercode's own store persists a session by
6//! REWRITING `<name>.jsonl` from the agent's in-memory history at the end of
7//! a turn (`SessionStore::save`). That is durable but not append-only: a
8//! crash between the first tool call and the end of the turn loses the whole
9//! turn. The genuine per-message writer with a flush on every line
10//! ([`crate::sidecar::SidecarWriter`]) exists, but it owns the native-v2
11//! SIDECAR file, whose presence means "this session is a reduced/imported
12//! family" to every resume door — so it cannot simply be switched on for an
13//! ordinary session without changing what that session IS.
14//!
15//! The journal is therefore a separate, additive family member with a
16//! discriminated record shape of its own. It is:
17//!
18//! * **append-only** — nothing in it is ever rewritten or truncated, so the
19//!   bytes of a message that a later rewind removed are still on disk;
20//! * **line-atomic** — full line + `\n`, then `flush()`, exactly
21//!   [`crate::sidecar::SidecarWriter::append`]'s guarantee, so a crash
22//!   mid-write can only tear the record being written, never one already
23//!   there, and [`replay_str`] skips a torn trailing line;
24//! * **replayable** — [`replay`] folds the log into the state it describes:
25//!   the live message list, the pending input queues, and the current plan;
26//! * **invertible** — every state-changing operation has an inverse that is
27//!   itself an appended record ([`JournalOp::Rewind`] ↔
28//!   [`JournalOp::Unrewind`]), so "undo" never means "delete a record".
29//!
30//! Records carry a `supercode_journal` discriminant for the same reason
31//! [`crate::sidecar::NativeTurn`] carries `supercode_turn`: neither Claude
32//! Code's nor Codex's own record shapes have that key, so a tolerant foreign
33//! loader that is ever pointed at this file skips these lines rather than
34//! erroring on them.
35
36use std::fs::{File, OpenOptions};
37use std::io::Write;
38use std::path::{Path, PathBuf};
39
40use serde::{Deserialize, Serialize};
41
42use crate::error::{Error, Result};
43use crate::sidecar::{now_rfc3339, NativeTurn};
44use crate::ChatMessage;
45
46/// Discriminant value every journal line carries.
47pub const JOURNAL_RECORD_VERSION: u8 = 1;
48
49/// Which of the agent's two input queues an [`JournalOp::Enqueue`] /
50/// [`JournalOp::Dequeue`] record is about (`crate::agent::Agent`'s
51/// `steer_queue` — mid-turn — and `follow_up_queue` — at idle).
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
53#[serde(rename_all = "snake_case")]
54pub enum QueueKind {
55    /// Mid-turn steering input (drained at the top of the next loop pass).
56    Steer,
57    /// At-idle follow-up input (drained when the loop would otherwise end).
58    FollowUp,
59}
60
61/// One step of a persisted plan (`update_plan`'s checklist).
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63pub struct PlanEntry {
64    /// The step text as the model wrote it.
65    pub step: String,
66    /// `pending` | `in_progress` | `completed`.
67    pub status: String,
68}
69
70/// What a journal line records.
71#[derive(Debug, Clone, Serialize, Deserialize)]
72#[serde(tag = "op", rename_all = "snake_case")]
73pub enum JournalOp {
74    /// One conversation message, appended the moment the agent produced it
75    /// — the append-only half of "every event flushed to disk during the
76    /// session".
77    Message {
78        /// The message, in the same full-fidelity wire shape the native-v2
79        /// sidecar uses (metadata included — see [`NativeTurn`]).
80        message: Box<NativeTurn>,
81    },
82    /// The conversation was rewound to `to` messages. Nothing is deleted:
83    /// [`replay_str`] moves the removed tail onto an undo stack that
84    /// [`JournalOp::Unrewind`] pops.
85    Rewind {
86        /// Message count kept (an index into the replayed message list).
87        to: usize,
88    },
89    /// The inverse of the most recent [`JournalOp::Rewind`]: the removed
90    /// tail comes back.
91    Unrewind,
92    /// A pending user input was queued.
93    Enqueue {
94        /// Which queue.
95        queue: QueueKind,
96        /// The queued text.
97        text: String,
98    },
99    /// `count` pending inputs were consumed from `queue` (drained into the
100    /// conversation), oldest first.
101    Dequeue {
102        /// Which queue.
103        queue: QueueKind,
104        /// How many entries were taken.
105        count: usize,
106    },
107    /// The session's plan was replaced wholesale (`update_plan` replaces,
108    /// it does not merge).
109    Plan {
110        /// The new checklist.
111        steps: Vec<PlanEntry>,
112    },
113    /// The session's resume handle changed.
114    Rename {
115        /// The handle before the rename.
116        from: String,
117        /// The handle after it.
118        to: String,
119    },
120    /// The durable view is caught up: `<name>.jsonl` now holds `messages`
121    /// messages, and every record above this line is already in it. What
122    /// follows is exactly what a crash would lose — see
123    /// [`JournalState::unpersisted`].
124    Checkpoint {
125        /// Messages in the transcript as just written (including the
126        /// system message at index 0).
127        messages: usize,
128    },
129    /// BP-13 (catalog Domain 9): the model this session sends to CHANGED
130    /// mid-session — a user `/model` switch or a fallback hop the loop
131    /// performed after a failure. The append-only journal is where every
132    /// persisted routing record lives; there is no second file.
133    ModelChange {
134        /// The typed record, exactly as `Agent::model_change_records`
135        /// carries it in memory.
136        record: crate::model_change::ModelChangeRecord,
137    },
138    /// BP-13: one completed model round-trip's usage accounting, carrying
139    /// the model REQUESTED and (when the provider reported one) the model
140    /// that actually SERVED it.
141    Usage {
142        /// The typed record, exactly as `Agent::usage_records` carries it.
143        record: crate::usage_log::UsageRecord,
144    },
145    /// An on-disk file of an older generation was upgraded in place.
146    Upgrade {
147        /// Store format version the file was at.
148        from_version: u32,
149        /// Store format version it is at now.
150        to_version: u32,
151        /// Store-relative file name holding the ORIGINAL bytes verbatim, so
152        /// the upgrade is reversible.
153        original: String,
154    },
155}
156
157/// One journal line: the discriminant, a timestamp, and the operation.
158#[derive(Debug, Clone, Serialize, Deserialize)]
159pub struct JournalRecord {
160    /// Always [`JOURNAL_RECORD_VERSION`].
161    pub supercode_journal: u8,
162    /// RFC3339 (UTC) time the record was appended.
163    pub ts: String,
164    /// The operation itself.
165    #[serde(flatten)]
166    pub op: JournalOp,
167}
168
169/// The append-only writer. Every [`Self::append`] is one line plus `\n`,
170/// flushed before returning — see the module docs.
171pub struct SessionJournal {
172    /// Opened on the FIRST append, never at construction: a session the user
173    /// opened and quit without saying anything must not leave a journal file
174    /// behind for a conversation that never happened.
175    file: Option<File>,
176    path: PathBuf,
177    fixed_timestamp: Option<String>,
178}
179
180impl SessionJournal {
181    /// Address the journal at `path` for appending. The file itself is
182    /// created by the first [`Self::append`], not here.
183    pub fn open_append(path: &Path) -> Result<Self> {
184        Ok(SessionJournal {
185            file: None,
186            path: path.to_path_buf(),
187            fixed_timestamp: None,
188        })
189    }
190
191    /// The same writer with a fixed RFC3339 stamp, so a byte-comparison
192    /// test has time as no variable at all — the
193    /// [`crate::sidecar::SidecarWriter::create_with_timestamp`] precedent.
194    pub fn with_fixed_timestamp(mut self, ts: impl Into<String>) -> Self {
195        self.fixed_timestamp = Some(ts.into());
196        self
197    }
198
199    /// The file this journal appends to.
200    pub fn path(&self) -> &Path {
201        &self.path
202    }
203
204    /// Append one operation: serialize, write the whole line, flush.
205    pub fn append(&mut self, op: JournalOp) -> Result<()> {
206        let record = JournalRecord {
207            supercode_journal: JOURNAL_RECORD_VERSION,
208            ts: self
209                .fixed_timestamp
210                .clone()
211                .unwrap_or_else(crate::sidecar::now_rfc3339),
212            op,
213        };
214        let mut line = serde_json::to_string(&record).map_err(Error::Decode)?;
215        line.push('\n');
216        let file = match &mut self.file {
217            Some(file) => file,
218            none => {
219                if let Some(parent) = self.path.parent() {
220                    std::fs::create_dir_all(parent)?;
221                }
222                none.insert(
223                    OpenOptions::new()
224                        .create(true)
225                        .append(true)
226                        .open(&self.path)?,
227                )
228            }
229        };
230        file.write_all(line.as_bytes())?;
231        file.flush()?;
232        Ok(())
233    }
234
235    /// Append one conversation message (the [`JournalOp::Message`] shortcut
236    /// every agent-loop call site uses). The record carries the same
237    /// full-fidelity [`NativeTurn`] shape the native-v2 sidecar writes,
238    /// metadata included — a journal line and a sidecar line describe the
239    /// same message identically.
240    pub fn append_message(&mut self, msg: &ChatMessage) -> Result<()> {
241        let mut turn = NativeTurn::from(msg);
242        if let Some(ts) = &self.fixed_timestamp {
243            turn.ts.clone_from(ts);
244            turn.metadata.insert("timestamp".to_string(), ts.clone());
245        }
246        self.append(JournalOp::Message {
247            message: Box::new(turn),
248        })
249    }
250}
251
252/// The state a journal describes once folded up.
253#[derive(Debug, Clone, Default)]
254pub struct JournalState {
255    /// The live conversation, after every recorded rewind/unrewind.
256    pub messages: Vec<ChatMessage>,
257    /// Messages recorded AFTER the last [`JournalOp::Checkpoint`] — the
258    /// turn (or part of a turn) that a crash would have lost, because the
259    /// end-of-turn transcript rewrite never ran. Empty in the ordinary
260    /// case, where the last thing that happened was a clean persist.
261    pub unpersisted: Vec<ChatMessage>,
262    /// Messages the last [`JournalOp::Checkpoint`] said the transcript
263    /// holds — `None` when the journal has no checkpoint yet.
264    pub checkpoint_messages: Option<usize>,
265    /// Tails removed by rewinds that have not been undone, newest last —
266    /// the proof that a rewind loses nothing.
267    pub undo_stack: Vec<Vec<ChatMessage>>,
268    /// Still-pending mid-turn steering inputs, oldest first.
269    pub steer_queue: Vec<String>,
270    /// Still-pending at-idle follow-up inputs, oldest first.
271    pub follow_up_queue: Vec<String>,
272    /// The current plan (empty when the session never recorded one).
273    pub plan: Vec<PlanEntry>,
274    /// Every rename this session's handle has been through, in order.
275    pub renames: Vec<(String, String)>,
276    /// Every in-place format upgrade recorded for this session.
277    pub upgrades: Vec<(u32, u32, String)>,
278    /// BP-13: every mid-session model change recorded, in order.
279    pub model_changes: Vec<crate::model_change::ModelChangeRecord>,
280    /// BP-13: every per-turn usage record written, in order.
281    pub usage: Vec<crate::usage_log::UsageRecord>,
282    /// Well-formed records read.
283    pub records: usize,
284    /// Lines skipped as unreadable — a torn trailing record after a crash,
285    /// or a foreign line someone concatenated in.
286    pub skipped: usize,
287}
288
289impl JournalState {
290    /// Pending inputs for one queue.
291    pub fn queue(&self, kind: QueueKind) -> &[String] {
292        match kind {
293            QueueKind::Steer => &self.steer_queue,
294            QueueKind::FollowUp => &self.follow_up_queue,
295        }
296    }
297
298    /// The plan as `(step, status)` pairs — the shape
299    /// `crate::tools::UpdatePlanTool` hands out.
300    pub fn plan_pairs(&self) -> Vec<(String, String)> {
301        self.plan
302            .iter()
303            .map(|s| (s.step.clone(), s.status.clone()))
304            .collect()
305    }
306}
307
308/// Fold a journal's text into the state it describes.
309///
310/// Unreadable lines are counted, never fatal: the last line of a journal
311/// whose process died mid-write is exactly the torn record this tolerates,
312/// and tolerating it is what makes the file safe to read after a crash.
313pub fn replay_str(text: &str) -> JournalState {
314    let mut state = JournalState::default();
315    for line in text.lines() {
316        if line.trim().is_empty() {
317            continue;
318        }
319        let Ok(record) = serde_json::from_str::<JournalRecord>(line) else {
320            state.skipped += 1;
321            continue;
322        };
323        if record.supercode_journal != JOURNAL_RECORD_VERSION {
324            state.skipped += 1;
325            continue;
326        }
327        state.records += 1;
328        match record.op {
329            JournalOp::Message { message } => {
330                let message = message.into_message();
331                state.unpersisted.push(message.clone());
332                state.messages.push(message);
333            }
334            JournalOp::Rewind { to } => {
335                let to = to.min(state.messages.len());
336                let tail = state.messages.split_off(to);
337                state.undo_stack.push(tail);
338                // A rewind reshapes the durable view too; whatever the
339                // checkpoint said is no longer the right base to recover
340                // against.
341                state.unpersisted.clear();
342                state.checkpoint_messages = None;
343            }
344            JournalOp::Unrewind => {
345                if let Some(mut tail) = state.undo_stack.pop() {
346                    state.messages.append(&mut tail);
347                }
348                state.unpersisted.clear();
349                state.checkpoint_messages = None;
350            }
351            JournalOp::Enqueue { queue, text } => match queue {
352                QueueKind::Steer => state.steer_queue.push(text),
353                QueueKind::FollowUp => state.follow_up_queue.push(text),
354            },
355            JournalOp::Dequeue { queue, count } => {
356                let q = match queue {
357                    QueueKind::Steer => &mut state.steer_queue,
358                    QueueKind::FollowUp => &mut state.follow_up_queue,
359                };
360                let count = count.min(q.len());
361                q.drain(..count);
362            }
363            JournalOp::Plan { steps } => state.plan = steps,
364            JournalOp::Checkpoint { messages } => {
365                state.unpersisted.clear();
366                state.checkpoint_messages = Some(messages);
367            }
368            JournalOp::Rename { from, to } => state.renames.push((from, to)),
369            JournalOp::Upgrade {
370                from_version,
371                to_version,
372                original,
373            } => state.upgrades.push((from_version, to_version, original)),
374            JournalOp::ModelChange { record } => state.model_changes.push(record),
375            JournalOp::Usage { record } => state.usage.push(record),
376        }
377    }
378    state
379}
380
381/// [`replay_str`] over a file. `Ok(None)` when no journal exists.
382pub fn replay(path: &Path) -> Result<Option<JournalState>> {
383    match std::fs::read_to_string(path) {
384        Ok(text) => Ok(Some(replay_str(&text))),
385        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
386        Err(e) => Err(e.into()),
387    }
388}
389
390/// A fresh RFC3339 stamp — re-exported so callers building records by hand
391/// do not reach into the sidecar module for it.
392pub fn timestamp() -> String {
393    now_rfc3339()
394}
395
396// ---------------------------------------------------------------------------
397// The composition every session door uses.
398// ---------------------------------------------------------------------------
399
400/// What [`arm`] found and put back.
401#[derive(Debug, Clone, Default, PartialEq, Eq)]
402pub struct RestoreReport {
403    /// Messages recovered from past the journal's last checkpoint — the turn
404    /// an interrupted process never got to persist.
405    pub recovered_messages: usize,
406    /// Pending steering + follow-up inputs re-queued.
407    pub restored_queue: usize,
408    /// Plan steps restored.
409    pub restored_plan: usize,
410    /// Rewinds whose undo is available again.
411    pub restored_rewinds: usize,
412    /// Whether a stored conversation tree was loaded (as opposed to
413    /// rebuilt from the linear history, or not armed at all).
414    pub tree_loaded: bool,
415    /// Whether the append-only journal is now open for writing.
416    pub journal_armed: bool,
417}
418
419/// BP-8 — arm what `[core.session]` promises for a session about to go live
420/// under `name`, and restore what a previous process left behind.
421///
422/// This is the composition, in the crate that owns both halves, so the CLI's
423/// session doors and an SDK embedder that owns a store run the SAME code —
424/// and so a test can drive the real path rather than a copy of it.
425///
426/// Order matters: everything is RESTORED first and the journal is opened for
427/// writing LAST, so a restored queue entry or plan is never re-recorded
428/// (which would double it on the next restart).
429pub fn arm(
430    agent: &mut crate::Agent,
431    store: &crate::store::SessionStore,
432    name: &str,
433) -> RestoreReport {
434    let mut report = RestoreReport::default();
435    if !agent.config().session_persist {
436        return report;
437    }
438    let state = store.load_journal(name).ok().flatten();
439    if let Some(state) = &state {
440        // catalog:150 — the turn a crash took with it. The journal is the
441        // only place those messages exist: the end-of-turn transcript
442        // rewrite never ran.
443        //
444        // Applied only when the loaded transcript is exactly the length the
445        // checkpoint claims. A mismatch means the two records disagree
446        // about what the session IS, and guessing there could duplicate
447        // turns; the honest move is to leave the transcript alone (the
448        // journal still holds every byte either way). A journal with NO
449        // checkpoint yet describes a session whose transcript was never
450        // written at all, so its base is the bare system message.
451        if !state.unpersisted.is_empty()
452            && state.checkpoint_messages.unwrap_or(1) == agent.history().len()
453        {
454            agent.append_recovered_messages(&state.unpersisted);
455            report.recovered_messages = state.unpersisted.len();
456        }
457        // catalog:154 — prompts typed while the agent was busy.
458        if agent.config().session_queue_persist {
459            agent.restore_queues(&state.steer_queue, &state.follow_up_queue);
460            report.restored_queue = state.steer_queue.len() + state.follow_up_queue.len();
461        }
462        // catalog:152 — the undo stack, so a rewind stays reversible across
463        // a restart.
464        if !state.undo_stack.is_empty() {
465            report.restored_rewinds = state.undo_stack.len();
466            agent.restore_rewind_undo(state.undo_stack.clone());
467        }
468    }
469    // catalog:156 — the plan. `<name>.plan.json` is the head `checkpoint`
470    // wrote; the journal is the fallback for a session whose last plan
471    // change never reached one.
472    if agent.config().todos_persist {
473        let plan = store
474            .load_plan(name)
475            .ok()
476            .flatten()
477            .filter(|p| !p.is_empty())
478            .or_else(|| state.as_ref().map(|s| s.plan.clone()))
479            .unwrap_or_default();
480        if !plan.is_empty() {
481            report.restored_plan = plan.len();
482            agent.set_plan(plan);
483        }
484    }
485    // catalog:151 — the conversation tree. A stored tree wins; otherwise
486    // materialize the degenerate single-path tree from the history just
487    // loaded, so a rewind has nodes to address.
488    if agent.config().session_tree_enabled {
489        match store.load_tree(name) {
490            Ok(Some(tree)) => {
491                report.tree_loaded = true;
492                agent.set_session_tree(tree);
493            }
494            _ => agent.rebuild_session_tree_from_history(),
495        }
496    }
497    // catalog:150 — arm the writer last.
498    if agent.config().session_append_only {
499        if let Ok(journal) = store.open_journal(name) {
500            agent.set_journal(journal);
501            // Only an EXISTING conversation needs its base declared; a
502            // fresh one's base is the bare system message, which is what
503            // the recovery comparison above assumes when no checkpoint
504            // record exists. Writing one here would create the journal file
505            // for a session that may never say anything.
506            if agent.history().len() > 1 {
507                agent.journal_checkpoint(agent.history().len());
508            }
509            report.journal_armed = true;
510        }
511    }
512    report
513}
514
515/// BP-8 — the mirror of [`arm`], run every time the durable view is
516/// rewritten: declare the journal caught up and write the plan and tree
517/// beside the transcript.
518///
519/// `messages` is the length of the view just written, which is what the
520/// recovery comparison in [`arm`] tests against.
521pub fn checkpoint(
522    agent: &crate::Agent,
523    store: &crate::store::SessionStore,
524    name: &str,
525    messages: usize,
526) {
527    agent.journal_checkpoint(messages);
528    if agent.config().todos_persist {
529        let _ = store.save_plan(name, &agent.plan());
530    }
531    // Written whenever the module is on — `has_branches()` alone would mean
532    // a session only acquires a tree at its FIRST rewind, and the rewind
533    // needs the tree that recorded the nodes it rewinds to.
534    if let Some(tree) = agent.session_tree() {
535        let _ = store.save_tree(name, tree);
536    }
537}
538
539#[cfg(test)]
540mod tests {
541    use super::*;
542    use crate::Role;
543
544    fn msg(role: Role, text: &str) -> ChatMessage {
545        match role {
546            Role::Assistant => ChatMessage::assistant(text),
547            _ => ChatMessage::user(text),
548        }
549    }
550
551    #[test]
552    fn every_append_is_a_flushed_line_readable_by_another_handle() {
553        let dir = std::env::temp_dir().join(format!("sc-journal-{}", std::process::id()));
554        std::fs::create_dir_all(&dir).unwrap();
555        let path = dir.join("a.journal.jsonl");
556        let _ = std::fs::remove_file(&path);
557        let mut j = SessionJournal::open_append(&path).unwrap();
558        j.append_message(&msg(Role::User, "one")).unwrap();
559        // Read with a SECOND, independent handle while the writer is still
560        // open: this is the "a crash right now keeps the record" property.
561        let seen = replay(&path).unwrap().unwrap();
562        assert_eq!(seen.messages.len(), 1);
563        j.append_message(&msg(Role::Assistant, "two")).unwrap();
564        let seen = replay(&path).unwrap().unwrap();
565        assert_eq!(seen.messages.len(), 2);
566        let _ = std::fs::remove_file(&path);
567    }
568
569    #[test]
570    fn a_rewind_is_recorded_and_invertible_without_losing_bytes() {
571        let text = [
572            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"a"}}"#,
573            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"assistant","content":"b"}}"#,
574            r#"{"supercode_journal":1,"ts":"t","op":"rewind","to":1}"#,
575        ]
576        .join("\n");
577        let state = replay_str(&text);
578        assert_eq!(state.messages.len(), 1);
579        assert_eq!(state.undo_stack.len(), 1);
580        assert_eq!(state.undo_stack[0][0].content.as_deref(), Some("b"));
581        // The removed message's bytes are still in the log — the file was
582        // only ever appended to.
583        assert!(text.contains(r#""content":"b""#));
584
585        let undone = format!(
586            "{text}\n{}",
587            r#"{"supercode_journal":1,"ts":"t","op":"unrewind"}"#
588        );
589        let state = replay_str(&undone);
590        assert_eq!(state.messages.len(), 2);
591        assert_eq!(state.messages[1].content.as_deref(), Some("b"));
592        assert!(state.undo_stack.is_empty());
593    }
594
595    #[test]
596    fn queue_records_fold_into_the_still_pending_inputs() {
597        let text = [
598            r#"{"supercode_journal":1,"ts":"t","op":"enqueue","queue":"steer","text":"s1"}"#,
599            r#"{"supercode_journal":1,"ts":"t","op":"enqueue","queue":"follow_up","text":"f1"}"#,
600            r#"{"supercode_journal":1,"ts":"t","op":"enqueue","queue":"follow_up","text":"f2"}"#,
601            r#"{"supercode_journal":1,"ts":"t","op":"dequeue","queue":"follow_up","count":1}"#,
602        ]
603        .join("\n");
604        let state = replay_str(&text);
605        assert_eq!(state.queue(QueueKind::Steer), ["s1"]);
606        assert_eq!(state.queue(QueueKind::FollowUp), ["f2"]);
607    }
608
609    #[test]
610    fn a_torn_trailing_line_is_skipped_not_fatal() {
611        let text = format!(
612            "{}\n{}",
613            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"a"}}"#,
614            r#"{"supercode_journal":1,"ts":"t","op":"messa"#
615        );
616        let state = replay_str(&text);
617        assert_eq!(state.messages.len(), 1);
618        assert_eq!(state.skipped, 1);
619    }
620
621    #[test]
622    fn messages_after_the_last_checkpoint_are_the_ones_a_crash_would_lose() {
623        let text = [
624            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"a"}}"#,
625            r#"{"supercode_journal":1,"ts":"t","op":"checkpoint","messages":2}"#,
626            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"b"}}"#,
627            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"assistant","content":"c"}}"#,
628        ]
629        .join("\n");
630        let state = replay_str(&text);
631        assert_eq!(state.messages.len(), 3);
632        assert_eq!(state.checkpoint_messages, Some(2));
633        let lost: Vec<_> = state
634            .unpersisted
635            .iter()
636            .map(|m| m.content.clone().unwrap_or_default())
637            .collect();
638        assert_eq!(lost, ["b", "c"]);
639    }
640
641    #[test]
642    fn plan_records_replace_rather_than_merge() {
643        let text = [
644            r#"{"supercode_journal":1,"ts":"t","op":"plan","steps":[{"step":"one","status":"pending"}]}"#,
645            r#"{"supercode_journal":1,"ts":"t","op":"plan","steps":[{"step":"one","status":"completed"},{"step":"two","status":"pending"}]}"#,
646        ]
647        .join("\n");
648        let state = replay_str(&text);
649        assert_eq!(
650            state.plan_pairs(),
651            vec![
652                ("one".to_string(), "completed".to_string()),
653                ("two".to_string(), "pending".to_string())
654            ]
655        );
656    }
657}