supercode-harness 0.4.20

The optional native Supercode agent and tool harness
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
//! BP-8 (catalog domain 5): the **append-only session journal** — one
//! flush-per-record log of everything that happens to a session while it is
//! live, written beside the transcript as `<name>.journal.jsonl`.
//!
//! The problem it solves. supercode's own store persists a session by
//! REWRITING `<name>.jsonl` from the agent's in-memory history at the end of
//! a turn (`SessionStore::save`). That is durable but not append-only: a
//! crash between the first tool call and the end of the turn loses the whole
//! turn. The genuine per-message writer with a flush on every line
//! ([`crate::sidecar::SidecarWriter`]) exists, but it owns the native-v2
//! SIDECAR file, whose presence means "this session is a reduced/imported
//! family" to every resume door — so it cannot simply be switched on for an
//! ordinary session without changing what that session IS.
//!
//! The journal is therefore a separate, additive family member with a
//! discriminated record shape of its own. It is:
//!
//! * **append-only** — nothing in it is ever rewritten or truncated, so the
//!   bytes of a message that a later rewind removed are still on disk;
//! * **line-atomic** — full line + `\n`, then `flush()`, exactly
//!   [`crate::sidecar::SidecarWriter::append`]'s guarantee, so a crash
//!   mid-write can only tear the record being written, never one already
//!   there, and [`replay_str`] skips a torn trailing line;
//! * **replayable** — [`replay`] folds the log into the state it describes:
//!   the live message list, the pending input queues, and the current plan;
//! * **invertible** — every state-changing operation has an inverse that is
//!   itself an appended record ([`JournalOp::Rewind`] ↔
//!   [`JournalOp::Unrewind`]), so "undo" never means "delete a record".
//!
//! Records carry a `supercode_journal` discriminant for the same reason
//! [`crate::sidecar::NativeTurn`] carries `supercode_turn`: neither Claude
//! Code's nor Codex's own record shapes have that key, so a tolerant foreign
//! loader that is ever pointed at this file skips these lines rather than
//! erroring on them.

use std::fs::{File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

use crate::error::{Error, Result};
use crate::sidecar::{now_rfc3339, NativeTurn};
use crate::ChatMessage;

/// Discriminant value every journal line carries.
pub const JOURNAL_RECORD_VERSION: u8 = 1;

/// Which of the agent's two input queues an [`JournalOp::Enqueue`] /
/// [`JournalOp::Dequeue`] record is about (`crate::agent::Agent`'s
/// `steer_queue` — mid-turn — and `follow_up_queue` — at idle).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum QueueKind {
    /// Mid-turn steering input (drained at the top of the next loop pass).
    Steer,
    /// At-idle follow-up input (drained when the loop would otherwise end).
    FollowUp,
}

/// One step of a persisted plan (`update_plan`'s checklist).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PlanEntry {
    /// The step text as the model wrote it.
    pub step: String,
    /// `pending` | `in_progress` | `completed`.
    pub status: String,
}

/// What a journal line records.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub enum JournalOp {
    /// One conversation message, appended the moment the agent produced it
    /// — the append-only half of "every event flushed to disk during the
    /// session".
    Message {
        /// The message, in the same full-fidelity wire shape the native-v2
        /// sidecar uses (metadata included — see [`NativeTurn`]).
        message: Box<NativeTurn>,
    },
    /// The conversation was rewound to `to` messages. Nothing is deleted:
    /// [`replay_str`] moves the removed tail onto an undo stack that
    /// [`JournalOp::Unrewind`] pops.
    Rewind {
        /// Message count kept (an index into the replayed message list).
        to: usize,
    },
    /// The inverse of the most recent [`JournalOp::Rewind`]: the removed
    /// tail comes back.
    Unrewind,
    /// A pending user input was queued.
    Enqueue {
        /// Which queue.
        queue: QueueKind,
        /// The queued text.
        text: String,
    },
    /// `count` pending inputs were consumed from `queue` (drained into the
    /// conversation), oldest first.
    Dequeue {
        /// Which queue.
        queue: QueueKind,
        /// How many entries were taken.
        count: usize,
    },
    /// The session's plan was replaced wholesale (`update_plan` replaces,
    /// it does not merge).
    Plan {
        /// The new checklist.
        steps: Vec<PlanEntry>,
    },
    /// The session's resume handle changed.
    Rename {
        /// The handle before the rename.
        from: String,
        /// The handle after it.
        to: String,
    },
    /// The durable view is caught up: `<name>.jsonl` now holds `messages`
    /// messages, and every record above this line is already in it. What
    /// follows is exactly what a crash would lose — see
    /// [`JournalState::unpersisted`].
    Checkpoint {
        /// Messages in the transcript as just written (including the
        /// system message at index 0).
        messages: usize,
    },
    /// BP-13 (catalog Domain 9): the model this session sends to CHANGED
    /// mid-session — a user `/model` switch or a fallback hop the loop
    /// performed after a failure. The append-only journal is where every
    /// persisted routing record lives; there is no second file.
    ModelChange {
        /// The typed record, exactly as `Agent::model_change_records`
        /// carries it in memory.
        record: crate::model_change::ModelChangeRecord,
    },
    /// BP-13: one completed model round-trip's usage accounting, carrying
    /// the model REQUESTED and (when the provider reported one) the model
    /// that actually SERVED it.
    Usage {
        /// The typed record, exactly as `Agent::usage_records` carries it.
        record: crate::usage_log::UsageRecord,
    },
    /// An on-disk file of an older generation was upgraded in place.
    Upgrade {
        /// Store format version the file was at.
        from_version: u32,
        /// Store format version it is at now.
        to_version: u32,
        /// Store-relative file name holding the ORIGINAL bytes verbatim, so
        /// the upgrade is reversible.
        original: String,
    },
}

/// One journal line: the discriminant, a timestamp, and the operation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JournalRecord {
    /// Always [`JOURNAL_RECORD_VERSION`].
    pub supercode_journal: u8,
    /// RFC3339 (UTC) time the record was appended.
    pub ts: String,
    /// The operation itself.
    #[serde(flatten)]
    pub op: JournalOp,
}

/// The append-only writer. Every [`Self::append`] is one line plus `\n`,
/// flushed before returning — see the module docs.
pub struct SessionJournal {
    /// Opened on the FIRST append, never at construction: a session the user
    /// opened and quit without saying anything must not leave a journal file
    /// behind for a conversation that never happened.
    file: Option<File>,
    path: PathBuf,
    fixed_timestamp: Option<String>,
}

impl SessionJournal {
    /// Address the journal at `path` for appending. The file itself is
    /// created by the first [`Self::append`], not here.
    pub fn open_append(path: &Path) -> Result<Self> {
        Ok(SessionJournal {
            file: None,
            path: path.to_path_buf(),
            fixed_timestamp: None,
        })
    }

    /// The same writer with a fixed RFC3339 stamp, so a byte-comparison
    /// test has time as no variable at all — the
    /// [`crate::sidecar::SidecarWriter::create_with_timestamp`] precedent.
    pub fn with_fixed_timestamp(mut self, ts: impl Into<String>) -> Self {
        self.fixed_timestamp = Some(ts.into());
        self
    }

    /// The file this journal appends to.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Append one operation: serialize, write the whole line, flush.
    pub fn append(&mut self, op: JournalOp) -> Result<()> {
        let record = JournalRecord {
            supercode_journal: JOURNAL_RECORD_VERSION,
            ts: self
                .fixed_timestamp
                .clone()
                .unwrap_or_else(crate::sidecar::now_rfc3339),
            op,
        };
        let mut line = serde_json::to_string(&record).map_err(Error::Decode)?;
        line.push('\n');
        let file = match &mut self.file {
            Some(file) => file,
            none => {
                if let Some(parent) = self.path.parent() {
                    std::fs::create_dir_all(parent)?;
                }
                none.insert(
                    OpenOptions::new()
                        .create(true)
                        .append(true)
                        .open(&self.path)?,
                )
            }
        };
        file.write_all(line.as_bytes())?;
        file.flush()?;
        Ok(())
    }

    /// Append one conversation message (the [`JournalOp::Message`] shortcut
    /// every agent-loop call site uses). The record carries the same
    /// full-fidelity [`NativeTurn`] shape the native-v2 sidecar writes,
    /// metadata included — a journal line and a sidecar line describe the
    /// same message identically.
    pub fn append_message(&mut self, msg: &ChatMessage) -> Result<()> {
        let mut turn = NativeTurn::from(msg);
        if let Some(ts) = &self.fixed_timestamp {
            turn.ts.clone_from(ts);
            turn.metadata.insert("timestamp".to_string(), ts.clone());
        }
        self.append(JournalOp::Message {
            message: Box::new(turn),
        })
    }
}

/// The state a journal describes once folded up.
#[derive(Debug, Clone, Default)]
pub struct JournalState {
    /// The live conversation, after every recorded rewind/unrewind.
    pub messages: Vec<ChatMessage>,
    /// Messages recorded AFTER the last [`JournalOp::Checkpoint`] — the
    /// turn (or part of a turn) that a crash would have lost, because the
    /// end-of-turn transcript rewrite never ran. Empty in the ordinary
    /// case, where the last thing that happened was a clean persist.
    pub unpersisted: Vec<ChatMessage>,
    /// Messages the last [`JournalOp::Checkpoint`] said the transcript
    /// holds — `None` when the journal has no checkpoint yet.
    pub checkpoint_messages: Option<usize>,
    /// Tails removed by rewinds that have not been undone, newest last —
    /// the proof that a rewind loses nothing.
    pub undo_stack: Vec<Vec<ChatMessage>>,
    /// Still-pending mid-turn steering inputs, oldest first.
    pub steer_queue: Vec<String>,
    /// Still-pending at-idle follow-up inputs, oldest first.
    pub follow_up_queue: Vec<String>,
    /// The current plan (empty when the session never recorded one).
    pub plan: Vec<PlanEntry>,
    /// Every rename this session's handle has been through, in order.
    pub renames: Vec<(String, String)>,
    /// Every in-place format upgrade recorded for this session.
    pub upgrades: Vec<(u32, u32, String)>,
    /// BP-13: every mid-session model change recorded, in order.
    pub model_changes: Vec<crate::model_change::ModelChangeRecord>,
    /// BP-13: every per-turn usage record written, in order.
    pub usage: Vec<crate::usage_log::UsageRecord>,
    /// Well-formed records read.
    pub records: usize,
    /// Lines skipped as unreadable — a torn trailing record after a crash,
    /// or a foreign line someone concatenated in.
    pub skipped: usize,
}

impl JournalState {
    /// Pending inputs for one queue.
    pub fn queue(&self, kind: QueueKind) -> &[String] {
        match kind {
            QueueKind::Steer => &self.steer_queue,
            QueueKind::FollowUp => &self.follow_up_queue,
        }
    }

    /// The plan as `(step, status)` pairs — the shape
    /// `crate::tools::UpdatePlanTool` hands out.
    pub fn plan_pairs(&self) -> Vec<(String, String)> {
        self.plan
            .iter()
            .map(|s| (s.step.clone(), s.status.clone()))
            .collect()
    }
}

/// Fold a journal's text into the state it describes.
///
/// Unreadable lines are counted, never fatal: the last line of a journal
/// whose process died mid-write is exactly the torn record this tolerates,
/// and tolerating it is what makes the file safe to read after a crash.
pub fn replay_str(text: &str) -> JournalState {
    let mut state = JournalState::default();
    for line in text.lines() {
        if line.trim().is_empty() {
            continue;
        }
        let Ok(record) = serde_json::from_str::<JournalRecord>(line) else {
            state.skipped += 1;
            continue;
        };
        if record.supercode_journal != JOURNAL_RECORD_VERSION {
            state.skipped += 1;
            continue;
        }
        state.records += 1;
        match record.op {
            JournalOp::Message { message } => {
                let message = message.into_message();
                state.unpersisted.push(message.clone());
                state.messages.push(message);
            }
            JournalOp::Rewind { to } => {
                let to = to.min(state.messages.len());
                let tail = state.messages.split_off(to);
                state.undo_stack.push(tail);
                // A rewind reshapes the durable view too; whatever the
                // checkpoint said is no longer the right base to recover
                // against.
                state.unpersisted.clear();
                state.checkpoint_messages = None;
            }
            JournalOp::Unrewind => {
                if let Some(mut tail) = state.undo_stack.pop() {
                    state.messages.append(&mut tail);
                }
                state.unpersisted.clear();
                state.checkpoint_messages = None;
            }
            JournalOp::Enqueue { queue, text } => match queue {
                QueueKind::Steer => state.steer_queue.push(text),
                QueueKind::FollowUp => state.follow_up_queue.push(text),
            },
            JournalOp::Dequeue { queue, count } => {
                let q = match queue {
                    QueueKind::Steer => &mut state.steer_queue,
                    QueueKind::FollowUp => &mut state.follow_up_queue,
                };
                let count = count.min(q.len());
                q.drain(..count);
            }
            JournalOp::Plan { steps } => state.plan = steps,
            JournalOp::Checkpoint { messages } => {
                state.unpersisted.clear();
                state.checkpoint_messages = Some(messages);
            }
            JournalOp::Rename { from, to } => state.renames.push((from, to)),
            JournalOp::Upgrade {
                from_version,
                to_version,
                original,
            } => state.upgrades.push((from_version, to_version, original)),
            JournalOp::ModelChange { record } => state.model_changes.push(record),
            JournalOp::Usage { record } => state.usage.push(record),
        }
    }
    state
}

/// [`replay_str`] over a file. `Ok(None)` when no journal exists.
pub fn replay(path: &Path) -> Result<Option<JournalState>> {
    match std::fs::read_to_string(path) {
        Ok(text) => Ok(Some(replay_str(&text))),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(e) => Err(e.into()),
    }
}

/// A fresh RFC3339 stamp — re-exported so callers building records by hand
/// do not reach into the sidecar module for it.
pub fn timestamp() -> String {
    now_rfc3339()
}

// ---------------------------------------------------------------------------
// The composition every session door uses.
// ---------------------------------------------------------------------------

/// What [`arm`] found and put back.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RestoreReport {
    /// Messages recovered from past the journal's last checkpoint — the turn
    /// an interrupted process never got to persist.
    pub recovered_messages: usize,
    /// Pending steering + follow-up inputs re-queued.
    pub restored_queue: usize,
    /// Plan steps restored.
    pub restored_plan: usize,
    /// Rewinds whose undo is available again.
    pub restored_rewinds: usize,
    /// Whether a stored conversation tree was loaded (as opposed to
    /// rebuilt from the linear history, or not armed at all).
    pub tree_loaded: bool,
    /// Whether the append-only journal is now open for writing.
    pub journal_armed: bool,
}

/// BP-8 — arm what `[core.session]` promises for a session about to go live
/// under `name`, and restore what a previous process left behind.
///
/// This is the composition, in the crate that owns both halves, so the CLI's
/// session doors and an SDK embedder that owns a store run the SAME code —
/// and so a test can drive the real path rather than a copy of it.
///
/// Order matters: everything is RESTORED first and the journal is opened for
/// writing LAST, so a restored queue entry or plan is never re-recorded
/// (which would double it on the next restart).
pub fn arm(
    agent: &mut crate::Agent,
    store: &crate::store::SessionStore,
    name: &str,
) -> RestoreReport {
    let mut report = RestoreReport::default();
    if !agent.config().session_persist {
        return report;
    }
    let state = store.load_journal(name).ok().flatten();
    if let Some(state) = &state {
        // catalog:150 — the turn a crash took with it. The journal is the
        // only place those messages exist: the end-of-turn transcript
        // rewrite never ran.
        //
        // Applied only when the loaded transcript is exactly the length the
        // checkpoint claims. A mismatch means the two records disagree
        // about what the session IS, and guessing there could duplicate
        // turns; the honest move is to leave the transcript alone (the
        // journal still holds every byte either way). A journal with NO
        // checkpoint yet describes a session whose transcript was never
        // written at all, so its base is the bare system message.
        if !state.unpersisted.is_empty()
            && state.checkpoint_messages.unwrap_or(1) == agent.history().len()
        {
            agent.append_recovered_messages(&state.unpersisted);
            report.recovered_messages = state.unpersisted.len();
        }
        // catalog:154 — prompts typed while the agent was busy.
        if agent.config().session_queue_persist {
            agent.restore_queues(&state.steer_queue, &state.follow_up_queue);
            report.restored_queue = state.steer_queue.len() + state.follow_up_queue.len();
        }
        // catalog:152 — the undo stack, so a rewind stays reversible across
        // a restart.
        if !state.undo_stack.is_empty() {
            report.restored_rewinds = state.undo_stack.len();
            agent.restore_rewind_undo(state.undo_stack.clone());
        }
    }
    // catalog:156 — the plan. `<name>.plan.json` is the head `checkpoint`
    // wrote; the journal is the fallback for a session whose last plan
    // change never reached one.
    if agent.config().todos_persist {
        let plan = store
            .load_plan(name)
            .ok()
            .flatten()
            .filter(|p| !p.is_empty())
            .or_else(|| state.as_ref().map(|s| s.plan.clone()))
            .unwrap_or_default();
        if !plan.is_empty() {
            report.restored_plan = plan.len();
            agent.set_plan(plan);
        }
    }
    // catalog:151 — the conversation tree. A stored tree wins; otherwise
    // materialize the degenerate single-path tree from the history just
    // loaded, so a rewind has nodes to address.
    if agent.config().session_tree_enabled {
        match store.load_tree(name) {
            Ok(Some(tree)) => {
                report.tree_loaded = true;
                agent.set_session_tree(tree);
            }
            _ => agent.rebuild_session_tree_from_history(),
        }
    }
    // catalog:150 — arm the writer last.
    if agent.config().session_append_only {
        if let Ok(journal) = store.open_journal(name) {
            agent.set_journal(journal);
            // Only an EXISTING conversation needs its base declared; a
            // fresh one's base is the bare system message, which is what
            // the recovery comparison above assumes when no checkpoint
            // record exists. Writing one here would create the journal file
            // for a session that may never say anything.
            if agent.history().len() > 1 {
                agent.journal_checkpoint(agent.history().len());
            }
            report.journal_armed = true;
        }
    }
    report
}

/// BP-8 — the mirror of [`arm`], run every time the durable view is
/// rewritten: declare the journal caught up and write the plan and tree
/// beside the transcript.
///
/// `messages` is the length of the view just written, which is what the
/// recovery comparison in [`arm`] tests against.
pub fn checkpoint(
    agent: &crate::Agent,
    store: &crate::store::SessionStore,
    name: &str,
    messages: usize,
) {
    agent.journal_checkpoint(messages);
    if agent.config().todos_persist {
        let _ = store.save_plan(name, &agent.plan());
    }
    // Written whenever the module is on — `has_branches()` alone would mean
    // a session only acquires a tree at its FIRST rewind, and the rewind
    // needs the tree that recorded the nodes it rewinds to.
    if let Some(tree) = agent.session_tree() {
        let _ = store.save_tree(name, tree);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Role;

    fn msg(role: Role, text: &str) -> ChatMessage {
        match role {
            Role::Assistant => ChatMessage::assistant(text),
            _ => ChatMessage::user(text),
        }
    }

    #[test]
    fn every_append_is_a_flushed_line_readable_by_another_handle() {
        let dir = std::env::temp_dir().join(format!("sc-journal-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("a.journal.jsonl");
        let _ = std::fs::remove_file(&path);
        let mut j = SessionJournal::open_append(&path).unwrap();
        j.append_message(&msg(Role::User, "one")).unwrap();
        // Read with a SECOND, independent handle while the writer is still
        // open: this is the "a crash right now keeps the record" property.
        let seen = replay(&path).unwrap().unwrap();
        assert_eq!(seen.messages.len(), 1);
        j.append_message(&msg(Role::Assistant, "two")).unwrap();
        let seen = replay(&path).unwrap().unwrap();
        assert_eq!(seen.messages.len(), 2);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn a_rewind_is_recorded_and_invertible_without_losing_bytes() {
        let text = [
            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"a"}}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"assistant","content":"b"}}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"rewind","to":1}"#,
        ]
        .join("\n");
        let state = replay_str(&text);
        assert_eq!(state.messages.len(), 1);
        assert_eq!(state.undo_stack.len(), 1);
        assert_eq!(state.undo_stack[0][0].content.as_deref(), Some("b"));
        // The removed message's bytes are still in the log — the file was
        // only ever appended to.
        assert!(text.contains(r#""content":"b""#));

        let undone = format!(
            "{text}\n{}",
            r#"{"supercode_journal":1,"ts":"t","op":"unrewind"}"#
        );
        let state = replay_str(&undone);
        assert_eq!(state.messages.len(), 2);
        assert_eq!(state.messages[1].content.as_deref(), Some("b"));
        assert!(state.undo_stack.is_empty());
    }

    #[test]
    fn queue_records_fold_into_the_still_pending_inputs() {
        let text = [
            r#"{"supercode_journal":1,"ts":"t","op":"enqueue","queue":"steer","text":"s1"}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"enqueue","queue":"follow_up","text":"f1"}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"enqueue","queue":"follow_up","text":"f2"}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"dequeue","queue":"follow_up","count":1}"#,
        ]
        .join("\n");
        let state = replay_str(&text);
        assert_eq!(state.queue(QueueKind::Steer), ["s1"]);
        assert_eq!(state.queue(QueueKind::FollowUp), ["f2"]);
    }

    #[test]
    fn a_torn_trailing_line_is_skipped_not_fatal() {
        let text = format!(
            "{}\n{}",
            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"a"}}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"messa"#
        );
        let state = replay_str(&text);
        assert_eq!(state.messages.len(), 1);
        assert_eq!(state.skipped, 1);
    }

    #[test]
    fn messages_after_the_last_checkpoint_are_the_ones_a_crash_would_lose() {
        let text = [
            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"a"}}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"checkpoint","messages":2}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"user","content":"b"}}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"message","message":{"supercode_turn":1,"ts":"t","role":"assistant","content":"c"}}"#,
        ]
        .join("\n");
        let state = replay_str(&text);
        assert_eq!(state.messages.len(), 3);
        assert_eq!(state.checkpoint_messages, Some(2));
        let lost: Vec<_> = state
            .unpersisted
            .iter()
            .map(|m| m.content.clone().unwrap_or_default())
            .collect();
        assert_eq!(lost, ["b", "c"]);
    }

    #[test]
    fn plan_records_replace_rather_than_merge() {
        let text = [
            r#"{"supercode_journal":1,"ts":"t","op":"plan","steps":[{"step":"one","status":"pending"}]}"#,
            r#"{"supercode_journal":1,"ts":"t","op":"plan","steps":[{"step":"one","status":"completed"},{"step":"two","status":"pending"}]}"#,
        ]
        .join("\n");
        let state = replay_str(&text);
        assert_eq!(
            state.plan_pairs(),
            vec![
                ("one".to_string(), "completed".to_string()),
                ("two".to_string(), "pending".to_string())
            ]
        );
    }
}