fix(session): bound the session journal, work graph and metadata read (#6842)

A 6-day session kept 49,258 journal entries (99.6% off the active branch),
a 98.7 MB document and 872 MB RSS. Three unbounded growers, fixed together:

1. Journal. `SessionJournal::prune_plan` (pure) keeps the root->leaf path and
   whole recent off-branch chains (Compaction/BranchSummary first) within a
   1,024-entry budget once more than 4,096 entries sit off-branch;
   `remove_entries` is atomic (validates or restores). Autosaves on the
   persistence actor (`persistence_actor.rs` flush_inner ->
   `SessionManager::save_session_bounded`), never the UI loop, append the
   planned entries to `sessions/.journal-archive/<id>.jsonl` (0600, fsync,
   dir fsync on create) and only then write the pruned document. Archive
   failure prunes nothing. Archived ids are handed to the UI through an
   in-memory registry and dropped from the live journal in
   `build_session_snapshot` (no I/O), so RAM matches disk. Plain
   `save_session` never prunes. `/branch <id>` restores an archived entry and
   its missing ancestors; `/tree` reports the archived count; export merges
   the archive into the container; session delete removes it; a Runtime API
   fork of an archived entry fails closed with a message naming the archive.

2. Work graph. New reducer change `PruneEndedOperations` keeps the newest
   256 ended, non-durable Operation nodes (not Stale, no non-Contains edges,
   not referenced by activities or proposals); applied at registration.
   These are a derived index of tool calls already in the transcript.

3. Metadata prefix. `load_session_metadata` grows its read geometrically
   (to 16 MB) until the metadata block closes instead of reading the whole
   file. `usage_source_fingerprints` is deliberately not capped: it is the
   money replay-idempotency set; documented as a known limit.

Not addressed here: snapshot construction still runs on the UI task (now
cheap with a bounded journal); resume still reads the document twice; a
child forked before a prune cannot /branch into the parent's archive.

Evidence (local, this checkout):
- cargo test -p codewhale-runtime --lib session_tree:
  test result: ok. 22 passed; 0 failed (4 new prune tests)
- cargo test -p codewhale-tui --lib -- journal_bound_tests
  canonical_journal_admission_tests ended_shell_operations
  container_includes_journal_archive:
  test result: ok. 9 passed; 0 failed
  (bounded save + live-journal follow, archive failure prunes nothing,
  archive-then-document-write-failure loses nothing, /branch to archived id
  round-trip, delete removes archive, >64 KB metadata listing, work-graph
  cap, export merge, existing superseded-entries-survive test unchanged)
- Negative control: with the work-graph prune disabled,
  ended_shell_operations_are_capped_oldest_first FAILED (0 passed; 1 failed).
- cargo test -p codewhale-tui --lib -- session_manager:: session_export::
  persistence_actor:: work_graph:: thread_history session_tree
  commands::contract: test result: FAILED. 290 passed; 1 failed. The one
  failure, control_hosted_work_target_resolves_and_never_echoes_credentials,
  expects repo slug Hmbown/CodeWhale but the checkout's remote is
  codewhale-hq/CodeWhale; unrelated to this change.
- RUST_MIN_STACK=16777216 cargo test -p codewhale-tui --lib -- fork canonical:
  test result: ok. 232 passed; 0 failed
- python3 scripts/check-blocking-calls-budget.py: 530 sites across 165
  files, within budget
- python3 scripts/check-dead-code-budget.py: 253 suppressions, budget 258

Refs #6842

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: CodeWhale Bot <[email protected]>

crates/runtime/src/session_tree.rs

@@ -363,6 +363,132 @@ impl SessionJournal {
             );
         }
     }
+
+    /// Ids a bounded save may move out of this journal, in journal order.
+    ///
+    /// Empty until more than `dead_threshold` entries sit off the active
+    /// branch — superseded versions left behind by compaction, undo and
+    /// edits (#6842). Past it, the plan keeps the root→leaf path, then whole
+    /// off-branch chains (an entry with every ancestor) newest first, with
+    /// `Compaction`/`BranchSummary` chains ahead of plain leaves, while the
+    /// kept off-branch total stays within `dead_budget`. A kept entry never
+    /// loses its parent, so the remainder always passes [`Self::validate`].
+    ///
+    /// Pure: callers archive the planned entries durably before removing
+    /// them with [`Self::remove_entries`]. Known limit: the active branch
+    /// itself is never pruned, so a session that never compacts stays as
+    /// long as its conversation.
+    pub fn prune_plan(&self, dead_threshold: usize, dead_budget: usize) -> Vec<EntryId> {
+        let active = self.root_to_leaf();
+        if self.entries.len().saturating_sub(active.len()) <= dead_threshold {
+            return Vec::new();
+        }
+        let index: HashMap<&str, usize> = self
+            .entries
+            .iter()
+            .enumerate()
+            .map(|(i, e)| (e.id.as_str(), i))
+            .collect();
+        let mut keep = vec![false; self.entries.len()];
+        for entry in &active {
+            if let Some(&i) = index.get(entry.id.as_str()) {
+                keep[i] = true;
+            }
+        }
+        let parents: HashSet<&str> = self
+            .entries
+            .iter()
+            .filter_map(|e| e.parent_id.as_deref())
+            .collect();
+        let summary = |e: &SessionEntry| {
+            matches!(
+                e.kind,
+                SessionEntryKind::Compaction { .. } | SessionEntryKind::BranchSummary { .. }
+            )
+        };
+        // Newest first: journal order is append order.
+        let candidates = self
+            .entries
+            .iter()
+            .enumerate()
+            .rev()
+            .filter(|(_, e)| summary(e))
+            .chain(
+                self.entries
+                    .iter()
+                    .enumerate()
+                    .rev()
+                    .filter(|(_, e)| !parents.contains(e.id.as_str())),
+            );
+        let mut used = 0usize;
+        for (start, _) in candidates {
+            if keep[start] {
+                continue;
+            }
+            let mut chain = Vec::new();
+            let mut seen = HashSet::new();
+            let mut cursor = Some(start);
+            let mut closed = true;
+            while let Some(i) = cursor {
+                if keep[i] {
+                    break;
+                }
+                if !seen.insert(i) || used + chain.len() >= dead_budget {
+                    closed = false;
+                    break;
+                }
+                chain.push(i);
+                cursor = match self.entries[i].parent_id.as_deref() {
+                    None => None,
+                    Some(parent) => match index.get(parent) {
+                        Some(&p) => Some(p),
+                        None => {
+                            closed = false;
+                            None
+                        }
+                    },
+                };
+            }
+            if closed {
+                used += chain.len();
+                for i in chain {
+                    keep[i] = true;
+                }
+            }
+        }
+        self.entries
+            .iter()
+            .zip(keep)
+            .filter(|(_, kept)| !kept)
+            .map(|(e, _)| e.id.clone())
+            .collect()
+    }
+
+    /// Remove `ids` atomically: either every listed entry leaves and the
+    /// remainder still validates, or the journal is left exactly as it was
+    /// and the validation error is returned. Returns how many were removed.
+    pub fn remove_entries(&mut self, ids: &HashSet<EntryId>) -> Result<usize, String> {
+        if ids.is_empty() {
+            return Ok(0);
+        }
+        let original = std::mem::take(&mut self.entries);
+        let mut removed = Vec::new();
+        for (position, entry) in original.into_iter().enumerate() {
+            if ids.contains(&entry.id) {
+                removed.push((position, entry));
+            } else {
+                self.entries.push(entry);
+            }
+        }
+        if let Err(error) = self.validate() {
+            // Ascending positions put every entry back where it was.
+            for (position, entry) in removed {
+                self.entries.insert(position, entry);
+            }
+            return Err(error);
+        }
+        Ok(removed.len())
+    }
 }
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct SessionImportContainer {
@@ -931,4 +1057,80 @@ mod tests {
             }
         }
     }
+
+    /// A compaction-heavy session: each round rebranches the whole tail, so
+    /// superseded copies pile up off the active branch (#6842).
+    fn compacted_journal(rounds: usize, tail: usize) -> SessionJournal {
+        let mut j = SessionJournal::new();
+        for round in 0..rounds {
+            let messages: Vec<Message> = (0..tail)
+                .map(|i| msg("user", &format!("round {round} message {i}")))
+                .collect();
+            j.rebranch_active_messages(&messages);
+        }
+        j
+    }
+
+    #[test]
+    fn prune_plan_is_empty_under_the_threshold() {
+        let j = compacted_journal(4, 10);
+        assert_eq!(j.len(), 40);
+        assert!(j.prune_plan(30, 5).is_empty());
+    }
+
+    #[test]
+    fn prune_plan_keeps_active_branch_and_whole_recent_chains() {
+        let mut j = compacted_journal(50, 10);
+        let active_before = j.active_messages(true);
+        let leaf = j.leaf_id.clone();
+        let plan = j.prune_plan(100, 25);
+        // 490 dead entries; 2 whole 10-entry chains fit a 25 budget.
+        assert_eq!(plan.len(), 490 - 20);
+        let ids: HashSet<EntryId> = plan.into_iter().collect();
+        assert_eq!(j.remove_entries(&ids).unwrap(), 470);
+        assert_eq!(j.len(), 30);
+        j.validate().unwrap();
+        assert_eq!(j.leaf_id, leaf);
+        assert_eq!(j.active_messages(true), active_before);
+        // The newest superseded round survives, not an arbitrary one.
+        let kept: Vec<String> = j
+            .entries
+            .iter()
+            .filter_map(|e| e.kind.as_message())
+            .map(|m| match &m.content[0] {
+                ContentBlock::Text { text, .. } => text.clone(),
+                _ => String::new(),
+            })
+            .collect();
+        assert!(kept.iter().any(|t| t.starts_with("round 48 ")), "{kept:?}");
+        assert!(!kept.iter().any(|t| t.starts_with("round 0 ")), "{kept:?}");
+    }
+
+    #[test]
+    fn prune_plan_prefers_summary_chains() {
+        let mut j = compacted_journal(30, 4);
+        let leaf = j.leaf_id.clone().unwrap();
+        // An old compaction hanging off the very first entry.
+        let first = j.entries[0].id.clone();
+        j.branch_to(&first).unwrap();
+        let summary = j.append_compaction("old summary".into(), None, None, None);
+        j.branch_to(&leaf).unwrap();
+        let plan = j.prune_plan(10, 2);
+        assert!(!plan.contains(&summary));
+        assert!(!plan.contains(&first));
+    }
+
+    #[test]
+    fn remove_entries_refuses_to_orphan_and_leaves_journal_untouched() {
+        let mut j = compacted_journal(3, 3);
+        let before = j.clone();
+        let active: HashSet<EntryId> = j
+            .root_to_leaf()
+            .iter()
+            .map(|e| e.id.clone())
+            .take(1)
+            .collect();
+        assert!(j.remove_entries(&active).is_err());
+        assert_eq!(j, before);
+    }
 }

crates/tui/src/commands/contract.rs

@@ -347,6 +347,13 @@ impl CommandSessionLifecycleContext for SessionLifecycleAdapter<'_> {
             .as_ref()
             .map(|j| j.entries.len())
             .unwrap_or(0);
+        // An entry a bounded save archived (#6842) is restored, with any
+        // ancestors the journal no longer holds, before branching to it.
+        if let Err(e) = manager.restore_archived_journal_chain(&mut session, entry_id) {
+            return Err(format!(
+                "branch failed: could not restore {entry_id} from the session's journal archive: {e}"
+            ));
+        }
         match session.journal_branch_to(entry_id) {
             Ok(()) => {
                 if let Err(e) = manager.save_session(&session) {
@@ -398,7 +405,22 @@ impl CommandSessionLifecycleContext for SessionLifecycleAdapter<'_> {
             if let Ok(mut session) = manager.load_session(&session_id) {
                 session.ensure_journal();
                 if let Some(journal) = session.journal.as_ref() {
-                    let rendered = crate::session_tree::render_tree(journal);
+                    let mut rendered = crate::session_tree::render_tree(journal);
+                    match manager.load_journal_archive(&session_id) {
+                        Ok(archived) if !archived.is_empty() => {
+                            rendered.push_str(&format!(
+                                "{} older off-branch entries are archived in {}/{}.jsonl \
+                                 (not drawn); `/branch <id>` restores one.\n",
+                                archived.len(),
+                                crate::session_manager::JOURNAL_ARCHIVE_DIR,
+                                session_id,
+                            ));
+                        }
+                        Ok(_) => {}
+                        Err(e) => {
+                            rendered.push_str(&format!("journal archive could not be read: {e}\n"))
+                        }
+                    }
                     return Ok(TreeBodyProjection::Journal { rendered });
                 }
             }

crates/tui/src/runtime_api/thread_history.rs

@@ -4,7 +4,7 @@ use std::collections::HashMap;
 use std::path::Path;
 use std::sync::Arc;
 
-use anyhow::{Context, Result, anyhow, ensure};
+use anyhow::{Context, Result, anyhow, bail, ensure};
 use axum::{Json, extract::State};
 use chrono::{DateTime, Utc};
 use codewhale_models::{ContentBlock, Message, Role};
@@ -1323,9 +1323,35 @@ async fn mutate_in_owner(
             selected_entry_id, ..
         } = &request.mutation
         {
-            let fork = journal
-                .fork_from(selected_entry_id.as_deref())
-                .map_err(anyhow::Error::msg)?;
+            let fork = match journal.fork_from(selected_entry_id.as_deref()) {
+                Ok(fork) => fork,
+                Err(error) => {
+                    // An entry a bounded TUI save archived (#6842) is not in
+                    // the document; fail closed and say where it went.
+                    let dir = sessions_dir.clone();
+                    let id = session.metadata.id.clone();
+                    let entry = selected_entry_id.clone();
+                    let archived = history_owner_work(move || {
+                        let Some(entry) = entry else {
+                            return Ok(false);
+                        };
+                        Ok(crate::session_manager::load_journal_archive(&dir, &id)?
+                            .iter()
+                            .any(|archived| archived.id == entry))
+                    })
+                    .await
+                    .unwrap_or(false);
+                    if archived {
+                        bail!(
+                            "{error}: the entry was moved to this session's journal archive to keep \
+                             the saved document bounded; restore it with `/branch {}` first. \
+                             Source retained",
+                            selected_entry_id.as_deref().unwrap_or_default()
+                        );
+                    }
+                    return Err(anyhow::Error::msg(error));
+                }
+            };
             session.metadata.mark_forked_from(&session.metadata.clone());
             session.metadata.spawn_depth = fork.spawn_depth;
             session.leaf_id = fork.leaf_id.clone();

crates/tui/src/session_export.rs

@@ -159,7 +159,7 @@ pub fn write_session_archive(
         ));
     }
     let session_json = session_json(session)?;
-    let container_json = container_json(session)?;
+    let container_json = container_json(session, sessions_dir)?;
     let parent = output
         .parent()
         .filter(|parent| !parent.as_os_str().is_empty())
@@ -299,8 +299,24 @@ fn session_json(session: &SavedSession) -> io::Result<String> {
         .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
 }
 
-fn container_json(session: &SavedSession) -> io::Result<String> {
-    let container: SessionImportContainer = session.export_container("session-archive");
+/// The portable journal container. With a session store it also carries the
+/// entries bounded saves moved into the journal archive (#6842), so an export
+/// holds the full history, not only what the document still keeps.
+fn container_json(session: &SavedSession, sessions_dir: Option<&Path>) -> io::Result<String> {
+    let mut container: SessionImportContainer = session.export_container("session-archive");
+    if let Some(root) = sessions_dir {
+        let archived = crate::session_manager::load_journal_archive(root, &session.metadata.id)?;
+        if !archived.is_empty() {
+            let present: std::collections::HashSet<String> =
+                container.entries.iter().map(|e| e.id.clone()).collect();
+            let mut entries: Vec<_> = archived
+                .into_iter()
+                .filter(|e| !present.contains(&e.id))
+                .collect();
+            entries.append(&mut container.entries);
+            container.entries = entries;
+        }
+    }
     serde_json::to_string_pretty(&container)
         .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
 }
@@ -1013,4 +1029,46 @@ mod tests {
             );
         }
     }
+
+    /// #6842: entries a bounded save archived travel with the export.
+    #[test]
+    fn container_includes_journal_archive_entries() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let sessions = dir.path().join("sessions");
+        let session = fixture_session();
+        let journal = session.journal.clone().expect("journal");
+        let root = journal.entries[0].id.clone();
+        let mut archived = journal.clone();
+        archived.branch_to(&root).expect("branch");
+        let extra = archived.append_message(Message {
+            role: Role::User,
+            content: vec![ContentBlock::Text {
+                text: "archived alternative".to_string(),
+                cache_control: None,
+            }],
+        });
+        let entry = archived.entries.last().expect("entry").clone();
+        assert_eq!(entry.id, extra);
+        let archive_dir = sessions.join(crate::session_manager::JOURNAL_ARCHIVE_DIR);
+        fs::create_dir_all(&archive_dir).expect("archive dir");
+        let path = archive_dir.join(format!("{}.jsonl", session.metadata.id));
+        fs::write(
+            &path,
+            format!("{}\n", serde_json::to_string(&entry).unwrap()),
+        )
+        .unwrap();
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt as _;
+            fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap();
+        }
+
+        let without: SessionImportContainer =
+            serde_json::from_str(&container_json(&session, None).unwrap()).unwrap();
+        let with: SessionImportContainer =
+            serde_json::from_str(&container_json(&session, Some(&sessions)).unwrap()).unwrap();
+        assert_eq!(with.entries.len(), without.entries.len() + 1);
+        assert!(with.entries.iter().any(|e| e.id == extra));
+        with.into_journal().expect("merged container validates");
+    }
 }

crates/tui/src/session_manager.rs

@@ -45,6 +45,16 @@ const MAX_EMPTY_SESSION_STUBS: usize = 10;
 /// picker's rename prompt has always enforced.
 pub const MAX_SESSION_TITLE_CHARS: usize = 100;
 pub(crate) const WORK_GRAPH_IMPORT_ARCHIVE_DIR: &str = ".work-graph-import-archive";
+/// Per-session JSONL sidecar holding journal entries a bounded save moved out
+/// of the session document (#6842): `<sessions>/.journal-archive/<id>.jsonl`.
+pub(crate) const JOURNAL_ARCHIVE_DIR: &str = ".journal-archive";
+/// Off-branch entries a session document may carry before an autosave
+/// archives the excess. Together with [`JOURNAL_RETAINED_DEAD_ENTRIES`] this
+/// keeps a compacted journal far below `MAX_CANONICAL_HISTORY_ENTRIES`.
+const JOURNAL_PRUNE_DEAD_THRESHOLD: usize = 4_096;
+/// Recent off-branch entries kept in the document after archiving, so `/tree`
+/// and `/branch` stay cheap for recent history.
+const JOURNAL_RETAINED_DEAD_ENTRIES: usize = 1_024;
 const SESSION_GOALS_DIR: &str = ".goals";
 const CURRENT_SESSION_GOAL_SCHEMA_VERSION: u32 = 2;
 const MAX_SESSION_GOAL_OBJECTIVE_CHARS: usize = 8_192;
@@ -686,6 +696,11 @@ pub struct SessionCostSnapshot {
     pub route_receipts: BTreeSet<String>,
     /// Redacted provider-response identities already included in the live and
     /// durable sub-agent totals. Worker records persist the same fingerprints.
+    ///
+    /// Known limit (#6842): unbounded, ~70 bytes per background call. It is
+    /// the replay-idempotency set for money (late-usage overlay and restored
+    /// dedupe), so it is never capped; `load_session_metadata` grows its read
+    /// to fit a large block instead of falling back to a whole-file read.
     #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
     pub usage_source_fingerprints: BTreeSet<String>,
     #[serde(
@@ -1398,6 +1413,94 @@ fn validate_saved_journal(journal: &SessionJournal) -> io::Result<()> {
     Ok(())
 }
 
+/// Archived ids per session that the live app has not yet dropped from its
+/// in-memory journal. A bounded save records them only after its document
+/// write lands; the UI takes them on its next snapshot (no I/O) so RAM
+/// matches disk, and a racing save skips re-appending ids still listed here.
+fn archived_journal_ids()
+-> &'static std::sync::Mutex<std::collections::HashMap<String, std::collections::HashSet<String>>> {
+    static IDS: std::sync::OnceLock<
+        std::sync::Mutex<std::collections::HashMap<String, std::collections::HashSet<String>>>,
+    > = std::sync::OnceLock::new();
+    IDS.get_or_init(Default::default)
+}
+
+/// Take the ids a bounded save archived for `session_id`, for the live
+/// journal to drop. Never touches the disk.
+pub(crate) fn take_archived_journal_ids(session_id: &str) -> std::collections::HashSet<String> {
+    archived_journal_ids()
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .remove(session_id)
+        .unwrap_or_default()
+}
+
+fn journal_archive_path(sessions_dir: &Path, session_id: &str) -> io::Result<PathBuf> {
+    let id = session_id.trim();
+    let mut components = Path::new(id).components();
+    if id.is_empty()
+        || !matches!(components.next(), Some(Component::Normal(_)))
+        || components.next().is_some()
+    {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "invalid session id for journal archive",
+        ));
+    }
+    Ok(sessions_dir
+        .join(JOURNAL_ARCHIVE_DIR)
+        .join(format!("{id}.jsonl")))
+}
+
+/// Every entry archived for `session_id`, oldest first, deduplicated by id
+/// (a save that crashed before its document landed re-archives the same
+/// entries). A torn final line — a crash mid-append, whose entries are still
+/// in the document — is skipped; damage anywhere else is an error.
+pub(crate) fn load_journal_archive(
+    sessions_dir: &Path,
+    session_id: &str,
+) -> io::Result<Vec<SessionEntry>> {
+    use std::io::Read as _;
+    let path = journal_archive_path(sessions_dir, session_id)?;
+    let mut file = match open_private_read_file(&path) {
+        Ok(file) => file,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(error) => return Err(error),
+    };
+    let mut raw = String::new();
+    file.read_to_string(&mut raw)?;
+    let lines: Vec<&str> = raw.lines().filter(|line| !line.trim().is_empty()).collect();
+    let mut seen = std::collections::HashSet::new();
+    let mut entries = Vec::with_capacity(lines.len());
+    for (index, line) in lines.iter().enumerate() {
+        match serde_json::from_str::<SessionEntry>(line) {
+            Ok(entry) => {
+                if seen.insert(entry.id.clone()) {
+                    entries.push(entry);
+                }
+            }
+            Err(error) if index + 1 == lines.len() && !raw.ends_with('\n') => {
+                tracing::warn!(
+                    path = %path.display(),
+                    %error,
+                    "skipping torn final journal-archive line"
+                );
+            }
+            Err(error) => {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    format!(
+                        "journal archive {} line {}: {error}",
+                        path.display(),
+                        index + 1
+                    ),
+                ));
+            }
+        }
+    }
+    Ok(entries)
+}
+
 fn serialize_saved_session(mut session: SavedSession) -> io::Result<String> {
     if let Some(journal) = session.journal.as_ref() {
         validate_saved_journal(journal)?;
@@ -2497,6 +2600,20 @@ impl SessionManager {
 
     /// Save a session to disk, consuming it.
     pub(crate) fn save_session_owned(&self, session: SavedSession) -> std::io::Result<PathBuf> {
+        self.save_session_inner(session, false)
+    }
+
+    /// The autosave path (#6842): like [`Self::save_session_owned`], but when
+    /// the journal carries more than `JOURNAL_PRUNE_DEAD_THRESHOLD`
+    /// off-branch entries, first append the excess to the session's journal
+    /// archive (fsynced), then write the document without them. If the
+    /// archive cannot be written nothing is pruned. Runs on the persistence
+    /// actor (`tui/persistence_actor.rs` `flush_inner`), never the UI loop.
+    pub(crate) fn save_session_bounded(&self, session: SavedSession) -> std::io::Result<PathBuf> {
+        self.save_session_inner(session, true)
+    }
+
+    fn save_session_inner(&self, session: SavedSession, bound: bool) -> std::io::Result<PathBuf> {
         let session_id = session.metadata.id.clone();
         let path = self.validated_session_path(&session_id)?;
         // Not a `move` closure: `session` is consumed inside, so inference
@@ -2512,6 +2629,11 @@ impl SessionManager {
             self.archive_before_first_graph_write(&session, &path)?;
 
             let mut durable_session = session;
+            let archived = if bound {
+                self.archive_dead_journal_entries(&mut durable_session)
+            } else {
+                Vec::new()
+            };
             self.hydrate_recovered_runtime_binding(&mut durable_session)?;
             self.hydrate_approval_receipts(&mut durable_session)?;
             let content = serialize_saved_session(durable_session)?;
@@ -2519,6 +2641,14 @@ impl SessionManager {
             // Atomic write via write_atomic (NamedTempFile + fsync + persist)
             write_atomic(&path, content.as_bytes())?;
             self.stamp_session_boot_owner_for_new_record(&session_id, already_persisted);
+            if !archived.is_empty() {
+                archived_journal_ids()
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner)
+                    .entry(session_id.clone())
+                    .or_default()
+                    .extend(archived);
+            }
             Ok(())
         })?
         .ok_or_else(Self::retired_session_write_error)?;
@@ -2667,6 +2797,189 @@ impl SessionManager {
         });
     }
 
+    /// Move the journal's excess off-branch entries into the archive sidecar.
+    /// Order is the data-loss guard: plan on the unchanged journal, append and
+    /// fsync the planned entries, and only then remove them from the document
+    /// about to be written. Any failure leaves the journal whole. Returns the
+    /// removed ids.
+    fn archive_dead_journal_entries(&self, session: &mut SavedSession) -> Vec<String> {
+        let session_id = session.metadata.id.clone();
+        let Some(journal) = session.journal.as_mut() else {
+            return Vec::new();
+        };
+        let plan = journal.prune_plan(JOURNAL_PRUNE_DEAD_THRESHOLD, JOURNAL_RETAINED_DEAD_ENTRIES);
+        if plan.is_empty() {
+            return Vec::new();
+        }
+        let ids: std::collections::HashSet<String> = plan.into_iter().collect();
+        // Ids a previous save archived but the live journal still carries
+        // are already durable; append only the rest.
+        let already = archived_journal_ids()
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(&session_id)
+            .cloned()
+            .unwrap_or_default();
+        let fresh: Vec<&SessionEntry> = journal
+            .entries
+            .iter()
+            .filter(|entry| ids.contains(&entry.id) && !already.contains(&entry.id))
+            .collect();
+        if let Err(error) = self.append_journal_archive(&session_id, &fresh) {
+            tracing::warn!(
+                %session_id,
+                %error,
+                "journal archive write failed; keeping the full journal"
+            );
+            return Vec::new();
+        }
+        match journal.remove_entries(&ids) {
+            Ok(_) => {
+                session.leaf_id = journal.leaf_id.clone();
+                ids.into_iter().collect()
+            }
+            Err(error) => {
+                tracing::warn!(%session_id, %error, "journal prune refused; keeping the full journal");
+                Vec::new()
+            }
+        }
+    }
+
+    /// Append `entries` to the session's journal archive, one JSON object per
+    /// line, and fsync before returning. The sidecar is owner-only and never
+    /// followed through a link.
+    fn append_journal_archive(
+        &self,
+        session_id: &str,
+        entries: &[&SessionEntry],
+    ) -> io::Result<()> {
+        use std::io::{Seek as _, SeekFrom, Write as _};
+        if entries.is_empty() {
+            return Ok(());
+        }
+        let path = journal_archive_path(&self.sessions_dir, session_id)?;
+        let dir = self.sessions_dir.join(JOURNAL_ARCHIVE_DIR);
+        if let Ok(metadata) = fs::symlink_metadata(&dir)
+            && crate::plugins::metadata_is_link_or_reparse(&metadata)
+        {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "journal archive directory is a link",
+            ));
+        }
+        let created_dir = !dir.exists();
+        fs::create_dir_all(&dir)?;
+        let mut buf = Vec::new();
+        for entry in entries {
+            serde_json::to_writer(&mut buf, entry)
+                .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
+            buf.push(b'\n');
+        }
+        let created_file = !path.exists();
+        let mut file = open_private_lock_file(&path)?;
+        file.seek(SeekFrom::End(0))?;
+        file.write_all(&buf)?;
+        file.sync_all()?;
+        #[cfg(unix)]
+        if created_file || created_dir {
+            // Make the new directory entries durable, not only the bytes.
+            fs::File::open(&dir)?.sync_all()?;
+            if created_dir {
+                fs::File::open(&self.sessions_dir)?.sync_all()?;
+            }
+        }
+        #[cfg(not(unix))]
+        let _ = (created_file, created_dir);
+        Ok(())
+    }
+
+    /// Entries archived out of `session_id`'s journal by bounded saves.
+    pub(crate) fn load_journal_archive(&self, session_id: &str) -> io::Result<Vec<SessionEntry>> {
+        load_journal_archive(&self.sessions_dir, self.validated_session_id(session_id)?)
+    }
+
+    /// Bring an archived entry, and every ancestor the journal no longer
+    /// holds, back into `session`'s journal so `/branch` can select it.
+    /// `Ok(false)` when `entry_id` is in neither the journal nor the archive.
+    pub(crate) fn restore_archived_journal_chain(
+        &self,
+        session: &mut SavedSession,
+        entry_id: &str,
+    ) -> io::Result<bool> {
+        session.ensure_journal();
+        let session_id = session.metadata.id.clone();
+        let journal = session.journal.as_mut().expect("journal ensured");
+        if journal.contains(entry_id) {
+            return Ok(true);
+        }
+        let archive = self.load_journal_archive(&session_id)?;
+        let by_id: std::collections::HashMap<&str, &SessionEntry> = archive
+            .iter()
+            .map(|entry| (entry.id.as_str(), entry))
+            .collect();
+        let mut chain = Vec::new();
+        let mut cursor = Some(entry_id);
+        while let Some(id) = cursor {
+            if journal.contains(id) {
+                break;
+            }
+            let Some(entry) = by_id.get(id) else {
+                if chain.is_empty() {
+                    return Ok(false);
+                }
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    format!("archived entry {entry_id} is missing ancestor {id}"),
+                ));
+            };
+            if chain.len() > by_id.len() {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "journal archive contains a cycle",
+                ));
+            }
+            chain.push((*entry).clone());
+            cursor = entry.parent_id.as_deref();
+        }
+        let restored: Vec<String> = chain.iter().map(|entry| entry.id.clone()).collect();
+        journal.entries.extend(chain.into_iter().rev());
+        validate_saved_journal(journal)?;
+        // The live app must not later drop what was just brought back.
+        if let Some(pending) = archived_journal_ids()
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get_mut(&session_id)
+        {
+            for id in &restored {
+                pending.remove(id);
+            }
+        }
+        Ok(true)
+    }
+
+    /// Remove the session's journal archive with the session itself. A
+    /// linked archive directory is not followed.
+    fn remove_journal_archive(&self, id: &str) -> io::Result<()> {
+        let dir = self.sessions_dir.join(JOURNAL_ARCHIVE_DIR);
+        match fs::symlink_metadata(&dir) {
+            Ok(metadata) if crate::plugins::metadata_is_link_or_reparse(&metadata) => {
+                tracing::warn!(
+                    path = %dir.display(),
+                    "journal archive directory is a link; its copies were not removed"
+                );
+                return Ok(());
+            }
+            Ok(_) => {}
+            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
+            Err(error) => return Err(error),
+        }
+        match fs::remove_file(journal_archive_path(&self.sessions_dir, id)?) {
+            Ok(()) => Ok(()),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
+            Err(error) => Err(error),
+        }
+    }
+
     /// Preserve the exact pre-import session once, before the first graph-
     /// bearing session or checkpoint write can replace it.
     fn archive_before_first_graph_write(
@@ -3472,9 +3785,25 @@ impl SessionManager {
             return Ok(metadata);
         }
 
-        // Metadata wasn't extractable from the prefix (truncated mid-block,
-        // unusual key ordering, etc.). Read the rest and try again with the
-        // full buffer before giving up.
+        // Metadata wasn't extractable from the prefix: usually a `metadata`
+        // block longer than 64 KB (a long-lived session's usage fingerprints,
+        // #6842), rarely unusual key ordering. Grow the read geometrically
+        // until the block closes, so the cost tracks the metadata's size
+        // rather than the transcript's; only past `GROWN_PREFIX_LIMIT` (or
+        // for a document that really lacks the block) read everything.
+        const GROWN_PREFIX_LIMIT: usize = 16 * 1024 * 1024;
+        let mut limit = PREFIX_BYTES;
+        while buf.len() == limit && limit < GROWN_PREFIX_LIMIT {
+            let next = (limit * 2).min(GROWN_PREFIX_LIMIT);
+            file.by_ref()
+                .take((next - limit) as u64)
+                .read_to_end(&mut buf)?;
+            limit = next;
+            if let Some(mut metadata) = extract_top_level_metadata(&buf) {
+                apply_legacy_title_recovery(&mut metadata, &buf);
+                return Ok(metadata);
+            }
+        }
         let mut rest = Vec::new();
         file.read_to_end(&mut rest)?;
         buf.extend_from_slice(&rest);
@@ -3648,6 +3977,7 @@ impl SessionManager {
             }
         }
         self.remove_import_archive_copy(&path)?;
+        self.remove_journal_archive(id)?;
         self.clear_session_boot_owner(id);
         let session_dir = self.sessions_dir.join(id.trim());
         if session_dir.exists() {
@@ -4987,3 +5317,237 @@ mod canonical_journal_admission_tests {
         );
     }
 }
+
+#[cfg(test)]
+mod journal_bound_tests {
+    //! #6842: bounded autosaves archive superseded journal entries before
+    //! dropping them from the session document.
+    use super::*;
+    use std::collections::HashSet;
+
+    fn text(text: &str) -> Message {
+        Message {
+            role: codewhale_models::Role::User,
+            content: vec![ContentBlock::Text {
+                text: text.into(),
+                cache_control: None,
+            }],
+        }
+    }
+
+    /// A session whose journal holds `rounds` superseded copies of a
+    /// 10-message tail, as repeated compaction leaves it.
+    fn compacted_session(id: &str, workspace: &Path, rounds: usize) -> SavedSession {
+        let mut session = create_saved_session_with_id_and_mode(
+            id.into(),
+            &[text("start")],
+            "test-model",
+            workspace,
+            0,
+            None,
+            None,
+        );
+        let journal = session.journal.as_mut().unwrap();
+        for round in 0..rounds {
+            let tail: Vec<Message> = (0..10)
+                .map(|i| text(&format!("round {round} message {i}")))
+                .collect();
+            journal.rebranch_active_messages(&tail);
+        }
+        session.leaf_id = journal.leaf_id.clone();
+        session.messages = journal.to_messages();
+        session.metadata.message_count = session.messages.len();
+        session
+    }
+
+    fn all_ids(entries: &[SessionEntry]) -> HashSet<String> {
+        entries.iter().map(|e| e.id.clone()).collect()
+    }
+
+    #[test]
+    fn bounded_save_archives_then_prunes_and_live_journal_follows() {
+        let tmp = tempfile::tempdir().unwrap();
+        let manager = SessionManager::new(tmp.path().join("sessions")).unwrap();
+        let session = compacted_session("bounded-a", tmp.path(), 460);
+        let original = session.journal.clone().unwrap();
+        let active = original.to_messages();
+
+        manager.save_session_bounded(session).unwrap();
+
+        let saved = manager.load_session("bounded-a").unwrap();
+        let journal = saved.journal.as_ref().unwrap();
+        assert!(
+            journal.len() <= 10 + JOURNAL_RETAINED_DEAD_ENTRIES,
+            "{}",
+            journal.len()
+        );
+        assert_eq!(journal.to_messages(), active);
+        assert_eq!(saved.leaf_id, original.leaf_id);
+        let archive = manager.load_journal_archive("bounded-a").unwrap();
+        // Nothing lost: document + archive is exactly the original journal.
+        let mut union = all_ids(&journal.entries);
+        union.extend(all_ids(&archive));
+        assert_eq!(union, all_ids(&original.entries));
+        assert_eq!(archive.len() + journal.len(), original.len());
+
+        // The live journal drops exactly what the save archived.
+        let mut live = original.clone();
+        let archived = take_archived_journal_ids("bounded-a");
+        assert_eq!(archived, all_ids(&archive));
+        live.remove_entries(&archived).unwrap();
+        assert_eq!(all_ids(&live.entries), all_ids(&journal.entries));
+        assert!(take_archived_journal_ids("bounded-a").is_empty());
+
+        // A plain save never prunes.
+        let small = compacted_session("bounded-plain", tmp.path(), 460);
+        let len = small.journal.as_ref().unwrap().len();
+        manager.save_session(&small).unwrap();
+        let reloaded = manager.load_session("bounded-plain").unwrap();
+        assert_eq!(reloaded.journal.unwrap().len(), len);
+    }
+
+    #[test]
+    fn archive_failure_prunes_nothing() {
+        let tmp = tempfile::tempdir().unwrap();
+        let sessions = tmp.path().join("sessions");
+        let manager = SessionManager::new(sessions.clone()).unwrap();
+        // A regular file where the archive directory belongs.
+        fs::write(sessions.join(JOURNAL_ARCHIVE_DIR), b"not a dir").unwrap();
+        let session = compacted_session("bounded-b", tmp.path(), 460);
+        let len = session.journal.as_ref().unwrap().len();
+        manager.save_session_bounded(session).unwrap();
+        let saved = manager.load_session("bounded-b").unwrap();
+        assert_eq!(saved.journal.unwrap().len(), len);
+        assert!(take_archived_journal_ids("bounded-b").is_empty());
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn archive_written_then_document_write_fails_loses_nothing() {
+        use std::os::unix::fs::PermissionsExt as _;
+        let tmp = tempfile::tempdir().unwrap();
+        let sessions = tmp.path().join("sessions");
+        let manager = SessionManager::new(sessions.clone()).unwrap();
+        let session = compacted_session("bounded-c", tmp.path(), 460);
+        let original = session.journal.clone().unwrap();
+        manager.save_session(&session).unwrap();
+        fs::create_dir_all(sessions.join(JOURNAL_ARCHIVE_DIR)).unwrap();
+        // The archive can be appended, but the document's temp file cannot
+        // be created: the crash window between the two writes.
+        fs::set_permissions(&sessions, fs::Permissions::from_mode(0o500)).unwrap();
+        let failed = manager.save_session_bounded(session.clone());
+        fs::set_permissions(&sessions, fs::Permissions::from_mode(0o700)).unwrap();
+        assert!(failed.is_err(), "document write must fail in this setup");
+        let on_disk = manager.load_session("bounded-c").unwrap();
+        assert_eq!(on_disk.journal.as_ref().unwrap(), &original);
+        assert!(
+            !manager
+                .load_journal_archive("bounded-c")
+                .unwrap()
+                .is_empty()
+        );
+        assert!(take_archived_journal_ids("bounded-c").is_empty());
+
+        // Retrying archives the same entries again; the reader dedupes and
+        // the union is still the whole journal.
+        manager.save_session_bounded(session).unwrap();
+        let saved = manager.load_session("bounded-c").unwrap();
+        let archive = manager.load_journal_archive("bounded-c").unwrap();
+        let mut union = all_ids(&saved.journal.unwrap().entries);
+        union.extend(all_ids(&archive));
+        assert_eq!(union, all_ids(&original.entries));
+        let _ = take_archived_journal_ids("bounded-c");
+    }
+
+    #[test]
+    fn branch_to_an_archived_entry_restores_its_chain() {
+        let tmp = tempfile::tempdir().unwrap();
+        let manager = SessionManager::new(tmp.path().join("sessions")).unwrap();
+        let session = compacted_session("bounded-d", tmp.path(), 460);
+        let original = session.journal.clone().unwrap();
+        manager.save_session_bounded(session).unwrap();
+        let archived = take_archived_journal_ids("bounded-d");
+        // The deepest archived entry of the oldest round.
+        let target = original
+            .entries
+            .iter()
+            .filter(|e| archived.contains(&e.id))
+            .find(|e| {
+                e.kind
+                    .as_message()
+                    .is_some_and(|m| m.content == text("round 0 message 9").content)
+            })
+            .unwrap()
+            .id
+            .clone();
+
+        let mut session = manager.load_session("bounded-d").unwrap();
+        assert!(session.journal_branch_to(&target).is_err());
+        assert!(
+            manager
+                .restore_archived_journal_chain(&mut session, &target)
+                .unwrap()
+        );
+        session.journal_branch_to(&target).unwrap();
+        manager.save_session(&session).unwrap();
+        let reloaded = manager.load_session("bounded-d").unwrap();
+        assert_eq!(reloaded.leaf_id.as_deref(), Some(target.as_str()));
+        // Each round replaced the whole tail, so round 0 is its own root chain.
+        let expected: Vec<Message> = (0..10)
+            .map(|i| text(&format!("round 0 message {i}")))
+            .collect();
+        assert_eq!(reloaded.messages, expected);
+
+        let mut missing = manager.load_session("bounded-d").unwrap();
+        assert!(
+            !manager
+                .restore_archived_journal_chain(&mut missing, "no-such-entry")
+                .unwrap()
+        );
+    }
+
+    #[test]
+    fn deleting_a_session_removes_its_journal_archive() {
+        let tmp = tempfile::tempdir().unwrap();
+        let sessions = tmp.path().join("sessions");
+        let manager = SessionManager::new(sessions.clone()).unwrap();
+        manager
+            .save_session_bounded(compacted_session("bounded-e", tmp.path(), 460))
+            .unwrap();
+        let _ = take_archived_journal_ids("bounded-e");
+        let path = journal_archive_path(&sessions, "bounded-e").unwrap();
+        assert!(path.exists());
+        manager.delete_session("bounded-e").unwrap();
+        assert!(!path.exists());
+    }
+
+    #[test]
+    fn metadata_larger_than_the_prefix_is_read_without_the_transcript() {
+        let tmp = tempfile::tempdir().unwrap();
+        let manager = SessionManager::new(tmp.path().join("sessions")).unwrap();
+        let mut session = compacted_session("bounded-f", tmp.path(), 10);
+        for i in 0..9_000 {
+            session.metadata.cost.usage_source_fingerprints.insert(
+                crate::cost_status::usage_source_fingerprint(&format!("call-{i}")),
+            );
+        }
+        let path = manager.save_session(&session).unwrap();
+        // Replace the transcript with 1 MB of non-JSON: listing must still
+        // find the >64 KB metadata block (the grown read stops once it
+        // closes; this checks correctness, not how many bytes were read).
+        let bytes = fs::read(&path).unwrap();
+        let start = find_json_key(&bytes, b"\"metadata\"").unwrap();
+        let open = json_value_start(&bytes, start, 10, b'{').unwrap();
+        let end = json_object_end(&bytes, open).unwrap();
+        assert!(
+            end > 64 * 1024,
+            "metadata must exceed the old prefix: {end}"
+        );
+        let mut damaged = bytes[..end].to_vec();
+        damaged.extend(std::iter::repeat_n(b'x', 1 << 20));
+        fs::write(&path, &damaged).unwrap();
+        let metadata = SessionManager::load_session_metadata(&path).unwrap();
+        assert_eq!(metadata.id, "bounded-f");
+        assert_eq!(metadata.cost.usage_source_fingerprints.len(), 9_000);
+    }
+}

crates/tui/src/tui/persistence_actor.rs

@@ -592,13 +592,15 @@ fn flush_inner(
         Err(err) => report.failures.push((what, err.kind())),
     };
 
+    // Document saves are bounded (#6842): an oversized journal is archived
+    // and pruned here, on the actor, never on the UI loop.
     for (session_id, session) in std::mem::take(&mut pending.sessions) {
-        let result = manager.save_session_owned(session).map(|_| ());
+        let result = manager.save_session_bounded(session).map(|_| ());
         health.record(&session_id, &result);
         record(format!("session:{session_id}"), result);
     }
     for (session_id, session) in std::mem::take(&mut pending.completed_commits) {
-        let commit_result = manager.save_session_owned(session);
+        let commit_result = manager.save_session_bounded(session);
         health.record(&session_id, &commit_result);
         let save_succeeded = commit_result.is_ok();
         record(

crates/tui/src/tui/ui/frame.rs

@@ -1150,6 +1150,14 @@ pub(crate) fn build_session_snapshot(
             format!("automatic session snapshot skipped while Work state is busy: {err}")
         })?,
     };
+    // Drop what a bounded save already archived (#6842) so the live journal
+    // matches the document. No I/O: the ids come from the save path.
+    if let Some(session_id) = app.current_session_id.as_deref() {
+        let archived = crate::session_manager::take_archived_journal_ids(session_id);
+        if let Err(error) = app.session_journal.remove_entries(&archived) {
+            tracing::warn!(%error, "kept archived journal entries in memory");
+        }
+    }
     app.session_journal
         .rebranch_active_messages_stamped(&app.api_messages, &app.api_message_stamps);
     let mut session = crate::session_manager::create_saved_session_journal_only(

crates/tui/src/work_graph/events.rs

@@ -81,6 +81,11 @@ pub enum WorkGraphChange {
     RecordActivity {
         event: WorkActivityEvent,
     },
+    /// Evict the oldest ended, non-durable Operation nodes beyond `keep`
+    /// (#6842). See `reducer::prune_ended_operations` for eligibility.
+    PruneEndedOperations {
+        keep: usize,
+    },
 }
 
 impl WorkGraphChange {
@@ -103,6 +108,7 @@ impl WorkGraphChange {
             WorkGraphChange::ReplaceCompatProjection { .. } => "replace_compat_projection",
             WorkGraphChange::SetImportDigest { .. } => "set_import_digest",
             WorkGraphChange::RecordActivity { .. } => "record_activity",
+            WorkGraphChange::PruneEndedOperations { .. } => "prune_ended_operations",
         }
     }
 }

crates/tui/src/work_graph/model.rs

@@ -39,6 +39,11 @@ pub const ACTIVITY_CAP: usize = 256;
 /// Bounded idempotency-key dedup window kept on the snapshot.
 pub const SEEN_KEYS_CAP: usize = 1024;
 
+/// Ended, non-durable Operation nodes (one per finished shell call) kept on
+/// the snapshot (#6842). They are a derived index — the calls and their output
+/// live in the session transcript — so older ones are evicted, not archived.
+pub const ENDED_OPERATION_CAP: usize = 256;
+
 /// Canonical reasoning-effort tiers recorded as configuration facts. This is
 /// deliberately an enum rather than free-form text so Work Graph activity can
 /// never become a side channel for model reasoning.

crates/tui/src/work_graph/reducer.rs

@@ -207,10 +207,92 @@ fn apply_pure(
             }
             next.activities.push_bounded(event.clone());
         }
+        WorkGraphChange::PruneEndedOperations { keep } => {
+            prune_ended_operations(&mut next, *keep);
+        }
     }
     Ok(next)
 }
 
+/// Operation nodes that may be evicted: ended (not live, not `Stale` — a
+/// stale shell can still be recovered by a late owner report), bound to a
+/// non-durable owner, touched by no edge except an incoming `Contains`, and
+/// referenced by no activity or pending proposal.
+pub(crate) fn evictable_operations(g: &WorkGraphSnapshot) -> Vec<&WorkNode> {
+    use std::collections::HashSet;
+    let mut pinned: HashSet<&WorkNodeId> = HashSet::new();
+    for edge in &g.edges {
+        if edge.kind != EdgeKind::Contains {
+            pinned.insert(&edge.from);
+            pinned.insert(&edge.to);
+        } else {
+            pinned.insert(&edge.from);
+        }
+    }
+    for activity in g.activities.iter() {
+        let WorkActivityEvent::ReasoningEffortChanged { operation, .. } = activity;
+        if let Some(operation) = operation {
+            pinned.insert(operation);
+        }
+    }
+    for proposal in &g.proposals {
+        pinned.extend(proposal.removed_nodes.iter());
+        pinned.extend(proposal.updated_nodes.iter().map(|update| &update.id));
+        for edge in &proposal.added_edges {
+            pinned.insert(&edge.from);
+            pinned.insert(&edge.to);
+        }
+    }
+    g.nodes
+        .iter()
+        .filter(|node| {
+            node.kind == NodeKind::Operation
+                && matches!(
+                    node.state,
+                    NodeState::Completed
+                        | NodeState::Failed
+                        | NodeState::Cancelled
+                        | NodeState::Superseded
+                        | NodeState::Verified
+                )
+                && node
+                    .binding
+                    .as_ref()
+                    .is_some_and(|binding| !binding.durable)
+                && !pinned.contains(&node.id)
+        })
+        .collect()
+}
+
+/// Keep the newest `keep` evictable operations (by `updated_at`, then
+/// insertion order, so the result is deterministic) and remove the rest with
+/// their `Contains` edges.
+fn prune_ended_operations(next: &mut WorkGraphSnapshot, keep: usize) {
+    let positions: std::collections::HashMap<&WorkNodeId, usize> = next
+        .nodes
+        .iter()
+        .enumerate()
+        .map(|(position, node)| (&node.id, position))
+        .collect();
+    let mut evictable: Vec<(i64, usize, WorkNodeId)> = evictable_operations(next)
+        .into_iter()
+        .map(|node| (node.updated_at, positions[&node.id], node.id.clone()))
+        .collect();
+    drop(positions);
+    let Some(excess) = evictable.len().checked_sub(keep).filter(|n| *n > 0) else {
+        return;
+    };
+    evictable.sort();
+    let evicted: std::collections::HashSet<WorkNodeId> = evictable
+        .into_iter()
+        .take(excess)
+        .map(|(_, _, id)| id)
+        .collect();
+    next.nodes.retain(|node| !evicted.contains(&node.id));
+    next.edges
+        .retain(|edge| !evicted.contains(&edge.from) && !evicted.contains(&edge.to));
+}
+
 fn add_node(next: &mut WorkGraphSnapshot, node: WorkNode) -> Result<(), ValidationReport> {
     if next.node(&node.id).is_some() {
         return Err(structural(format!("duplicate node {}", node.id)));

crates/tui/src/work_graph/runtime.rs

@@ -196,6 +196,20 @@ impl WorkRuntime {
             title,
             NodeState::Initializing,
         )?;
+        // Registration is where finished shell calls accumulate (#6842):
+        // keep only the newest ended, non-durable operations.
+        if super::reducer::evictable_operations(graph.snapshot()).len()
+            > super::model::ENDED_OPERATION_CAP
+        {
+            apply_change(
+                &mut graph,
+                session_id,
+                &intent.source,
+                WorkGraphChange::PruneEndedOperations {
+                    keep: super::model::ENDED_OPERATION_CAP,
+                },
+            )?;
+        }
         let next = graph.into_snapshot();
         validate_combined(&next, &project_plan(&next), &project_todos(&next))?;
         active.snapshot = Some(next);
@@ -1931,4 +1945,62 @@ mod tests {
         assert!(!runtime.has_pending_publish());
         assert!(todos.blocking_lock().snapshot().is_empty());
     }
+
+    /// #6842: one Operation node per finished shell call grew without bound.
+    #[test]
+    fn ended_shell_operations_are_capped_oldest_first() {
+        let runtime = new_shared_work_runtime(
+            crate::tools::todo::new_shared_todo_list(),
+            crate::tools::plan::new_shared_plan_state(),
+        );
+        let total = super::super::model::ENDED_OPERATION_CAP + 44;
+        let mut ids = Vec::new();
+        for i in 0..total {
+            let external = format!("shell:cap_{i}");
+            let intent =
+                OperationIntent::new(&external, "ls", false, "exec_shell", &format!("c{i}"));
+            ids.push(
+                runtime
+                    .register_operation("session", intent)
+                    .expect("register"),
+            );
+            for (seq, state) in [(1, OwnerState::Running), (2, OwnerState::Completed)] {
+                runtime
+                    .reconcile_operation(
+                        "session",
+                        OperationOwnerSnapshot::new(
+                            &external,
+                            state,
+                            seq,
+                            i as i64 * 10 + seq as i64,
+                        ),
+                    )
+                    .expect("reconcile");
+            }
+        }
+        let graph = runtime
+            .capture(Some("session"))
+            .expect("capture")
+            .expect("graph")
+            .graph;
+        crate::work_graph::validate(&graph).expect("pruned graph validates");
+        let operations = graph
+            .nodes
+            .iter()
+            .filter(|node| node.kind == NodeKind::Operation)
+            .count();
+        // Pruned to the cap at each registration; the last call then ended.
+        assert!(
+            operations <= super::super::model::ENDED_OPERATION_CAP + 1,
+            "{operations}"
+        );
+        assert!(graph.node(&ids[0]).is_none(), "oldest ended call evicted");
+        assert!(graph.node(&ids[total - 1]).is_some(), "newest call kept");
+        assert!(
+            graph
+                .edges
+                .iter()
+                .all(|edge| graph.node(&edge.from).is_some() && graph.node(&edge.to).is_some())
+        );
+    }
 }