fix(inspect): bound batches across writer races

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"

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
 

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 <run-id>` opens a tail-first snapshot and

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,

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<u64> {
+        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<DecodeResult> {
         let pending = self
             .pending

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<TimelineRefresh> {
+        self.refresh_timeline_with_hook(|_| Ok(()))
+    }
+
+    pub(super) fn refresh_timeline_with_hook(
+        &mut self,
+        mut after_stat: impl FnMut(usize) -> Result<()>,
+    ) -> Result<TimelineRefresh> {
+        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<FileObservation>,
+        after_stat: impl FnOnce() -> Result<()>,
+    ) -> Result<TimelineRefresh> {
         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<SystemTime>,
+}
+
+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<FileObservation> {
+    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::<io::Error>()
+            .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