diff --git a/Cargo.lock b/Cargo.lock index db2238e..ef1e757 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -663,9 +663,9 @@ dependencies = [ [[package]] name = "fmtview" -version = "0.6.1" +version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7256652607c8f5eff696a873e73690a3c4ad26fab62d00e85a3e4d6a4d4cdda" +checksum = "9d2798863a1347f246f212fbf7e294de1728d84d692c8496d9f169632b04f54b" dependencies = [ "anyhow", "clap", @@ -677,9 +677,9 @@ dependencies = [ [[package]] name = "fmtview-core" -version = "0.6.1" +version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "80239d61ba3a62c8764d9d34bc939a637d4439a794a71abf4d2e427e684c9def" +checksum = "1c281ad6eb69f32992a66bd07ee2c3c203faaeb17a43aad9bfb7166985404e02" dependencies = [ "anyhow", "memchr", diff --git a/Cargo.toml b/Cargo.toml index b922131..9ad91c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,7 @@ chrono = { version = "0.4", features = ["serde"] } clap = { version = "4", features = ["derive", "env"] } eventsource-stream = "0.2" futures-util = "0.3" -fmtview = "0.6.1" +fmtview = "0.6.2" image = { version = "0.25", default-features = false, features = ["bmp", "gif", "png"] } libc = "0.2" regex = "1" diff --git a/README.md b/README.md index 3cb4ba2..cdf498f 100644 --- a/README.md +++ b/README.md @@ -7,9 +7,10 @@ delegated agents and long-running background jobs through one durable runtime for local automation and cloud workloads. The runtime deliberately shares one `AgentRunner` implementation and one tool -contract across root and child agents. Runs are portable, resumable, and -inspectable without an embedded TUI, database, distributed scheduler, sandbox, -or approval system. +contract across root and child agents. Runs are portable and resumable without +a database, distributed scheduler, sandbox, or approval system. Transcript +inspection embeds fmtview behind a storage-neutral boundary; the agent runtime +itself remains headless. ## Features diff --git a/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md b/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md index fcfda86..b3bb15a 100644 --- a/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md +++ b/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md @@ -32,13 +32,14 @@ the storage boundary in the other direction. by reverse physical-line scanning, then feeds that candidate forward through the same `CheckpointDecoder` used by trajectory loading and append recovery. Older reads repeat this by whole groups; newer reads continue forward from a - known sequence and byte boundary. A checkpoint can cross a load budget but is - never partially published. + known sequence and byte boundary. A checkpoint can cross a load budget only + as the first group in a batch and is never partially published. - Follow refresh retains the pending decoder, torn-line bytes, and scan cursor. Bounded samples detect committed-prefix and pending-suffix changes. Rewriting only an uncommitted tail rebuilds pending state from the same committed - boundary without duplicating records. Committed truncation or replacement - starts a new record-id epoch. + boundary without duplicating records. Refresh retries a concurrently changing + file without publishing partial decoder progress. Committed truncation or + replacement starts a new record-id epoch. - Queued, running, and idle run states are live. Completed, failed, cancelled, and closed states are terminal. - On a TTY, `fiasco inspect ` opens a tail-first snapshot and diff --git a/docs/architecture.md b/docs/architecture.md index 3ad5a25..81f5eec 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -200,16 +200,19 @@ Fiasco depends on the released `fmtview` facade and does not depend directly on The timeline opens at the last committed checkpoint by scanning physical lines backward from EOF. It loads older checkpoints backward and newer checkpoints -forward; one logical checkpoint may exceed a requested record/byte budget but -is never split across published batches. Prefix probes are bounded. The initial -tail path does not index or validate every earlier message, so a large history -can show its first screen without a forward scan. +forward; one logical checkpoint may exceed a requested record/byte budget when +it is the first group in a batch, but is never split across published batches. +Prefix probes are bounded. The initial tail path does not index or validate +every earlier message, so a large history can show its first screen without a +forward scan. Follow refresh retains the shared `CheckpointDecoder`, its pending checkpoint, a torn-line buffer, and the scanned suffix cursor. An unchanged large pending checkpoint therefore costs only bounded head/middle/tail probes per refresh. Rewriting only the uncommitted suffix invalidates and rebuilds this tracker from -the unchanged committed boundary without changing the epoch. Truncating a +the unchanged committed boundary without changing the epoch. Reads from a +concurrently shrinking or rewritten suffix are retried from a clean working +tracker and published only after one coherent file observation. Truncating a committed prefix, replacing the file identity, or changing bounded committed prefix probes starts a new epoch so fmtview discards old record identities. Queued, running, and idle runs report a live boundary; completed, failed, diff --git a/src/storage/message_log/decoder.rs b/src/storage/message_log/decoder.rs index 25e56f7..71025a8 100644 --- a/src/storage/message_log/decoder.rs +++ b/src/storage/message_log/decoder.rs @@ -73,6 +73,29 @@ impl CheckpointDecoder { self.next_seq } + /// Validate one complete line as the first record of the next checkpoint + /// without advancing decoder state. Directional readers use this to stop + /// before a checkpoint that cannot fit the remainder of a non-empty batch. + pub(crate) fn preflight_checkpoint_start( + &self, + path: &std::path::Path, + line_with_newline: &[u8], + ) -> Result { + ensure!( + self.pending.is_none(), + "checkpoint preflight requires a checkpoint boundary" + ); + ensure!( + line_with_newline.ends_with(b"\n"), + "checkpoint preflight requires a newline-terminated record" + ); + let stored = parse_stored_line(path, line_with_newline)?; + let checkpoint = stored.local.checkpoint.clone(); + self.validate_checkpoint_start(&stored, checkpoint.as_ref())?; + let _ = trajectory_record(stored, self.next_seq)?; + Ok(checkpoint.map_or(1, |checkpoint| checkpoint.count)) + } + pub(crate) fn push_complete_line( &mut self, path: &std::path::Path, @@ -113,28 +136,8 @@ impl CheckpointDecoder { "message `{}` has inconsistent checkpoint metadata", stored.message_ref ); - } else if let Some(checkpoint) = checkpoint.as_ref() { - ensure!( - checkpoint.count > 0, - "message checkpoint count must be positive" - ); - ensure!( - checkpoint.index == 0, - "message checkpoint `{}` starts at index {} instead of 0", - checkpoint.first_message_ref, - checkpoint.index - ); - ensure!( - checkpoint.first_message_ref == stored.message_ref, - "message checkpoint `{}` starts with message `{}`", - checkpoint.first_message_ref, - stored.message_ref - ); - usize::try_from(checkpoint.count) - .context("message checkpoint count does not fit in memory")?; - self.next_seq - .checked_add(checkpoint.count) - .context("message sequence overflow after checkpoint")?; + } else { + self.validate_checkpoint_start(&stored, checkpoint.as_ref())?; } let expected_seq = self .next_seq @@ -183,6 +186,37 @@ impl CheckpointDecoder { } } + fn validate_checkpoint_start( + &self, + stored: &super::StoredMessage, + checkpoint: Option<&MessageCheckpoint>, + ) -> Result<()> { + if let Some(checkpoint) = checkpoint { + ensure!( + checkpoint.count > 0, + "message checkpoint count must be positive" + ); + ensure!( + checkpoint.index == 0, + "message checkpoint `{}` starts at index {} instead of 0", + checkpoint.first_message_ref, + checkpoint.index + ); + ensure!( + checkpoint.first_message_ref == stored.message_ref, + "message checkpoint `{}` starts with message `{}`", + checkpoint.first_message_ref, + stored.message_ref + ); + usize::try_from(checkpoint.count) + .context("message checkpoint count does not fit in memory")?; + self.next_seq + .checked_add(checkpoint.count) + .context("message sequence overflow after checkpoint")?; + } + Ok(()) + } + fn commit_pending(&mut self, committed_end: u64) -> Result { let pending = self .pending diff --git a/src/storage/message_log/transcript/refresh.rs b/src/storage/message_log/transcript/refresh.rs index a6c4e99..77d1a97 100644 --- a/src/storage/message_log/transcript/refresh.rs +++ b/src/storage/message_log/transcript/refresh.rs @@ -1,7 +1,8 @@ use std::{ fs::{File, Metadata}, - io::{BufRead, BufReader, Read, Seek, SeekFrom}, + io::{self, BufRead, BufReader, Read, Seek, SeekFrom}, path::Path, + time::SystemTime, }; use anyhow::{Context, Result, ensure}; @@ -11,15 +12,53 @@ use super::{TranscriptInstrumentation, TranscriptTimeline, read_run, scan::read_ use crate::storage::message_log::decoder::CheckpointDecoder; const SAMPLE_BYTES: usize = 64; +const REFRESH_SHORT_READ_ATTEMPTS: usize = 3; impl TranscriptTimeline { pub(super) fn refresh_timeline(&mut self) -> Result { + self.refresh_timeline_with_hook(|_| Ok(())) + } + + pub(super) fn refresh_timeline_with_hook( + &mut self, + mut after_stat: impl FnMut(usize) -> Result<()>, + ) -> Result { + for attempt in 1..=REFRESH_SHORT_READ_ATTEMPTS { + let mut observation = None; + match self.refresh_once(&mut observation, || after_stat(attempt)) { + Ok(refresh) => return Ok(refresh), + Err(error) if is_unexpected_eof(&error) => { + let changed = observation.is_some_and(|previous| { + current_observation(&self.messages_path) + .is_ok_and(|current| current != previous) + }); + if changed { + if attempt < REFRESH_SHORT_READ_ATTEMPTS { + continue; + } + return Ok(TimelineRefresh::NoChange(self.snapshot())); + } + return Err(error); + } + Err(error) => return Err(error), + } + } + unreachable!("refresh retry loop always returns") + } + + fn refresh_once( + &mut self, + observation: &mut Option, + after_stat: impl FnOnce() -> Result<()>, + ) -> Result { let run = read_run(&self.metadata_path)?; let mut file = File::open(&self.messages_path) .with_context(|| format!("reopen transcript {}", self.messages_path.display()))?; let metadata = file .metadata() .with_context(|| format!("stat transcript {}", self.messages_path.display()))?; + *observation = Some(FileObservation::from_metadata(&metadata)); + after_stat()?; let identity = FileIdentity::from_metadata(&metadata); if identity != self.identity { return self.reset(TimelineResetReason::IdentityChanged); @@ -37,33 +76,66 @@ impl TranscriptTimeline { } let old_committed_end = self.committed_end; - if !self.suffix_tracker.matches( + let suffix_matches = self.suffix_tracker.matches( &mut file, metadata.len(), &mut self.instrumentation, &self.label, - )? { - self.suffix_tracker = SuffixTracker::new(self.committed_end, self.committed_next_seq); - } - self.suffix_tracker.scan_to( + )?; + let mut suffix_tracker = if suffix_matches { + std::mem::replace( + &mut self.suffix_tracker, + SuffixTracker::new(self.committed_end, self.committed_next_seq), + ) + } else { + SuffixTracker::new(self.committed_end, self.committed_next_seq) + }; + if let Err(error) = suffix_tracker.scan_to( &mut file, &self.messages_path, metadata.len(), &mut self.instrumentation, &self.label, - )?; + ) { + // The working tracker may have consumed bytes from a concurrent + // truncate. Keep only a clean committed-boundary tracker so a + // retry cannot inherit its partial cursor or decoder state. + self.suffix_tracker = SuffixTracker::new(self.committed_end, self.committed_next_seq); + return Err(error); + } + let committed_end = suffix_tracker.committed_end(); + let committed_next_seq = suffix_tracker.next_seq(); + let prefix_sample = if committed_end != old_committed_end { + Some(PrefixSample::read( + &mut file, + committed_end, + &mut self.instrumentation, + &self.label, + )?) + } else { + None + }; + let after_reads = current_observation(&self.messages_path)?; + if observation.is_some_and(|before_reads| after_reads != before_reads) { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!("transcript {} changed during refresh", self.label), + ) + .into()); + } + + // Publish the new snapshot only after every read against the statted + // length succeeds. A concurrent shrink leaves the prior decoder, + // partial line, cursor, and samples untouched for a clean retry. self.file = file; self.state = run.state; self.observed_end = metadata.len(); - self.committed_end = self.suffix_tracker.committed_end(); - self.committed_next_seq = self.suffix_tracker.next_seq(); + self.committed_end = committed_end; + self.committed_next_seq = committed_next_seq; + self.suffix_tracker = suffix_tracker; if self.committed_end != old_committed_end { - self.prefix_sample = PrefixSample::read( - &mut self.file, - self.committed_end, - &mut self.instrumentation, - &self.label, - )?; + self.prefix_sample = + prefix_sample.context("appended transcript lost its prefix sample")?; return Ok(TimelineRefresh::Appended(self.snapshot())); } if self.is_terminal() { @@ -94,6 +166,37 @@ impl TranscriptTimeline { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct FileObservation { + identity: FileIdentity, + len: u64, + modified: Option, +} + +impl FileObservation { + fn from_metadata(metadata: &Metadata) -> Self { + Self { + identity: FileIdentity::from_metadata(metadata), + len: metadata.len(), + modified: metadata.modified().ok(), + } + } +} + +fn current_observation(path: &Path) -> Result { + let metadata = std::fs::metadata(path) + .with_context(|| format!("restat transcript {} after short read", path.display()))?; + Ok(FileObservation::from_metadata(&metadata)) +} + +fn is_unexpected_eof(error: &anyhow::Error) -> bool { + error.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(|error| error.kind() == io::ErrorKind::UnexpectedEof) + }) +} + pub(super) struct SuffixTracker { decoder: CheckpointDecoder, scan_cursor: u64, @@ -162,7 +265,14 @@ impl SuffixTracker { instrumentation.read_operations = instrumentation.read_operations.saturating_add(1); instrumentation.bytes_read = instrumentation.bytes_read.saturating_add(read as u64); if read == 0 { - break; + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "transcript {label} ended at byte {} before statted byte {observed_end}", + self.scan_cursor + ), + ) + .into()); } self.scan_cursor = self .scan_cursor diff --git a/src/storage/message_log/transcript/scan.rs b/src/storage/message_log/transcript/scan.rs index 3dc7ace..90da5bd 100644 --- a/src/storage/message_log/transcript/scan.rs +++ b/src/storage/message_log/transcript/scan.rs @@ -233,10 +233,38 @@ pub(super) fn read_forward_batch( read > 0, "transcript at byte {cursor} made no read progress" ); - cursor = cursor + let line_end = cursor .checked_add(read as u64) .context("transcript byte offset overflow")?; ensure!(line.ends_with(b"\n"), "committed transcript line is torn"); + if !records.is_empty() && decoder.committed_end() == cursor { + let declared_records = decoder.preflight_checkpoint_start(path, &line)?; + let remaining_records = limit.max_records.saturating_sub(records.len()); + let remaining_bytes = limit.max_bytes.saturating_sub(bytes); + if declared_records > u64::try_from(remaining_records).unwrap_or(u64::MAX) + || line.len() > remaining_bytes + { + break; + } + if !remaining_checkpoint_lines_fit( + file, + line_end, + end, + declared_records.saturating_sub(1), + remaining_bytes.saturating_sub(line.len()), + instrumentation, + label, + )? { + break; + } + // `File::try_clone` may share the underlying file offset with + // `file`. The bounded preflight seeks that handle, so discard any + // BufReader read-ahead and restore this reader's logical position. + reader.seek(SeekFrom::Start(line_end)).with_context(|| { + format!("restore transcript {label} after checkpoint preflight") + })?; + } + cursor = line_end; if let DecodeResult::Checkpoint(checkpoint) = decoder.push_complete_line(path, line, cursor)? { @@ -261,6 +289,53 @@ pub(super) fn read_forward_batch( }) } +fn remaining_checkpoint_lines_fit( + file: &mut File, + start: u64, + committed_end: u64, + mut remaining_lines: u64, + byte_budget: usize, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + if remaining_lines == 0 { + return Ok(true); + } + if start >= committed_end { + // Force the decoder to report the incomplete checkpoint instead of + // treating malformed metadata as a budget-limited boundary. + return Ok(true); + } + let budget_end = start + .saturating_add(u64::try_from(byte_budget).unwrap_or(u64::MAX)) + .min(committed_end); + let mut cursor = start; + let mut buffer = vec![0_u8; REVERSE_SCAN_CHUNK_BYTES]; + while cursor < budget_end { + let count = + usize::try_from((budget_end - cursor).min(buffer.len() as u64)).unwrap_or(buffer.len()); + read_exact_at(file, cursor, &mut buffer[..count], instrumentation, label)?; + for byte in &buffer[..count] { + if *byte == b'\n' { + remaining_lines -= 1; + if remaining_lines == 0 { + return Ok(true); + } + } + } + cursor = cursor + .checked_add(count as u64) + .context("transcript byte offset overflow")?; + } + if budget_end == committed_end { + // The declaration runs past the committed boundary. Decode it so the + // shared validator returns the structural error on this call. + Ok(true) + } else { + Ok(false) + } +} + pub(super) fn timeline_records(epoch: u64, checkpoint: CommittedCheckpoint) -> Vec { checkpoint .records diff --git a/src/storage/message_log/transcript/tests.rs b/src/storage/message_log/transcript/tests.rs index 4eb49be..37201f5 100644 --- a/src/storage/message_log/transcript/tests.rs +++ b/src/storage/message_log/transcript/tests.rs @@ -17,6 +17,8 @@ use crate::{ trajectory::message_ref, }; +mod review; + async fn test_store(state: RunState) -> (TempDir, RunDirStore) { let workspace = tempfile::tempdir().unwrap(); let store = RunDirStore::new(workspace.path()); diff --git a/src/storage/message_log/transcript/tests/review.rs b/src/storage/message_log/transcript/tests/review.rs new file mode 100644 index 0000000..45a3864 --- /dev/null +++ b/src/storage/message_log/transcript/tests/review.rs @@ -0,0 +1,248 @@ +use super::*; + +#[tokio::test] +async fn concurrent_pending_truncate_retries_without_polluting_suffix_state() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let committed_end = std::fs::metadata(&paths.messages).unwrap().len(); + let discarded = raw_checkpoint( + 2, + vec![ + Message::text(Role::Assistant, format!("discarded-{}", "x".repeat(4096))), + Message::text(Role::User, "missing"), + ], + ); + let replacement = raw_checkpoint(2, vec![Message::text(Role::Assistant, "replacement")]); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw(&paths.messages, &discarded[..1]); + + let mut attempts = Vec::new(); + let refresh = timeline + .refresh_timeline_with_hook(|attempt| { + attempts.push(attempt); + if attempt == 1 { + StdOpenOptions::new() + .write(true) + .open(&paths.messages) + .unwrap() + .set_len(committed_end) + .unwrap(); + append_raw(&paths.messages, &replacement); + } + Ok(()) + }) + .unwrap(); + assert!(matches!(refresh, TimelineRefresh::Appended(_))); + assert_eq!(attempts, [1, 2]); + assert_eq!(timeline.snapshot().epoch, 1); + + let (records, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(records, ["m2"]); + assert_eq!(next, TimelineReadNext::Pending); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::NoChange(_) + )); + assert!(matches!( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + TimelineRead::Pending + )); +} + +#[tokio::test] +async fn forward_budget_stops_before_a_huge_checkpoint_after_a_small_append() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "initial")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "small append")]), + ); + append_raw( + &paths.messages, + &raw_checkpoint( + 3, + (0..256) + .map(|index| { + Message::text(Role::User, format!("large-{index}-{}", "x".repeat(4096))) + }) + .collect(), + ), + ); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + + let before = timeline.instrumentation(); + let (small, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap(), + ); + let after_small = timeline.instrumentation(); + assert_eq!(small, ["m2"]); + assert_eq!(next, TimelineReadNext::More); + assert!(after_small.bytes_read - before.bytes_read < 64 * 1024); + + let (large, next) = record_refs(timeline.load_newer(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(large.len(), 256); + assert_eq!(large.first().unwrap(), "m3"); + assert_eq!(large.last().unwrap(), "m258"); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn forward_byte_preflight_restores_the_buffered_reader_position() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "initial")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "small append")]), + ); + append_raw( + &paths.messages, + &raw_checkpoint( + 3, + vec![ + Message::text(Role::Assistant, "first in group"), + Message::text(Role::User, "x".repeat(32 * 1024)), + ], + ), + ); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + + let (records, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(4, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(records, ["m2", "m3", "m4"]); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn forward_byte_budget_scans_only_the_bounded_prefix_of_the_next_checkpoint() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "initial")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "small append")]), + ); + append_raw( + &paths.messages, + &raw_checkpoint( + 3, + vec![ + Message::text(Role::Assistant, "small first line"), + Message::text(Role::User, "x".repeat(1024 * 1024)), + ], + ), + ); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + + let before = timeline.instrumentation(); + let (small, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(10, 8 * 1024)) + .unwrap(), + ); + let after_small = timeline.instrumentation(); + assert_eq!(small, ["m2"]); + assert_eq!(next, TimelineReadNext::More); + assert!(after_small.bytes_read - before.bytes_read < 32 * 1024); + + let (large, next) = record_refs(timeline.load_newer(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(large, ["m3", "m4"]); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn forward_preflight_rejects_malformed_checkpoint_start_before_budget_skip() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + append_raw( + &paths.messages, + &raw_checkpoint(1, vec![Message::text(Role::Assistant, "valid prefix")]), + ); + let mut malformed = raw_checkpoint(2, vec![Message::text(Role::User, "malformed")]); + let mut value: serde_json::Value = serde_json::from_slice(&malformed[0]).unwrap(); + value["_fiasco"]["checkpoint"]["index"] = serde_json::json!(1_u64); + value["_fiasco"]["checkpoint"]["count"] = serde_json::json!(1_000_000_u64); + malformed[0] = serde_json::to_vec(&value).unwrap(); + malformed[0].push(b'\n'); + append_raw(&paths.messages, &malformed); + append_raw( + &paths.messages, + &raw_checkpoint(3, vec![Message::text(Role::Assistant, "valid tail")]), + ); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + + let error = timeline + .probe_prefix(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap_err(); + assert!(error.to_string().contains("starts at index 1 instead of 0")); +} + +#[tokio::test] +async fn skipped_malformed_older_checkpoint_makes_progress_then_errors() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + let mut malformed = raw_checkpoint(1, vec![Message::text(Role::User, "malformed")]); + let mut value: serde_json::Value = serde_json::from_slice(&malformed[0]).unwrap(); + value["_fiasco"]["checkpoint"]["count"] = serde_json::json!(1_000_000_u64); + malformed[0] = serde_json::to_vec(&value).unwrap(); + malformed[0].push(b'\n'); + append_raw(&paths.messages, &malformed); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "tail")]), + ); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + + let (tail, next) = record_refs( + timeline + .load_older(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(tail, ["m2"]); + assert_eq!(next, TimelineReadNext::More); + let error = timeline + .load_older(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap_err(); + assert!( + error + .to_string() + .contains("incomplete checkpoint ends inside committed transcript") + ); +}