Merge durability lane 9413a891 (atomic external session mutations, seed intent, goal CAS)

External rename/archive/delete/stub-retire hold the existing SessionLease for the
whole mutation and refuse local live/borrowed or uncertain lock/symlink states;
seed history is journaled and committed last with every record validated before
cleanup; goal status transitions compare-and-swap the read revision; bounded
REPL rounds poison on error; model catalog/fleet/plugin validation tightened.
No new store, lock or lease registry.

Lane receipts at exact 9413a891: 154 passed / 0 failed / 1 ignored across the
session/seed/goal consumers; released-lease production control 1 passed / 1
failed; byte-exact restore then 2 passed / 0 failed; CI-policy TUI Clippy pass.
Independent review (no blockers) left should-fix items that are tracked for the
next batch: a leftover post-commit seed journal can refuse the whole store,
HeldRuntimeStore readers ignore seed journals, lease reservation can sleep on
async handlers, and three external writers still probe-then-write.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/runtime/src/goal_loop.rs

@@ -268,9 +268,10 @@ pub const fn continuation_wait(delay_seconds: u64) -> Option<Duration> {
 }
 
 /// Wait out the between-continuation quiet period, honoring cancellation.
-/// `None` resolves to `Elapsed` immediately so callers have a single dispatch
-/// gate. Cancellation is biased and always wins over a racing expiry — the
-/// same semantics as the interactive cadence (#5508). Two dispatchers await
+/// `None` resolves immediately so callers have a single dispatch gate: to
+/// `Cancelled` when the token is already cancelled, otherwise `Elapsed`.
+/// Cancellation is biased and always wins over a racing expiry — the same
+/// semantics as the interactive cadence (#5508), including a zero delay. Two dispatchers await
 /// this gate: the turn loop's intra-turn passes (every session), and the
 /// runtime host's cross-turn re-arm for host-managed engines
 /// (`RuntimeThreadManager::spawn_goal_continuation`), which never
@@ -279,6 +280,9 @@ pub async fn await_continuation_wait(
     wait: Option<Duration>,
     cancel_token: &tokio_util::sync::CancellationToken,
 ) -> ContinuationWaitOutcome {
+    if cancel_token.is_cancelled() {
+        return ContinuationWaitOutcome::Cancelled;
+    }
     let Some(wait) = wait else {
         return ContinuationWaitOutcome::Elapsed;
     };
@@ -289,14 +293,19 @@ pub async fn await_continuation_wait(
     }
 }
 
-/// Whether the durable token usage has reached the active goal's budget.
-///
-/// Budgets are telemetry-only in unbounded goal mode. Keeping this shared
-/// predicate false ensures preview and the live continuation loop agree that
-/// crossing a token budget does not close the outbound gate.
+/// Whether the durable token usage has reached an *enforced* token budget —
+/// the same condition on which [`decide_continuation`] stops with
+/// [`StopReason::BudgetLimit`]. Budgets are telemetry-only unless
+/// `[goal] enforce_token_budget` opts in, so without it this stays false and
+/// crossing a budget never closes the outbound gate. Preview and the live
+/// continuation loop share this predicate so they cannot disagree about spend.
 #[must_use]
-pub const fn token_budget_exhausted(_progress: GoalProgress, _budget: GoalBudget) -> bool {
-    false
+pub const fn token_budget_exhausted(progress: GoalProgress, budget: GoalBudget) -> bool {
+    budget.enforce_token_budget
+        && match budget.token_budget {
+            Some(limit) => progress.tokens_used >= limit,
+            None => false,
+        }
 }
 
 /// Whether a stop reason represents success (Completed) vs. an early/forced exit.
@@ -573,6 +582,39 @@ mod tests {
         );
     }
 
+    #[test]
+    fn exhausted_budget_predicate_matches_the_live_stop() {
+        let progress = GoalProgress {
+            tokens_used: 1_000,
+            ..GoalProgress::default()
+        };
+        for enforce in [false, true] {
+            let budget = GoalBudget {
+                token_budget: Some(1_000),
+                time_budget_seconds: None,
+                enforce_token_budget: enforce,
+                max_continuations: 0,
+            };
+            assert_eq!(
+                token_budget_exhausted(progress, budget),
+                decide_continuation(GoalRunStatus::Active, progress, budget)
+                    == ContinuationDecision::Stop(StopReason::BudgetLimit),
+                "preview must report exactly the budget stop the live loop takes (enforce={enforce})"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn cancellation_wins_over_a_zero_delay() {
+        let cancel_token = tokio_util::sync::CancellationToken::new();
+        cancel_token.cancel();
+        assert_eq!(
+            await_continuation_wait(continuation_wait(0), &cancel_token).await,
+            ContinuationWaitOutcome::Cancelled,
+            "a cancelled goal with no quiet period must never dispatch"
+        );
+    }
+
     #[tokio::test]
     async fn elapsed_quiet_period_dispatches() {
         let cancel_token = tokio_util::sync::CancellationToken::new();

crates/tui/src/automation_manager.rs

@@ -2222,6 +2222,22 @@ where
     Ok(run)
 }
 
+/// Run one of the scheduler's whole-store scans (every automation or run
+/// record read and parsed) on the blocking pool (#6149), holding the manager
+/// lock exactly as the inline call did. Known limit: the per-record claim,
+/// recovery, and receipt writes between scans still run inline; each touches
+/// one small record under the store lock.
+async fn with_manager_blocking<T, F>(automations: &SharedAutomationManager, scan: F) -> Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce(&AutomationManager) -> Result<T> + Send + 'static,
+{
+    let manager = Arc::clone(automations).lock_owned().await;
+    tokio::task::spawn_blocking(move || scan(&manager))
+        .await
+        .context("automation store scan task failed")?
+}
+
 async fn scheduler_tick_shared(
     automations: &SharedAutomationManager,
     task_manager: &SharedTaskManager,
@@ -2255,7 +2271,8 @@ where
     };
     // Repair admitted work before collecting a new occurrence, including claims
     // whose definitions were edited/deleted or whose final enqueue save tore.
-    let pending = automations.lock().await.collect_pending_runs()?;
+    let pending =
+        with_manager_blocking(automations, AutomationManager::collect_pending_runs).await?;
     for run in pending.into_iter().filter(|run| {
         run.dispatch
             .as_ref()
@@ -2290,7 +2307,8 @@ where
         }
     }
     let now = Utc::now();
-    let due = automations.lock().await.collect_due_runs(now)?;
+    let due =
+        with_manager_blocking(automations, move |manager| manager.collect_due_runs(now)).await?;
     for (observed, proposed) in due {
         if !automations
             .lock()
@@ -2386,7 +2404,10 @@ where
         Err(error) => return Err(error).context("claim delayed-trigger dispatch"),
     };
     let now = Utc::now();
-    let candidates = automations.lock().await.collect_due_triggers(now)?;
+    let candidates = with_manager_blocking(automations, move |manager| {
+        manager.collect_due_triggers(now)
+    })
+    .await?;
     for candidate in candidates {
         let scope = if candidate.status == DelayedTriggerStatus::Dispatching {
             candidate
@@ -2576,7 +2597,8 @@ async fn reconcile_run_statuses_shared(
         Err(error) if dispatch_lock_busy(&error) => return Ok(()),
         Err(error) => return Err(error).context("claim automation reconciliation"),
     };
-    let pending = automations.lock().await.collect_pending_runs()?;
+    let pending =
+        with_manager_blocking(automations, AutomationManager::collect_pending_runs).await?;
     for mut run in pending {
         // Shared storage is not shared ownership. A receipt admitted under
         // another execution scope is reconciled by that scope's owner; this

crates/tui/src/fleet/host.rs

@@ -220,6 +220,14 @@ impl LocalProcessFleetHostAdapter {
                     request.worker_id
                 )));
             }
+            // A draining worker's dispatcher exited but its tree still runs;
+            // forgetting it here would overlap a replacement with it.
+            if matches!(status.state, FleetHostWorkerState::Draining) {
+                return Err(FleetHostError::retryable(format!(
+                    "worker {} is still draining; stop it before starting a replacement",
+                    request.worker_id
+                )));
+            }
             self.processes.remove(&request.worker_id);
         }
 
@@ -440,9 +448,10 @@ impl FleetHostAdapter for LocalProcessFleetHostAdapter {
             .get(worker_id)
             .map(|process| process.request.clone())
             .ok_or_else(|| FleetHostError::terminal(format!("unknown worker {worker_id}")))?;
-        let _ = self.stop_worker(worker_id);
-        self.processes.remove(worker_id);
-        self.start_worker(request)
+        restart_after_confirmed_stop(self, worker_id, |adapter| {
+            adapter.processes.remove(worker_id);
+            adapter.start_worker(request)
+        })
     }
 
     fn stop_worker(&mut self, worker_id: &str) -> FleetHostResult<FleetHostWorkerStatus> {
@@ -739,9 +748,10 @@ impl FleetHostAdapter for SshFleetHostAdapter {
             .get(worker_id)
             .map(|process| process.request.clone())
             .ok_or_else(|| FleetHostError::terminal(format!("unknown worker {worker_id}")))?;
-        let _ = self.stop_worker(worker_id);
-        self.local.processes.remove(worker_id);
-        self.local.start_with_kind(request, FleetHostKind::Ssh)
+        restart_after_confirmed_stop(self, worker_id, |adapter| {
+            adapter.local.processes.remove(worker_id);
+            adapter.local.start_with_kind(request, FleetHostKind::Ssh)
+        })
     }
 
     fn stop_worker(&mut self, worker_id: &str) -> FleetHostResult<FleetHostWorkerStatus> {
@@ -753,6 +763,38 @@ impl FleetHostAdapter for SshFleetHostAdapter {
     }
 }
 
+/// Restart policy shared by the process-backed adapters. The previous worker
+/// is released — and `replace` spawns its successor — only once a stop has
+/// confirmed the whole worker tree is gone. A failed or unconfirmed stop keeps
+/// the old handle so status, logs, and cleanup still reach a worker that may
+/// be running, and no replacement starts beside it (duplicate execution). The
+/// process-tree lifecycle in `stop_worker` stays the authority on "gone".
+fn restart_after_confirmed_stop<A: FleetHostAdapter>(
+    adapter: &mut A,
+    worker_id: &str,
+    replace: impl FnOnce(&mut A) -> FleetHostResult<FleetWorkerHandle>,
+) -> FleetHostResult<FleetWorkerHandle> {
+    let stopped = adapter.stop_worker(worker_id).map_err(|err| FleetHostError {
+        kind: err.kind,
+        message: format!(
+            "restart of worker {worker_id} refused: the previous worker was not confirmed stopped ({})",
+            err.message
+        ),
+    })?;
+    if matches!(
+        stopped.state,
+        FleetHostWorkerState::Running
+            | FleetHostWorkerState::Draining
+            | FleetHostWorkerState::Unknown
+    ) {
+        return Err(FleetHostError::retryable(format!(
+            "restart of worker {worker_id} refused: the previous worker is still {:?} after stop",
+            stopped.state
+        )));
+    }
+    replace(adapter)
+}
+
 fn open_worker_log(path: &Path) -> FleetHostResult<File> {
     if let Some(parent) = path.parent() {
         std::fs::create_dir_all(parent).map_err(|err| {
@@ -2023,6 +2065,95 @@ mod tests {
         adapter.stop_worker("local-restart").unwrap();
     }
 
+    /// A host whose stop outcome is scripted: `None` fails the stop, `Some`
+    /// reports that state. Only the restart path is exercised.
+    struct ScriptedStopHost {
+        stop_state: Option<FleetHostWorkerState>,
+        owned: bool,
+        spawned: usize,
+    }
+
+    impl FleetHostAdapter for ScriptedStopHost {
+        fn start_worker(
+            &mut self,
+            request: FleetWorkerStartRequest,
+        ) -> FleetHostResult<FleetWorkerHandle> {
+            self.spawned += 1;
+            self.owned = true;
+            Ok(FleetWorkerHandle {
+                worker_id: request.worker_id,
+                host_kind: FleetHostKind::LocalProcess,
+                pid: None,
+                log_path: PathBuf::new(),
+            })
+        }
+        fn read_status(&mut self, _: &str) -> FleetHostResult<FleetHostWorkerStatus> {
+            unreachable!("restart must not consult status outside stop")
+        }
+        fn read_logs(&self, _: &str, _: usize) -> FleetHostResult<String> {
+            unreachable!()
+        }
+        fn interrupt_worker(&mut self, _: &str) -> FleetHostResult<FleetHostWorkerStatus> {
+            unreachable!()
+        }
+        fn restart_worker(&mut self, worker_id: &str) -> FleetHostResult<FleetWorkerHandle> {
+            restart_after_confirmed_stop(self, worker_id, |host| {
+                host.owned = false;
+                host.start_worker(FleetWorkerStartRequest::new(
+                    worker_id,
+                    shell_command("true"),
+                ))
+            })
+        }
+        fn stop_worker(&mut self, worker_id: &str) -> FleetHostResult<FleetHostWorkerStatus> {
+            let Some(state) = self.stop_state else {
+                return Err(FleetHostError::retryable(
+                    "Fleet session still has a live tracked leader after SIGKILL",
+                ));
+            };
+            Ok(FleetHostWorkerStatus {
+                worker_id: worker_id.to_string(),
+                state,
+                pid: Some(4242),
+                exit_code: None,
+                memory_mb: None,
+                retryable: false,
+            })
+        }
+        fn cleanup_worker(&mut self, _: &str) -> FleetHostResult<()> {
+            unreachable!()
+        }
+    }
+
+    #[test]
+    fn fleet_host_restart_keeps_the_old_worker_when_stop_is_unconfirmed() {
+        for stop_state in [None, Some(FleetHostWorkerState::Draining)] {
+            let mut host = ScriptedStopHost {
+                stop_state,
+                owned: true,
+                spawned: 0,
+            };
+            let err = host
+                .restart_worker("overlap")
+                .expect_err("an unconfirmed stop must refuse the restart");
+            assert!(err.message.contains("refused"), "{}", err.message);
+            assert_eq!(
+                host.spawned, 0,
+                "no replacement may start beside a surviving worker ({stop_state:?})"
+            );
+            assert!(host.owned, "the old worker's handle is kept");
+        }
+
+        let mut host = ScriptedStopHost {
+            stop_state: Some(FleetHostWorkerState::Stopped),
+            owned: true,
+            spawned: 0,
+        };
+        host.restart_worker("overlap")
+            .expect("a confirmed stop restarts");
+        assert_eq!(host.spawned, 1);
+    }
+
     #[cfg(unix)]
     #[test]
     fn fleet_host_local_adapter_reports_running_worker_memory_usage() {

crates/tui/src/fleet/store.rs

@@ -38,6 +38,36 @@ use super::roster::FleetRoster;
 pub const FLEET_SCHEMA_KIND: &str = "fleet";
 pub const FLEET_SCHEMA_REVISION: u32 = 2;
 const MAX_MEMBER_DISPLAY_NAME_CHARS: usize = 80;
+const MAX_FLEET_NAME_CHARS: usize = 120;
+const MAX_MEMBER_ID_CHARS: usize = 64;
+const MAX_MEMBER_ROLE_CHARS: usize = 80;
+const MAX_ROUTE_FIELD_CHARS: usize = 256;
+const MAX_MEMBER_INSTRUCTIONS_CHARS: usize = 32 * 1024;
+
+/// A value that reaches selectors, receipts, and terminal rendering: printable
+/// and bounded. `multiline` admits newlines and tabs (instruction overlays)
+/// but no other control characters.
+fn validate_text(
+    what: &str,
+    value: &str,
+    max_chars: usize,
+    multiline: bool,
+) -> Result<(), FleetStoreError> {
+    let printable = value
+        .chars()
+        .all(|ch| !ch.is_control() || (multiline && matches!(ch, '\n' | '\r' | '\t')));
+    if !printable || value.chars().count() > max_chars {
+        return Err(FleetStoreError::Invalid(format!(
+            "{what} must be printable {} no longer than {max_chars} characters",
+            if multiline {
+                "text"
+            } else {
+                "single-line text"
+            }
+        )));
+    }
+    Ok(())
+}
 
 /// The directory name used by both roots (next to `agents/` for legacy
 /// profiles). Also used by the workflow crate for its own legacy/exact files;
@@ -286,6 +316,7 @@ impl FleetFile {
                 "fleet name must not be empty".to_string(),
             ));
         }
+        validate_text("fleet name", name, MAX_FLEET_NAME_CHARS, false)?;
         let mut seen: BTreeMap<String, String> = BTreeMap::new();
         for member in &self.members {
             let member_id = member.id.trim();
@@ -294,6 +325,56 @@ impl FleetFile {
                     "member id must not be empty".to_string(),
                 ));
             }
+            // The id is dispatch identity: exact, not merely equal once trimmed.
+            if member_id != member.id {
+                return Err(FleetStoreError::Invalid(format!(
+                    "member id {:?} must not carry leading or trailing whitespace",
+                    member.id
+                )));
+            }
+            validate_text("member id", member_id, MAX_MEMBER_ID_CHARS, false)?;
+            let fields = [
+                (
+                    "role",
+                    Some(member.role.as_str()),
+                    MAX_MEMBER_ROLE_CHARS,
+                    false,
+                ),
+                (
+                    "provider",
+                    member.provider.as_deref(),
+                    MAX_ROUTE_FIELD_CHARS,
+                    false,
+                ),
+                (
+                    "model",
+                    member.model.as_deref(),
+                    MAX_ROUTE_FIELD_CHARS,
+                    false,
+                ),
+                (
+                    "reasoning",
+                    member.reasoning.as_deref(),
+                    MAX_ROUTE_FIELD_CHARS,
+                    false,
+                ),
+                (
+                    "instructions",
+                    member.instructions.as_deref(),
+                    MAX_MEMBER_INSTRUCTIONS_CHARS,
+                    true,
+                ),
+            ];
+            for (field, value, max_chars, multiline) in fields {
+                if let Some(value) = value {
+                    validate_text(
+                        &format!("member `{member_id}` {field}"),
+                        value,
+                        max_chars,
+                        multiline,
+                    )?;
+                }
+            }
             let member_key = member_id.to_ascii_lowercase();
             if let Some(existing) = seen.insert(member_key, member.id.clone()) {
                 return Err(FleetStoreError::Invalid(format!(
@@ -1249,6 +1330,33 @@ mod tests {
             "{err}"
         );
 
+        // Identity and route values are bounded and control-free too; only
+        // the instruction overlay may span lines.
+        let mut fleet = sample_fleet();
+        fleet.members[0].id = "scout\u{1b}[2J".to_string();
+        let err = fleet.validate().unwrap_err();
+        assert!(
+            err.to_string().contains("member id must be printable"),
+            "{err}"
+        );
+        let mut fleet = sample_fleet();
+        fleet.members[0].id = "x".repeat(MAX_MEMBER_ID_CHARS + 1);
+        assert!(fleet.validate().is_err());
+        let mut fleet = sample_fleet();
+        fleet.members[0].id = format!("{} ", fleet.members[0].id);
+        assert!(
+            fleet.validate().is_err(),
+            "an untrimmed id is not its trimmed twin"
+        );
+        let mut fleet = sample_fleet();
+        fleet.name = "Team\u{7}".to_string();
+        assert!(fleet.validate().is_err());
+        let mut fleet = sample_fleet();
+        fleet.members[0].instructions = Some("line one\nline two\ttabbed".to_string());
+        fleet.validate().expect("instructions may span lines");
+        fleet.members[0].instructions = Some("x".repeat(MAX_MEMBER_INSTRUCTIONS_CHARS + 1));
+        assert!(fleet.validate().is_err());
+
         // Lone provider / lone model: never silently reinterpreted.
         let mut fleet = sample_fleet();
         fleet.members[0].provider = Some("deepseek".to_string());

crates/tui/src/operate.rs

@@ -1331,16 +1331,32 @@ pub(crate) fn upsert_keepalive(
 /// Cancel tears the operation down *including* its keepalive: an unattended
 /// hourly lead run after cancel is pure cost. The automation is paused
 /// (never deleted) so its run history survives and a later start reactivates
-/// it. A missing keepalive is not an error.
+/// it. A missing keepalive is not an error; any other failure to establish
+/// its state is. A requested spending stop that cannot confirm the record is
+/// paused must never look like one that happened — an unreadable record may
+/// still be Active and scheduled.
 pub fn pause_keepalive(manager: &AutomationManager) -> Result<()> {
-    match manager.get_automation(OPERATE_KEEPALIVE_ID) {
-        Ok(record) if matches!(record.status, AutomationStatus::Active) => {
-            manager.pause_automation(OPERATE_KEEPALIVE_ID)?;
-            Ok(())
+    let record = match manager.get_automation(OPERATE_KEEPALIVE_ID) {
+        Ok(record) => record,
+        Err(error) if is_not_found(&error) => return Ok(()),
+        Err(error) => {
+            return Err(error.context(
+                "Operate keepalive state could not be established; it may still be scheduled",
+            ));
         }
-        Ok(_) => Ok(()),
-        Err(_) => Ok(()),
+    };
+    if matches!(record.status, AutomationStatus::Active) {
+        manager.pause_automation(OPERATE_KEEPALIVE_ID)?;
     }
+    Ok(())
+}
+
+fn is_not_found(error: &anyhow::Error) -> bool {
+    error.chain().any(|cause| {
+        cause
+            .downcast_ref::<std::io::Error>()
+            .is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound)
+    })
 }
 
 /// Pull the next keepalive lead run to the next scheduler tick (for example
@@ -1984,6 +2000,28 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
         assert!(!kick_keepalive(&manager).expect("kick paused"));
     }
 
+    #[test]
+    fn pause_keepalive_refuses_to_report_a_stop_it_cannot_confirm() {
+        let dir = TempDir::new().expect("temp");
+        let manager = AutomationManager::open_for_test(dir.path().to_path_buf()).expect("manager");
+        upsert_keepalive(&manager, dir.path(), false, &route_fixture_config(), None)
+            .expect("upsert");
+        // Fault fixture: the keepalive record exists but cannot be parsed, so
+        // whether it is still Active (and spending) is unknown.
+        let record = dir
+            .path()
+            .join("automations")
+            .join(format!("{OPERATE_KEEPALIVE_ID}.json"));
+        assert!(record.exists(), "fixture targets the real record path");
+        fs::write(&record, b"{\"status\": \"active\", truncated").expect("corrupt record");
+
+        let error = pause_keepalive(&manager).expect_err("an unconfirmed stop must fail");
+        assert!(
+            format!("{error:#}").contains("could not be established"),
+            "{error:#}"
+        );
+    }
+
     #[test]
     fn explicit_env_paths_reject_traversal() {
         assert!(explicit_file_path("  /tmp/does-not-exist.md ").is_none());

crates/tui/src/plugins/marketplace/document.rs

@@ -8,7 +8,7 @@
 //! command and the HTTP API can never disagree about what would be fetched.
 
 use std::io::Read;
-use std::path::{Path, PathBuf};
+use std::path::{Component, Path, PathBuf};
 
 use super::parsers::{self, MarketplaceDocument};
 use super::store::StoredMarketplaceCatalog;
@@ -151,17 +151,18 @@ pub fn resolve_candidate_install<'a>(
         };
     }
     match &candidate.install_plan {
-        MarketplaceInstallPlan::Supported { spec, source_kind } => {
-            CatalogInstallResolution::Supported {
-                spec: resolve_spec(
-                    &entry.source_path,
-                    entry.catalog.format,
-                    &candidate.source,
-                    spec,
-                ),
+        MarketplaceInstallPlan::Supported { spec, source_kind } => match resolve_spec(
+            &entry.source_path,
+            entry.catalog.format,
+            &candidate.source,
+            spec,
+        ) {
+            Ok(spec) => CatalogInstallResolution::Supported {
+                spec,
                 source_kind: source_kind.clone(),
-            }
-        }
+            },
+            Err(reason) => CatalogInstallResolution::Unsupported { reason },
+        },
         MarketplaceInstallPlan::Unsupported { reason, .. } => {
             CatalogInstallResolution::Unsupported {
                 reason: reason.clone(),
@@ -170,14 +171,28 @@ pub fn resolve_candidate_install<'a>(
     }
 }
 
+/// A catalog's local source must name a directory inside the catalog's own
+/// tree. An absolute path, a root or drive prefix, or any `..` component would
+/// let catalog metadata point the installer at an arbitrary directory on this
+/// machine, so such an entry is not installable from the catalog; installing a
+/// directory outside it stays an explicit `/plugin install path:...`.
 fn resolve_spec(
     source_path: &str,
     format: MarketplaceFormat,
     source: &MarketplaceSourceSpec,
     spec: &str,
-) -> String {
+) -> Result<String, String> {
+    if let MarketplaceSourceSpec::LocalPath { path } = source
+        && !path
+            .components()
+            .all(|part| matches!(part, Component::Normal(_) | Component::CurDir))
+    {
+        return Err(format!(
+            "Local catalog source `{}` leaves the catalog directory; only paths inside it (no absolute paths or `..`) are installable from a catalog",
+            path.display()
+        ));
+    }
     if let MarketplaceSourceSpec::LocalPath { path } = source
-        && path.is_relative()
         && let Some(dir) = Path::new(source_path).parent()
     {
         // Claude keeps its catalog in a manifest-only metadata directory;
@@ -189,9 +204,9 @@ fn resolve_spec(
         } else {
             dir
         };
-        return format!("path:{}", dir.join(path).display());
+        return Ok(format!("path:{}", dir.join(path).display()));
     }
-    spec.to_string()
+    Ok(spec.to_string())
 }
 
 /// Resolve a user-supplied document path to an existing regular file without

crates/tui/src/plugins/marketplace/parsers/codewhale.rs

@@ -171,6 +171,23 @@ fn parse_codewhale_entry(
         ));
         return None;
     };
+    // The name is the candidate's identity (registry lookup, `/plugin show`,
+    // install), so it is held to the plugin-name rules the other formats
+    // normalize to, rather than stored verbatim with controls or unbounded.
+    if crate::plugins::manifest::validate_plugin_name(name).is_err()
+        && !crate::plugins::agent_plugin::is_standard_plugin_name(name)
+    {
+        diagnostics.push(MarketplaceDiagnostic::error(
+            "INVALID_NAME",
+            format!(
+                "Codewhale plugin at index {index} has an invalid name {:?}: use lowercase letters, digits, and internal hyphens or dots",
+                name.chars().take(80).collect::<String>()
+            ),
+            None,
+            Some(index),
+        ));
+        return None;
+    }
     let name = name.to_string();
 
     let (source, bad_source) = str_field(obj, "source");

crates/tui/src/plugins/marketplace/parsers/mod.rs

@@ -77,12 +77,39 @@ pub fn parse_catalog(document: MarketplaceDocument) -> MarketplaceCatalog {
         explicit => explicit,
     };
 
-    match format {
+    let mut catalog = match format {
         MarketplaceFormat::Kimi => kimi::parse_kimi_catalog(document),
         MarketplaceFormat::Claude => claude::parse_claude_catalog(document),
         MarketplaceFormat::Codex => codex::parse_codex_catalog(document),
         MarketplaceFormat::Codewhale => codewhale::parse_codewhale_catalog(document),
         MarketplaceFormat::Auto => unreachable!("resolved above"),
+    };
+    flag_duplicate_names(&mut catalog);
+    catalog
+}
+
+/// A candidate's name is its identity: `/plugin show`, install, and registry
+/// lookups all resolve by it, and a lookup returns the first match. Entries
+/// sharing a name (after each format's normalization) are ambiguous, so every
+/// one of them carries an error and none installs from the catalog.
+fn flag_duplicate_names(catalog: &mut super::types::MarketplaceCatalog) {
+    let mut counts: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
+    for candidate in &catalog.candidates {
+        *counts.entry(candidate.name.clone()).or_default() += 1;
+    }
+    for candidate in &mut catalog.candidates {
+        let count = counts.get(&candidate.name).copied().unwrap_or(0);
+        if count > 1 {
+            candidate.diagnostics.push(MarketplaceDiagnostic::error(
+                "DUPLICATE_NAME",
+                format!(
+                    "{count} catalog entries are named `{}`; the entry's identity is ambiguous",
+                    candidate.name
+                ),
+                Some(candidate.name.clone()),
+                None,
+            ));
+        }
     }
 }
 

crates/tui/src/plugins/marketplace/tests.rs

@@ -607,3 +607,81 @@ fn claude_relative_install_source_is_rooted_outside_catalog_metadata() {
         tmp.path().join("external_plugins/linear")
     );
 }
+
+#[test]
+fn catalog_names_are_validated_and_duplicates_fail_closed() {
+    let body = r#"{
+        "name": "team",
+        "plugins": [
+          {"name": "fmt", "source": "github:owner/fmt"},
+          {"name": "fmt", "source": "github:someone-else/fmt"},
+          {"name": "bad\u001b[31mname", "source": "github:owner/x"},
+          {"name": "ok", "source": "github:owner/ok"}
+        ]
+      }"#;
+    let catalog = parse_auto("native", body);
+    assert_eq!(catalog.format, MarketplaceFormat::Codewhale);
+    assert_eq!(
+        catalog.total_candidates(),
+        3,
+        "a control-bearing name never becomes a candidate"
+    );
+    assert!(catalog.diagnostics.iter().any(|d| d.code == "INVALID_NAME"));
+    let duplicates: Vec<_> = catalog
+        .candidates
+        .iter()
+        .filter(|candidate| candidate.name == "fmt")
+        .collect();
+    assert_eq!(duplicates.len(), 2);
+    for candidate in duplicates {
+        assert!(
+            candidate
+                .diagnostics
+                .iter()
+                .any(|d| d.code == "DUPLICATE_NAME"),
+            "an ambiguous identity must not be installable"
+        );
+        assert!(candidate.has_errors());
+    }
+    assert!(!catalog.candidate_by_name("ok").unwrap().has_errors());
+}
+
+#[test]
+fn catalog_local_sources_cannot_leave_the_catalog_directory() {
+    use super::document::{
+        CatalogInstallResolution, load_catalog_document, resolve_candidate_install,
+    };
+    let tmp = tempfile::tempdir().unwrap();
+    let path = tmp.path().join("marketplace.json");
+    std::fs::write(
+        &path,
+        r#"{"name":"n","plugins":[
+            {"name":"absolute","source":{"source":"local","path":"/etc"}},
+            {"name":"parent","source":{"source":"local","path":"./plugins/../../outside"}},
+            {"name":"inside","source":{"source":"local","path":"./plugins/inside"}}
+        ]}"#,
+    )
+    .unwrap();
+    let loaded = load_catalog_document("n", tmp.path(), path.to_str().unwrap()).unwrap();
+    assert_eq!(loaded.entry.catalog.format, MarketplaceFormat::Codex);
+    let registry = crate::plugins::PluginRegistry::empty(tmp.path());
+    for name in ["absolute", "parent"] {
+        let candidate = loaded.entry.catalog.candidate_by_name(name).unwrap();
+        match resolve_candidate_install(&loaded.entry, candidate, &registry) {
+            CatalogInstallResolution::Unsupported { reason } => {
+                assert!(reason.contains("leaves the catalog directory"), "{reason}");
+            }
+            _ => panic!("`{name}` must not resolve outside the catalog"),
+        }
+    }
+    let inside = loaded.entry.catalog.candidate_by_name("inside").unwrap();
+    let CatalogInstallResolution::Supported { spec, .. } =
+        resolve_candidate_install(&loaded.entry, inside, &registry)
+    else {
+        panic!("a path inside the catalog stays installable")
+    };
+    assert_eq!(
+        std::path::Path::new(spec.strip_prefix("path:").unwrap()),
+        tmp.path().join("./plugins/inside")
+    );
+}

crates/tui/src/repl/runtime.rs

@@ -141,6 +141,14 @@ pub trait RpcDispatcher: Send + Sync {
 // ---------------------------------------------------------------------------
 
 const DEFAULT_STDOUT_LIMIT: usize = 8_192;
+/// Longest single stdout line (protocol or plain output) held in memory. An
+/// RPC request is one line, so this also bounds what a request can make Rust
+/// deserialize before the bridge's own caps apply; an over-long request is
+/// truncated, fails to parse, and is answered with an error.
+const MAX_STDOUT_LINE_BYTES: usize = 32 * 1024 * 1024;
+/// Most stdout one round retains (`full_stdout`); the rest is counted and
+/// reported, never buffered.
+const MAX_ROUND_STDOUT_BYTES: usize = 32 * 1024 * 1024;
 const ROUND_TIMEOUT: Duration = Duration::from_secs(180);
 #[cfg(not(windows))]
 const SPAWN_READY_TIMEOUT: Duration = Duration::from_secs(10);
@@ -165,6 +173,13 @@ pub struct PythonRuntime {
     round_count: u64,
     started: Instant,
     round_timeout: Option<Duration>,
+    /// Set while a round has not yet read its own DONE sentinel, and kept
+    /// when a round ends without it (timeout, I/O failure, or a caller that
+    /// dropped the round mid-flight). The protocol stream is then out of step
+    /// — the interrupted round may still print — so a failed round kills the
+    /// interpreter and every later round refuses instead of reading another
+    /// round's output as its own.
+    broken: Option<String>,
 }
 
 impl PythonRuntime {
@@ -250,6 +265,7 @@ impl PythonRuntime {
             round_count: 0,
             started: Instant::now(),
             round_timeout,
+            broken: None,
         };
 
         // Wait for `__RLM_READY_<sid>__` before handing control back. If
@@ -289,18 +305,43 @@ impl PythonRuntime {
         }
     }
 
+    /// One stdout line, holding at most [`MAX_STDOUT_LINE_BYTES`] of it: the
+    /// rest of an over-long line is consumed and replaced by an omission note.
     async fn read_stdout_line_lossy(&mut self) -> Result<Option<String>, String> {
         let mut buf = Vec::new();
-        let n = self
-            .stdout
-            .read_until(b'\n', &mut buf)
-            .await
-            .map_err(|e| format!("stdout read: {e}"))?;
-        if n == 0 {
-            Ok(None)
-        } else {
-            Ok(Some(String::from_utf8_lossy(&buf).into_owned()))
+        let mut omitted = 0usize;
+        loop {
+            let available = self
+                .stdout
+                .fill_buf()
+                .await
+                .map_err(|e| format!("stdout read: {e}"))?;
+            if available.is_empty() {
+                break;
+            }
+            let (take, ends_line) = match available.iter().position(|&byte| byte == b'\n') {
+                Some(index) => (index + 1, true),
+                None => (available.len(), false),
+            };
+            let keep = take.min(MAX_STDOUT_LINE_BYTES.saturating_sub(buf.len()));
+            buf.extend_from_slice(&available[..keep]);
+            omitted += take - keep;
+            self.stdout.consume(take);
+            if ends_line {
+                break;
+            }
         }
+        if buf.is_empty() && omitted == 0 {
+            return Ok(None);
+        }
+        let mut line = String::from_utf8_lossy(&buf).into_owned();
+        if omitted > 0 {
+            line.truncate(line.trim_end_matches(['\n', '\r']).len());
+            line.push_str(&format!(
+                "[... {omitted} bytes of this line omitted by the REPL line limit ...]\n"
+            ));
+        }
+        Ok(Some(line))
     }
 
     /// Execute a Python code block with no RPC dispatcher. Used for inline
@@ -359,9 +400,18 @@ impl PythonRuntime {
     where
         D: RpcDispatcher + ?Sized,
     {
+        if let Some(reason) = &self.broken {
+            return Err(format!(
+                "Python REPL is unusable after an earlier round failed ({reason}); start a new REPL"
+            ));
+        }
+        // Pessimistic until this round reads its own DONE: a caller that
+        // drops this future mid-round leaves the stream out of step too.
+        self.broken = Some("an earlier round was cancelled before it finished".to_string());
         let started = Instant::now();
         self.round_count += 1;
         let round_id = self.round_count;
+        let round_tag = round_id.to_string();
 
         // Send the code header + body + end marker in one write.
         let header = format!("__RLM_RUN_{}__::{round_id}\n", self.session_id);
@@ -383,6 +433,7 @@ impl PythonRuntime {
         let done_prefix = format!("__RLM_DONE_{}__::", self.session_id);
 
         let mut stdout_buf = String::new();
+        let mut stdout_omitted = 0usize;
         let mut final_value: Option<String> = None;
         let mut final_json: Option<Value> = None;
         let mut final_confidence: Option<Value> = None;
@@ -401,8 +452,13 @@ impl PythonRuntime {
                 let trimmed = line.trim_end_matches(['\n', '\r']);
 
                 if let Some(rest) = trimmed.strip_prefix(&done_prefix) {
-                    let _ = rest;
-                    break;
+                    if rest == round_tag {
+                        break;
+                    }
+                    // Another round's completion (or user output imitating
+                    // one) never ends this round.
+                    push_bounded(&mut stdout_buf, &line, &mut stdout_omitted);
+                    continue;
                 }
                 if let Some(rest) = trimmed.strip_prefix(&final_prefix) {
                     // New sessions emit an object with value/confidence;
@@ -442,7 +498,11 @@ impl PythonRuntime {
                     let traceback =
                         serde_json::from_str::<String>(rest).unwrap_or_else(|_| rest.to_string());
                     had_error = true;
-                    stdout_buf.push_str(&format!("[traceback]\n{traceback}\n"));
+                    push_bounded(
+                        &mut stdout_buf,
+                        &format!("[traceback]\n{traceback}\n"),
+                        &mut stdout_omitted,
+                    );
                     continue;
                 }
                 if let Some(rest) = trimmed.strip_prefix(&req_prefix) {
@@ -470,24 +530,35 @@ impl PythonRuntime {
                     continue;
                 }
 
-                stdout_buf.push_str(&line);
+                push_bounded(&mut stdout_buf, &line, &mut stdout_omitted);
             }
             Ok::<_, String>(())
         };
 
-        if let Some(round_timeout) = round_timeout {
-            match tokio::time::timeout(round_timeout, read_loop).await {
-                Ok(Ok(())) => {}
-                Ok(Err(e)) => return Err(e),
-                Err(_) => {
-                    return Err(format!(
+        let outcome = match round_timeout {
+            Some(round_timeout) => tokio::time::timeout(round_timeout, read_loop)
+                .await
+                .unwrap_or_else(|_| {
+                    Err(format!(
                         "REPL round timed out after {}s",
                         round_timeout.as_secs()
-                    ));
-                }
-            }
-        } else {
-            read_loop.await?;
+                    ))
+                }),
+            None => read_loop.await,
+        };
+        if let Err(error) = outcome {
+            // The interpreter may still be executing this round. Kill it now
+            // (the handle is reaped on drop) rather than leave it running and
+            // let its late output land in a later round.
+            let _ = self.child.start_kill();
+            self.broken = Some(error.clone());
+            return Err(error);
+        }
+        self.broken = None;
+        if stdout_omitted > 0 {
+            stdout_buf.push_str(&format!(
+                "\n[... REPL round output capped: {stdout_omitted} bytes omitted ...]\n"
+            ));
         }
 
         let stderr = self.drain_stderr().await;
@@ -1038,6 +1109,22 @@ _main_loop()
 // Helpers
 // ---------------------------------------------------------------------------
 
+/// Append to a round's retained stdout without exceeding
+/// [`MAX_ROUND_STDOUT_BYTES`]; whatever does not fit is only counted.
+fn push_bounded(buf: &mut String, text: &str, omitted: &mut usize) {
+    let room = MAX_ROUND_STDOUT_BYTES.saturating_sub(buf.len());
+    if text.len() <= room {
+        buf.push_str(text);
+        return;
+    }
+    let cut = (0..=room)
+        .rev()
+        .find(|&index| text.is_char_boundary(index))
+        .unwrap_or(0);
+    buf.push_str(&text[..cut]);
+    *omitted += text.len() - cut;
+}
+
 fn truncate_stdout(stdout: &str, limit: usize) -> String {
     if stdout.len() <= limit {
         return stdout.to_string();
@@ -1352,6 +1439,50 @@ mod tests {
         rt.shutdown().await;
     }
 
+    #[tokio::test]
+    async fn timed_out_round_kills_the_interpreter_and_refuses_later_rounds() {
+        let mut rt = PythonRuntime::new().await.expect("spawn");
+        rt.round_timeout = Some(Duration::from_millis(300));
+        let error = rt
+            .execute("import time\ntime.sleep(30)\nprint('late output')")
+            .await
+            .expect_err("the round must time out");
+        assert!(error.contains("timed out"), "{error}");
+
+        let refused = rt
+            .execute("print('next round')")
+            .await
+            .expect_err("a desynchronized REPL must not run another round");
+        assert!(refused.contains("unusable"), "{refused}");
+
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while rt.child.try_wait().expect("child status").is_none() {
+            assert!(
+                Instant::now() < deadline,
+                "the timed-out interpreter must be killed"
+            );
+            tokio::time::sleep(Duration::from_millis(25)).await;
+        }
+    }
+
+    #[tokio::test]
+    async fn another_rounds_done_sentinel_does_not_end_this_round() {
+        let mut rt = PythonRuntime::new().await.expect("spawn");
+        let round = rt
+            .execute("print(_DONE + '999')\nprint('still this round')")
+            .await
+            .expect("execute");
+        assert!(
+            round.stdout.contains("still this round"),
+            "{}",
+            round.stdout
+        );
+        let next = rt.execute("print('second')").await.expect("next round");
+        assert!(next.stdout.contains("second"), "{}", next.stdout);
+        assert!(!next.stdout.contains("still this round"), "{}", next.stdout);
+        rt.shutdown().await;
+    }
+
     #[tokio::test]
     async fn inline_runtime_keeps_bounded_round_timeout() {
         let rt = PythonRuntime::new().await.expect("spawn");

crates/tui/src/runtime_api.rs

@@ -7163,6 +7163,7 @@ async fn complete_thread_goal(
         .transition_goal_status(
             &id,
             &goal.goal_id,
+            goal.status.clone(),
             codewhale_protocol::ThreadGoalStatus::Complete,
         )
         .await
@@ -7215,6 +7216,7 @@ async fn block_thread_goal(
         .transition_goal_status(
             &id,
             &goal.goal_id,
+            goal.status.clone(),
             codewhale_protocol::ThreadGoalStatus::Blocked,
         )
         .await

crates/tui/src/runtime_threads.rs

@@ -13,7 +13,7 @@
 
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::fs::{self, File, OpenOptions};
-use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
+use std::io::{self, BufRead, BufReader, Read, Seek, SeekFrom, Write};
 use std::path::{Component, Path, PathBuf};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -2697,17 +2697,25 @@ impl RuntimeThreadStore {
 
     /// A late turn can only update the revision admitted with that turn.
     /// Load/compare/write share the same guard as explicit save/delete.
+    /// Mutate the goal only while it is still the revision the caller read:
+    /// the same `goal_id` and, when `expected_status` is given, the same
+    /// status. Returns `Ok(None)` without writing when either changed, so a
+    /// transition decided against a stale read (for example Blocked computed
+    /// while a concurrent Complete landed) can never overwrite the newer state.
     fn update_goal_if_revision(
         &self,
         thread_id: &str,
         goal_id: &str,
+        expected_status: Option<codewhale_protocol::ThreadGoalStatus>,
         update: impl FnOnce(&mut codewhale_protocol::ThreadGoal),
     ) -> Result<Option<codewhale_protocol::ThreadGoal>> {
         let _guard = self.goal_mutation.lock();
         let Some(mut goal) = self.load_goal(thread_id)? else {
             return Ok(None);
         };
-        if goal.goal_id != goal_id {
+        if goal.goal_id != goal_id
+            || expected_status.is_some_and(|expected| goal.status != expected)
+        {
             return Ok(None);
         }
         update(&mut goal);
@@ -2847,6 +2855,128 @@ impl RuntimeThreadStore {
         Ok(())
     }
 
+    /// Journal for one history seed into an existing thread (#6555), kept
+    /// beside the thread record as `<thread>.seed` (listings read only
+    /// `.json`). It names every record the seed will write before the first
+    /// one lands; the thread record — the seed's commit record — is written
+    /// after them, and the journal is removed last.
+    fn seed_journal_path(&self, thread_id: &str) -> Result<PathBuf> {
+        Self::record_path(&self.threads_dir, thread_id, "seed", "thread id")
+    }
+
+    fn save_seed_journal(&self, journal: &SeedJournal) -> Result<()> {
+        let path = self.seed_journal_path(&journal.thread_id)?;
+        reject_symlinked_store_file(&path)?;
+        write_json_atomic(&path, journal)
+            .with_context(|| format!("Failed to write seed journal {}", path.display()))
+    }
+
+    fn remove_seed_journal(&self, thread_id: &str) -> Result<()> {
+        remove_file_if_exists(&self.seed_journal_path(thread_id)?)
+    }
+
+    /// Remove every record an uncommitted seed named, then its journal. Ids in
+    /// a journal were minted fresh for that seed, so nothing else is touched.
+    fn discard_seed(&self, journal: &SeedJournal) -> Result<()> {
+        // Validate the whole deletion set first. A damaged journal must not
+        // retire records from another thread, even when some earlier records
+        // in its list were legitimate members of this seed.
+        for turn_id in &journal.turn_ids {
+            match self.load_turn(turn_id) {
+                Ok(turn) if turn.id == *turn_id && turn.thread_id == journal.thread_id => {}
+                Ok(_) => anyhow::bail!("seed journal names a turn outside its thread"),
+                Err(error)
+                    if error
+                        .downcast_ref::<io::Error>()
+                        .is_some_and(|error| error.kind() == io::ErrorKind::NotFound) => {}
+                Err(error) => {
+                    return Err(error).context("cannot validate seed turn before cleanup");
+                }
+            }
+        }
+        for item_id in &journal.item_ids {
+            match self.load_item(item_id) {
+                Ok(item) if item.id == *item_id && journal.turn_ids.contains(&item.turn_id) => {}
+                Ok(_) => anyhow::bail!("seed journal names an item outside its turns"),
+                Err(error)
+                    if error
+                        .downcast_ref::<io::Error>()
+                        .is_some_and(|error| error.kind() == io::ErrorKind::NotFound) => {}
+                Err(error) => {
+                    return Err(error).context("cannot validate seed item before cleanup");
+                }
+            }
+        }
+        for turn_id in &journal.turn_ids {
+            self.remove_turn(turn_id)?;
+        }
+        for item_id in &journal.item_ids {
+            self.remove_item(item_id)?;
+        }
+        self.remove_seed_journal(&journal.thread_id)
+    }
+
+    /// Startup half of the seed transaction: a journal whose thread pointer
+    /// still matches the one it recorded is an unpublished partial seed and
+    /// is discarded, so recovery can never restore it as history. A journal
+    /// whose pointer names its final turn belongs to a committed seed and is
+    /// simply removed. Uncertain intent or ownership prevents recovery.
+    fn settle_seed_journals(&self) -> Result<()> {
+        let threads_dir = checked_existing_runtime_store_dir(&self.threads_dir)?;
+        for entry in fs::read_dir(&threads_dir)
+            .with_context(|| format!("Failed to read {}", threads_dir.display()))?
+        {
+            let path = entry?.path();
+            if path.extension().is_none_or(|ext| ext != "seed") {
+                continue;
+            }
+            // Recovery recomputes latest-turn pointers from every turn file.
+            // Without readable intent it cannot distinguish an unpublished
+            // seed from accepted history. Refuse startup and preserve evidence
+            // instead of publishing the incomplete transaction.
+            let journal = read_store_file(&path)
+                .and_then(|raw| {
+                    serde_json::from_str::<SeedJournal>(&raw)
+                        .with_context(|| format!("Failed to parse {}", path.display()))
+                })
+                .with_context(|| {
+                    format!(
+                        "seed recovery refused; preserve {} for repair",
+                        path.display()
+                    )
+                })?;
+            if self.seed_journal_path(&journal.thread_id)?.file_name() != path.file_name() {
+                anyhow::bail!(
+                    "seed journal {} names another thread; recovery refused",
+                    path.display()
+                );
+            }
+            let committed = match self.load_thread(&journal.thread_id) {
+                Ok(thread) if thread.latest_turn_id == journal.previous_latest_turn_id => false,
+                Ok(thread) if journal.turn_ids.last() == thread.latest_turn_id.as_ref() => true,
+                Ok(_) => {
+                    anyhow::bail!("seed journal commit pointer is inconsistent; recovery refused")
+                }
+                Err(error)
+                    if error
+                        .downcast_ref::<io::Error>()
+                        .is_some_and(|error| error.kind() == io::ErrorKind::NotFound) =>
+                {
+                    false
+                }
+                Err(error) => {
+                    return Err(error).context("seed thread cannot be read; recovery refused");
+                }
+            };
+            if committed {
+                self.remove_seed_journal(&journal.thread_id)?;
+            } else {
+                self.discard_seed(&journal)?;
+            }
+        }
+        Ok(())
+    }
+
     fn remove_turn(&self, turn_id: &str) -> Result<()> {
         remove_file_if_exists(&self.turn_path(turn_id)?)
     }
@@ -6205,6 +6335,18 @@ impl RuntimeThreadManager {
     }
 }
 
+/// Durable intent of one `seed_thread_from_messages` transaction (#6555).
+/// The seed committed when the thread's `latest_turn_id` moved from
+/// `previous_latest_turn_id` to the final journaled turn; see
+/// [`RuntimeThreadStore::settle_seed_journals`].
+#[derive(Debug, Serialize, Deserialize)]
+struct SeedJournal {
+    thread_id: String,
+    previous_latest_turn_id: Option<String>,
+    turn_ids: Vec<String>,
+    item_ids: Vec<String>,
+}
+
 /// Helper types for `seed_thread_from_messages` — intermediate representation
 /// of a turn being built from session messages before persisting as items.
 ///
@@ -7698,23 +7840,30 @@ impl RuntimeThreadManager {
 
     /// Transition the goal status through a revision-checked mutation. A
     /// stale load-then-save could otherwise overwrite a concurrent PUT
-    /// replacement or DELETE; here the write commits only when the record is
-    /// still the revision the caller read. Returns the updated goal, or
-    /// `Ok(None)` when the goal changed or vanished since the read.
+    /// replacement or DELETE, or a concurrent terminal transition; here the
+    /// write commits only when the record still carries the goal id *and* the
+    /// status the caller read. Returns the updated goal, or `Ok(None)` when
+    /// the goal changed or vanished since the read.
     pub async fn transition_goal_status(
         &self,
         thread_id: &str,
         expected_goal_id: &str,
+        expected_status: codewhale_protocol::ThreadGoalStatus,
         status: codewhale_protocol::ThreadGoalStatus,
     ) -> Result<Option<codewhale_protocol::ThreadGoal>> {
         let thread_id = thread_id.to_string();
         let expected_goal_id = expected_goal_id.to_string();
         let store = self.store.clone();
         tokio::task::spawn_blocking(move || {
-            store.update_goal_if_revision(&thread_id, &expected_goal_id, |goal| {
-                goal.status = status;
-                goal.updated_at = chrono::Utc::now().timestamp();
-            })
+            store.update_goal_if_revision(
+                &thread_id,
+                &expected_goal_id,
+                Some(expected_status),
+                |goal| {
+                    goal.status = status;
+                    goal.updated_at = chrono::Utc::now().timestamp();
+                },
+            )
         })
         .await
         .context("goal status transition task panicked")?
@@ -7828,7 +7977,7 @@ impl RuntimeThreadManager {
             return;
         };
         let mut continue_after: Option<u64> = None;
-        let updated = self.store.update_goal_if_revision(thread_id, admitted_goal_id, |goal| {
+        let updated = self.store.update_goal_if_revision(thread_id, admitted_goal_id, None, |goal| {
         // Accrue this turn's provider spend onto the durable counters. The
         // engine tracks the same totals in memory; the record is the
         // cross-restart authority.
@@ -11251,6 +11400,9 @@ impl RuntimeThreadManager {
             }
         }
 
+        let previous_latest_turn_id = thread.latest_turn_id.clone();
+        let mut seed_items: Vec<TurnItemRecord> = Vec::new();
+        let mut seed_turns: Vec<TurnRecord> = Vec::new();
         for turn_seed in turns {
             let turn_at = next_seed_stamp();
             let turn_id = runtime_record_id("turn");
@@ -11280,7 +11432,7 @@ impl RuntimeThreadManager {
                     item.set_image_content(turn_seed.image_content.clone());
                     thread.schema_version = IMAGE_RUNTIME_SCHEMA_VERSION;
                 }
-                self.store.save_item(&item)?;
+                seed_items.push(item);
                 item_ids.push(item_id);
             }
 
@@ -11295,7 +11447,7 @@ impl RuntimeThreadManager {
                         } else {
                             text.clone()
                         };
-                        self.store.save_item(&TurnItemRecord {
+                        seed_items.push(TurnItemRecord {
                             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
                             id: item_id.clone(),
                             turn_id: turn_id.clone(),
@@ -11308,7 +11460,7 @@ impl RuntimeThreadManager {
                             artifacts: Vec::new(),
                             started_at: Some(item_at),
                             ended_at: Some(item_at),
-                        })?;
+                        });
                     }
                     SeedItem::Thinking(thinking) => {
                         let thinking_summary = if thinking.len() > SUMMARY_LIMIT {
@@ -11316,7 +11468,7 @@ impl RuntimeThreadManager {
                         } else {
                             thinking.clone()
                         };
-                        self.store.save_item(&TurnItemRecord {
+                        seed_items.push(TurnItemRecord {
                             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
                             id: item_id.clone(),
                             turn_id: turn_id.clone(),
@@ -11329,7 +11481,7 @@ impl RuntimeThreadManager {
                             artifacts: Vec::new(),
                             started_at: Some(item_at),
                             ended_at: Some(item_at),
-                        })?;
+                        });
                     }
                     SeedItem::ToolUse {
                         id: tool_id,
@@ -11349,7 +11501,7 @@ impl RuntimeThreadManager {
                                 s.clone()
                             }
                         });
-                        self.store.save_item(&TurnItemRecord {
+                        seed_items.push(TurnItemRecord {
                             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
                             id: item_id.clone(),
                             turn_id: turn_id.clone(),
@@ -11378,7 +11530,7 @@ impl RuntimeThreadManager {
                             artifacts: Vec::new(),
                             started_at: Some(item_at),
                             ended_at: Some(item_at),
-                        })?;
+                        });
                     }
                     SeedItem::ToolResult {
                         tool_use_id,
@@ -11406,7 +11558,7 @@ impl RuntimeThreadManager {
                             metadata
                                 .insert("content_blocks".to_string(), Value::Array(blocks.clone()));
                         }
-                        self.store.save_item(&TurnItemRecord {
+                        seed_items.push(TurnItemRecord {
                             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
                             id: item_id.clone(),
                             turn_id: turn_id.clone(),
@@ -11423,7 +11575,7 @@ impl RuntimeThreadManager {
                             artifacts: Vec::new(),
                             started_at: Some(item_at),
                             ended_at: Some(item_at),
-                        })?;
+                        });
                     }
                 }
                 item_ids.push(item_id);
@@ -11431,7 +11583,7 @@ impl RuntimeThreadManager {
 
             // Only create a turn if there's content.
             if !item_ids.is_empty() {
-                self.store.save_turn(&TurnRecord {
+                seed_turns.push(TurnRecord {
                     max_output_tokens: None,
                     schema_version: if turn_seed.image_content.is_empty() {
                         CURRENT_RUNTIME_SCHEMA_VERSION
@@ -11474,14 +11626,49 @@ impl RuntimeThreadManager {
                     artifacts: Vec::new(),
                     workspace: None,
                     workspace_snapshots: Vec::new(),
-                })?;
+                });
 
                 thread.latest_turn_id = Some(turn_id);
                 thread.updated_at = turn_at;
             }
         }
 
-        self.store.save_thread(&thread)?;
+        // One journaled transaction (#6555): the journal names every record
+        // first, then items, their turns, and last the thread record — the
+        // commit record that advances `latest_turn_id`. A crash in between
+        // leaves a journal whose pointer never moved, and startup recovery
+        // discards those records instead of restoring a partial history.
+        let journal = SeedJournal {
+            thread_id: thread_id.to_string(),
+            previous_latest_turn_id,
+            turn_ids: seed_turns.iter().map(|turn| turn.id.clone()).collect(),
+            item_ids: seed_items.iter().map(|item| item.id.clone()).collect(),
+        };
+        self.store.save_seed_journal(&journal)?;
+        let published = (|| -> Result<()> {
+            self.store
+                .save_items_batch(&seed_items.iter().collect::<Vec<_>>())?;
+            for turn in &seed_turns {
+                self.store.save_turn(turn)?;
+            }
+            self.store.save_thread(&thread)
+        })();
+        if let Err(error) = published {
+            // Roll back now; a rollback that fails leaves the journal, and
+            // startup finishes it.
+            if let Err(cleanup) = self.store.discard_seed(&journal) {
+                tracing::warn!(
+                    target: "runtime",
+                    thread_id,
+                    "partial seed left for startup recovery: {cleanup:#}"
+                );
+            }
+            return Err(error);
+        }
+        // Committed. A journal that outlives this line is settled as committed.
+        if let Err(error) = self.store.remove_seed_journal(thread_id) {
+            tracing::warn!(target: "runtime", thread_id, "seed journal not removed: {error:#}");
+        }
         drop(thread_mutation);
         self.emit_event(
             thread_id,
@@ -16028,7 +16215,7 @@ impl RuntimeThreadManager {
                         // Persist an acknowledged review before awaiting another
                         // event, so restart midway through a turn retains it.
                         self.store
-                            .update_goal_if_revision(&thread_id, goal_id, |goal| {
+                            .update_goal_if_revision(&thread_id, goal_id, None, |goal| {
                                 merge_engine_goal_progress(goal, &snapshot);
                             })?;
                     } else if snapshot.is_active() {
@@ -16538,6 +16725,9 @@ impl RuntimeThreadManager {
     }
 
     fn recover_interrupted_state(&self) -> Result<()> {
+        // Unpublished history seeds go first: the pointer recomputation below
+        // would otherwise publish their turns as recovered history (#6555).
+        self.store.settle_seed_journals()?;
         let now = Utc::now();
         let mut threads = self
             .store

crates/tui/src/runtime_threads/tests.rs

@@ -11729,6 +11729,216 @@ async fn live_goal_progress_is_revision_fenced_and_settled_once() -> Result<()>
     Ok(())
 }
 
+#[tokio::test]
+async fn startup_discards_an_unpublished_partial_seed_and_keeps_a_committed_one() -> Result<()> {
+    let runtime_dir = test_runtime_dir();
+    let manager = test_manager(runtime_dir.clone())?;
+    let messages: Vec<Message> = serde_json::from_value(json!([
+        {"role":"user","content":[{"type":"text","text":"SEEDED"}]},
+        {"role":"assistant","content":[{"type":"text","text":"SEEDED ANSWER"}]}
+    ]))?;
+    let mut fixtures = Vec::new();
+    for _ in 0..2 {
+        let thread = manager
+            .create_thread(CreateThreadRequest::default())
+            .await?;
+        manager
+            .seed_thread_from_messages(&thread.id, &messages)
+            .await?;
+        assert!(
+            !manager.store.seed_journal_path(&thread.id)?.exists(),
+            "a committed seed leaves no journal"
+        );
+        let turns = manager.store.list_turns_for_thread(&thread.id)?;
+        assert_eq!(turns.len(), 1);
+        let items = manager.store.list_items_for_turn(&turns[0].id)?;
+        assert_eq!(items.len(), 2);
+        // Fault fixture: the journal the seed wrote before its first record.
+        manager.store.save_seed_journal(&SeedJournal {
+            thread_id: thread.id.clone(),
+            previous_latest_turn_id: None,
+            turn_ids: turns.iter().map(|turn| turn.id.clone()).collect(),
+            item_ids: items.iter().map(|item| item.id.clone()).collect(),
+        })?;
+        fixtures.push((thread.id, turns, items));
+    }
+    // The first thread crashed before its commit record: every turn and item
+    // is on disk, but the thread pointer never advanced.
+    let (partial_id, partial_turns, partial_items) = &fixtures[0];
+    let mut uncommitted = manager.store.load_thread(partial_id)?;
+    uncommitted.latest_turn_id = None;
+    manager.store.save_thread(&uncommitted)?;
+    drop(manager);
+
+    let reopened = test_manager(runtime_dir)?;
+    assert!(
+        reopened.store.list_turns_for_thread(partial_id)?.is_empty(),
+        "an unpublished partial seed must not be restored as history"
+    );
+    for item in partial_items {
+        assert!(!reopened.store.item_path(&item.id)?.exists());
+    }
+    assert!(!reopened.store.turn_path(&partial_turns[0].id)?.exists());
+    assert_eq!(reopened.store.load_thread(partial_id)?.latest_turn_id, None);
+    assert!(!reopened.store.seed_journal_path(partial_id)?.exists());
+
+    // The second crashed after its commit, before removing the journal.
+    let (committed_id, committed_turns, _) = &fixtures[1];
+    assert_eq!(
+        reopened.store.list_turns_for_thread(committed_id)?.len(),
+        1,
+        "a committed seed survives recovery"
+    );
+    assert_eq!(
+        reopened.store.load_thread(committed_id)?.latest_turn_id,
+        Some(committed_turns[0].id.clone())
+    );
+    assert!(!reopened.store.seed_journal_path(committed_id)?.exists());
+    Ok(())
+}
+
+#[tokio::test]
+async fn startup_refuses_uncertain_seed_journals_without_publishing_partial_history() -> Result<()>
+{
+    for failure in [
+        "unreadable-journal",
+        "wrong-thread",
+        "unreadable-thread",
+        "wrong-commit",
+    ] {
+        let runtime_dir = test_runtime_dir();
+        let manager = test_manager(runtime_dir.clone())?;
+        let messages: Vec<Message> = serde_json::from_value(json!([
+            {"role":"user","content":[{"type":"text","text":"seed recovery canary"}]}
+        ]))?;
+        let thread = manager
+            .create_thread(CreateThreadRequest::default())
+            .await?;
+        manager
+            .seed_thread_from_messages(&thread.id, &messages)
+            .await?;
+        let turns = manager.store.list_turns_for_thread(&thread.id)?;
+        let items = manager.store.list_items_for_turn(&turns[0].id)?;
+        let mut uncommitted = manager.store.load_thread(&thread.id)?;
+        uncommitted.latest_turn_id = None;
+        manager.store.save_thread(&uncommitted)?;
+        let journal = SeedJournal {
+            thread_id: thread.id.clone(),
+            previous_latest_turn_id: None,
+            turn_ids: turns.iter().map(|turn| turn.id.clone()).collect(),
+            item_ids: items.iter().map(|item| item.id.clone()).collect(),
+        };
+        manager.store.save_seed_journal(&journal)?;
+        let journal_path = manager.store.seed_journal_path(&thread.id)?;
+        let thread_path = manager.store.thread_path(&thread.id)?;
+        match failure {
+            "unreadable-journal" => fs::write(&journal_path, b"{truncated seed intent")?,
+            "wrong-thread" => {
+                let mut wrong = serde_json::to_value(&journal)?;
+                wrong["thread_id"] = json!("another-thread");
+                fs::write(&journal_path, serde_json::to_vec(&wrong)?)?;
+            }
+            "unreadable-thread" => fs::write(&thread_path, b"{truncated thread")?,
+            "wrong-commit" => {
+                uncommitted.latest_turn_id = Some("unrelated-turn".into());
+                manager.store.save_thread(&uncommitted)?;
+            }
+            _ => unreachable!(),
+        }
+        let mut paths = vec![journal_path, thread_path];
+        for turn in &turns {
+            paths.push(manager.store.turn_path(&turn.id)?);
+        }
+        for item in &items {
+            paths.push(manager.store.item_path(&item.id)?);
+        }
+        let original = paths
+            .iter()
+            .map(fs::read)
+            .collect::<std::io::Result<Vec<_>>>()?;
+        drop(manager);
+        let error = test_manager(runtime_dir)
+            .err()
+            .expect("uncertain seed must refuse startup");
+        assert!(
+            format!("{error:#}").contains("seed"),
+            "{failure}: {error:#}"
+        );
+        for (path, bytes) in paths.iter().zip(original) {
+            assert_eq!(
+                fs::read(path)?,
+                bytes,
+                "{failure}: retain {}",
+                path.display()
+            );
+        }
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn seed_cleanup_validates_every_record_before_removing_anything() -> Result<()> {
+    for foreign in ["turn", "item"] {
+        let manager = test_manager(test_runtime_dir())?;
+        let messages: Vec<Message> = serde_json::from_value(json!([
+            {"role":"user","content":[{"type":"text","text":"keep both histories"}]}
+        ]))?;
+        let mut histories = Vec::new();
+        for _ in 0..2 {
+            let thread = manager
+                .create_thread(CreateThreadRequest::default())
+                .await?;
+            manager
+                .seed_thread_from_messages(&thread.id, &messages)
+                .await?;
+            let turns = manager.store.list_turns_for_thread(&thread.id)?;
+            let items = manager.store.list_items_for_turn(&turns[0].id)?;
+            histories.push((thread, turns, items));
+        }
+        let (thread, turns, items) = &histories[0];
+        let mut journal = SeedJournal {
+            thread_id: thread.id.clone(),
+            previous_latest_turn_id: None,
+            turn_ids: turns.iter().map(|turn| turn.id.clone()).collect(),
+            item_ids: items.iter().map(|item| item.id.clone()).collect(),
+        };
+        if foreign == "turn" {
+            journal.turn_ids.push(histories[1].1[0].id.clone());
+        } else {
+            journal.item_ids.push(histories[1].2[0].id.clone());
+        }
+        manager.store.save_seed_journal(&journal)?;
+        let mut paths = vec![manager.store.seed_journal_path(&thread.id)?];
+        for (thread, turns, items) in &histories {
+            paths.push(manager.store.thread_path(&thread.id)?);
+            for turn in turns {
+                paths.push(manager.store.turn_path(&turn.id)?);
+            }
+            for item in items {
+                paths.push(manager.store.item_path(&item.id)?);
+            }
+        }
+        let original = paths
+            .iter()
+            .map(fs::read)
+            .collect::<std::io::Result<Vec<_>>>()?;
+        let error = manager
+            .store
+            .discard_seed(&journal)
+            .expect_err("foreign history is not cleanup ownership");
+        assert!(format!("{error:#}").contains("outside"), "{error:#}");
+        for (path, bytes) in paths.iter().zip(original) {
+            assert_eq!(
+                fs::read(path)?,
+                bytes,
+                "validate the whole set before removing {}",
+                path.display()
+            );
+        }
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn transition_goal_status_commits_only_the_read_revision() -> Result<()> {
     let manager = test_manager(test_runtime_dir())?;
@@ -11743,6 +11953,7 @@ async fn transition_goal_status_commits_only_the_read_revision() -> Result<()> {
         .transition_goal_status(
             &thread.id,
             "goal_a",
+            codewhale_protocol::ThreadGoalStatus::Active,
             codewhale_protocol::ThreadGoalStatus::Complete,
         )
         .await?
@@ -11752,6 +11963,26 @@ async fn transition_goal_status_commits_only_the_read_revision() -> Result<()> {
         codewhale_protocol::ThreadGoalStatus::Complete
     );
 
+    // A Blocked transition decided from the earlier Active read races the
+    // Complete above; the terminal state must survive it.
+    let raced = manager
+        .transition_goal_status(
+            &thread.id,
+            "goal_a",
+            codewhale_protocol::ThreadGoalStatus::Active,
+            codewhale_protocol::ThreadGoalStatus::Blocked,
+        )
+        .await?;
+    assert!(raced.is_none(), "a stale status read must not commit");
+    assert_eq!(
+        manager
+            .store
+            .load_goal(&thread.id)?
+            .ok_or_else(|| anyhow::anyhow!("goal record missing"))?
+            .status,
+        codewhale_protocol::ThreadGoalStatus::Complete
+    );
+
     // A concurrent replacement makes the previously read revision stale.
     manager
         .store
@@ -11760,6 +11991,7 @@ async fn transition_goal_status_commits_only_the_read_revision() -> Result<()> {
         .transition_goal_status(
             &thread.id,
             "goal_a",
+            codewhale_protocol::ThreadGoalStatus::Active,
             codewhale_protocol::ThreadGoalStatus::Blocked,
         )
         .await?;

crates/tui/src/session_manager.rs

@@ -386,9 +386,10 @@ fn is_not_archived(archived: &bool) -> bool {
 /// the id here and any external writer is refused.
 ///
 /// A static registry rather than a field on `RuntimeApiState` because the
-/// embedded Runtime API runs inside the TUI process; a standalone
-/// `codewhale web` has an empty registry and is therefore never blocked, which
-/// is exactly right — there is no TUI holding anything.
+/// embedded Runtime API runs inside the TUI process. A standalone
+/// `codewhale web` has an empty registry, so external writers consult
+/// [`SessionManager::is_session_live_anywhere`], which also sees the
+/// cross-process lease a TUI in another process holds.
 static LIVE_SESSIONS: std::sync::OnceLock<std::sync::RwLock<std::collections::HashSet<String>>> =
     std::sync::OnceLock::new();
 
@@ -1624,27 +1625,48 @@ impl SessionManager {
         Err(session_open_elsewhere(id))
     }
 
+    /// Hold the existing live lease throughout an external mutation. A
+    /// liveness probe releases its lock before returning and cannot protect
+    /// the subsequent read/write from another surface attaching meanwhile.
+    fn reserve_session_for_external_write(&self, id: &str) -> io::Result<SessionLease> {
+        let live = live_sessions()
+            .read()
+            .map_err(|_| io::Error::other("session ownership registry unavailable"))?;
+        if live.contains(id.trim()) {
+            return Err(live_session_conflict(id));
+        }
+        drop(live);
+        let lease = self.reserve_session_for_attach(id)?;
+        // A same-process attach may have committed after the registry read.
+        // Its already-held lease is borrowed, never ours to mutate through.
+        if lease.file.is_none() {
+            return Err(live_session_conflict(id));
+        }
+        Ok(lease)
+    }
+
     /// Is `session_id` open in an interactive session in this process *or any
     /// other*? External writers (the Runtime API, retention) check this before
-    /// rewriting or deleting a document, because the process holding it would
-    /// revert the change at its next autosave (#6144).
+    /// listing recovery candidates. Mutations reserve the live lease instead;
+    /// a released probe cannot authorize a subsequent write (#6144).
     #[must_use]
     pub fn is_session_live_anywhere(&self, session_id: &str) -> bool {
         if is_live_session(session_id) {
             return true;
         }
         let Ok(path) = self.live_lease_path(session_id, false) else {
-            return false;
+            return true;
         };
         let file = match open_private_read_file(&path) {
             Ok(file) => file,
-            Err(_) => return false,
+            Err(error) if error.kind() == io::ErrorKind::NotFound => return false,
+            Err(_) => return true,
         };
         // Contention means a live holder; acquiring proves none, and the
         // probe's lock is released when `file` drops here.
-        matches!(
+        !matches!(
             crate::runtime_threads::try_lock_file_exclusive(&file),
-            Ok(false)
+            Ok(true)
         )
     }
 
@@ -2905,9 +2927,9 @@ impl SessionManager {
         archived: bool,
         mutator: SessionMutator,
     ) -> std::io::Result<SessionMetadata> {
-        if mutator == SessionMutator::External && is_live_session(id) {
-            return Err(live_session_conflict(id));
-        }
+        let _lease = (mutator == SessionMutator::External)
+            .then(|| self.reserve_session_for_external_write(id))
+            .transpose()?;
         let mut session = self.load_session(id)?;
         if session.metadata.archived == archived {
             return Ok(session.metadata);
@@ -2963,9 +2985,9 @@ impl SessionManager {
         mutator: SessionMutator,
     ) -> std::io::Result<SessionMetadata> {
         let title = normalize_session_title(title)?;
-        if mutator == SessionMutator::External && is_live_session(id) {
-            return Err(live_session_conflict(id));
-        }
+        let _lease = (mutator == SessionMutator::External)
+            .then(|| self.reserve_session_for_external_write(id))
+            .transpose()?;
         let mut session = self.load_session(id)?;
         if session.metadata.title == title {
             return Ok(session.metadata);
@@ -3058,6 +3080,7 @@ impl SessionManager {
 
     fn remove_session(&self, id: &str, removal: SessionRemoval) -> std::io::Result<()> {
         let path = self.validated_session_path(id)?;
+        let _lease = self.reserve_session_for_external_write(id)?;
         // Older ordinary snapshots may use a name reserved by the checkpoint
         // directory. Such a name must never address its shared legacy files.
         let checkpoint = self.validated_checkpoint_path(id).ok();
@@ -3083,13 +3106,25 @@ impl SessionManager {
                 Err(error) => return Err(error),
             }
         }
-        if matches!(removal, SessionRemoval::Retention)
-            && (has_recovery || legacy_origin.is_err())
-            && !already_deleted
-        {
+        if matches!(removal, SessionRemoval::Retention) && !already_deleted {
+            // Retention never acts on an uncertain origin. An unreadable
+            // legacy checkpoint may be this id's crash recovery, and a
+            // damaged recovery leaves the ordinary snapshot as the only
+            // readable copy of the conversation. Fail closed: keep every byte
+            // and let the caller skip this record.
+            if let Err(error) = &legacy_origin {
+                return Err(io::Error::new(
+                    error.kind(),
+                    format!(
+                        "retention left session '{}' untouched: the legacy checkpoint's origin is unreadable ({error})",
+                        id.trim()
+                    ),
+                ));
+            }
+        }
+        if matches!(removal, SessionRemoval::Retention) && has_recovery && !already_deleted {
             // Retention owns the ordinary snapshot, not crash recovery. Keep
             // the origin and its accounting/evidence writable for resume.
-            // An unreadable legacy origin cannot justify retiring any id.
             match fs::remove_file(&path) {
                 Ok(()) => {}
                 Err(error) if error.kind() == io::ErrorKind::NotFound => {}
@@ -3301,6 +3336,7 @@ impl SessionManager {
         let result = if listed_path == canonical {
             self.remove_session(id, SessionRemoval::Retention)
         } else {
+            let _lease = self.reserve_session_for_external_write(id)?;
             fs::remove_file(listed_path)
         };
         match result {
@@ -4816,6 +4852,174 @@ mod tests {
         }
     }
 
+    #[test]
+    fn external_writers_are_refused_while_another_process_holds_the_live_lease() {
+        let tmp = tempdir().expect("tempdir");
+        let manager = SessionManager::new(tmp.path().join("sessions")).expect("manager");
+        let id = "77777777-7777-4777-8777-777777777777";
+        save_late_usage_test_session(&manager, id);
+        let snapshot = manager.validated_session_path(id).expect("snapshot path");
+        let original = fs::read(&snapshot).expect("original bytes");
+        // Another process's TUI holds the lease; this process's in-memory
+        // registry knows nothing about it.
+        let lease = manager.hold_live_lease_elsewhere(id);
+        assert!(!is_live_session(id));
+
+        let rename = manager
+            .rename_session(id, "Renamed elsewhere", SessionMutator::External)
+            .expect_err("a rename would be reverted by the owner's next autosave");
+        assert_eq!(rename.kind(), io::ErrorKind::ResourceBusy);
+        let archive = manager
+            .set_session_archived(id, true, SessionMutator::External)
+            .expect_err("an archive would be reverted by the owner's next autosave");
+        assert_eq!(archive.kind(), io::ErrorKind::ResourceBusy);
+        assert_eq!(fs::read(&snapshot).expect("bytes"), original);
+
+        drop(lease);
+        manager
+            .rename_session(id, "Renamed after close", SessionMutator::External)
+            .expect("an unowned session accepts external writes");
+    }
+
+    #[test]
+    fn external_mutations_keep_the_live_lease_until_the_blocked_write_finishes() {
+        let _env = crate::test_support::lock_test_env();
+        let mut failures = Vec::new();
+        for operation in ["rename", "archive", "delete"] {
+            let tmp = tempdir().expect("tempdir");
+            let manager = std::sync::Arc::new(
+                SessionManager::new(tmp.path().join("sessions")).expect("manager"),
+            );
+            let id = "88888888-8888-4888-8888-888888888888";
+            save_late_usage_test_session(&manager, id);
+            // Hold the actual persistence lock so the shipping mutation stays
+            // between admission and commit while another surface tries attach.
+            let (_, lock_path) = manager.ensure_late_usage_paths(id).expect("paths");
+            let mut lock = fd_lock::RwLock::new(open_private_lock_file(&lock_path).unwrap());
+            let write_guard = lock.write().expect("hold persistence lock");
+            let (entered_tx, entered_rx) = std::sync::mpsc::channel();
+            let worker_manager = manager.clone();
+            let ticket = crate::test_support::env_scope_ticket();
+            let worker = std::thread::spawn(move || {
+                let _membership = crate::test_support::join_env_scope(ticket);
+                entered_tx.send(()).unwrap();
+                match operation {
+                    "rename" => worker_manager
+                        .rename_session(id, "Atomic rename", SessionMutator::External)
+                        .map(|_| ()),
+                    "archive" => worker_manager
+                        .set_session_archived(id, true, SessionMutator::External)
+                        .map(|_| ()),
+                    "delete" => worker_manager.delete_session(id),
+                    _ => unreachable!(),
+                }
+            });
+            entered_rx
+                .recv_timeout(std::time::Duration::from_secs(2))
+                .unwrap();
+            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
+            while !manager.is_session_live_anywhere(id)
+                && !worker.is_finished()
+                && std::time::Instant::now() < deadline
+            {
+                std::thread::sleep(std::time::Duration::from_millis(5));
+            }
+            let holds_lease = manager.is_session_live_anywhere(id);
+            let attach_refused = manager
+                .reserve_session_for_attach(id)
+                .is_err_and(|error| error.kind() == io::ErrorKind::ResourceBusy);
+            // Release and join before asserting, even in the fixes-off case;
+            // no failed assertion may leave a blocked mutation behind.
+            drop(write_guard);
+            worker
+                .join()
+                .expect("mutation worker")
+                .expect("mutation succeeds");
+            if !holds_lease {
+                failures.push(format!("{operation}: released its lease before commit"));
+            }
+            if !attach_refused {
+                failures.push(format!("{operation}: admitted a competing session owner"));
+            }
+            assert!(
+                !manager.is_session_live_anywhere(id),
+                "{operation}: leaked its lease"
+            );
+        }
+        assert!(failures.is_empty(), "{failures:?}");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn uncertain_live_lease_refuses_external_mutations_without_changing_history() {
+        let tmp = tempdir().expect("tempdir");
+        let manager = SessionManager::new(tmp.path().join("sessions")).expect("manager");
+        let id = "99999999-9999-4999-8999-999999999999";
+        save_late_usage_test_session(&manager, id);
+        let snapshot = manager.validated_session_path(id).unwrap();
+        let original = fs::read(&snapshot).unwrap();
+        let target = tmp.path().join("untouched");
+        fs::write(&target, b"unrelated bytes").unwrap();
+        let live_path = manager.live_lease_path(id, true).unwrap();
+        std::os::unix::fs::symlink(&target, &live_path).unwrap();
+        assert!(
+            manager.is_session_live_anywhere(id),
+            "uncertain ownership is not free"
+        );
+        assert!(
+            manager
+                .rename_session(id, "Unsafe", SessionMutator::External)
+                .is_err()
+        );
+        assert!(
+            manager
+                .set_session_archived(id, true, SessionMutator::External)
+                .is_err()
+        );
+        assert!(manager.delete_session(id).is_err());
+        assert_eq!(fs::read(snapshot).unwrap(), original);
+        assert_eq!(fs::read(target).unwrap(), b"unrelated bytes");
+    }
+
+    #[test]
+    fn retention_fails_closed_on_an_unreadable_legacy_checkpoint_origin() {
+        let tmp = tempdir().expect("tempdir");
+        let manager = SessionManager::new(tmp.path().join("sessions")).expect("manager");
+        let id = "66666666-6666-4666-8666-666666666666";
+        let mut old = save_late_usage_test_session(&manager, id);
+        old.metadata.updated_at = Utc::now() - chrono::Duration::days(60);
+        manager.save_session(&old).expect("old snapshot");
+        let snapshot = manager.validated_session_path(id).expect("snapshot path");
+        let original = fs::read(&snapshot).expect("original snapshot bytes");
+        // Fault fixture: a legacy checkpoint whose owner cannot be read. It
+        // may be this session's (damaged) recovery.
+        fs::create_dir_all(manager.checkpoints_dir()).expect("checkpoints");
+        let legacy = manager.checkpoints_dir().join(LEGACY_CHECKPOINT_FILE);
+        let corrupt: &[u8] = b"{\"messages\": [truncated";
+        write_atomic(&legacy, corrupt).expect("corrupt legacy checkpoint");
+        assert!(manager.legacy_checkpoint_origin().is_err());
+
+        assert_eq!(
+            manager
+                .prune_sessions_older_than(std::time::Duration::from_secs(24 * 3600))
+                .expect("age prune"),
+            0,
+            "an uncertain origin must not be pruned"
+        );
+        assert_eq!(
+            fs::read(&snapshot).expect("snapshot survives retention"),
+            original,
+            "the ordinary snapshot keeps its original bytes"
+        );
+        assert_eq!(fs::read(&legacy).expect("legacy survives"), corrupt);
+        let (ledger, _) = manager.late_usage_paths(id).expect("ledger paths");
+        assert!(!SessionManager::late_usage_is_deleted(&ledger).expect("not retired"));
+
+        // Explicit deletion remains the user's decision.
+        manager.delete_session(id).expect("explicit delete");
+        assert!(!snapshot.exists());
+    }
+
     #[test]
     fn interrupted_session_deletion_keeps_recovery_incomplete_and_can_finish() {
         let tmp = tempdir().expect("tempdir");
@@ -8770,10 +8974,19 @@ mod tests {
         let manager = SessionManager::new(sessions_dir.clone()).expect("new");
         let checkpoint_dir = sessions_dir.join("checkpoints");
         fs::create_dir_all(&checkpoint_dir).expect("mkdir checkpoints");
-        // Drop a stale-looking JSON inside the checkpoint dir; prune
-        // should leave it alone.
-        let checkpoint_file = checkpoint_dir.join("latest.json");
-        fs::write(&checkpoint_file, "{}").expect("write checkpoint");
+        // Drop a legacy checkpoint inside the checkpoint dir; prune should
+        // leave it alone. It belongs to a readable, unrelated origin: an
+        // unreadable origin makes retention fail closed instead (see
+        // `retention_fails_closed_on_an_unreadable_legacy_checkpoint_origin`).
+        let checkpoint_file = checkpoint_dir.join(LEGACY_CHECKPOINT_FILE);
+        let unrelated = save_late_usage_test_session(&manager, "unrelated-origin");
+        write_atomic(
+            &checkpoint_file,
+            serialize_saved_session(unrelated)
+                .expect("legacy bytes")
+                .as_bytes(),
+        )
+        .expect("write checkpoint");
 
         write_session_with_updated_at(&manager, "stale", Utc::now() - chrono::Duration::days(60));
         let pruned = manager

crates/tui/src/session_resume.rs

@@ -264,28 +264,21 @@ fn candidate_ids(
     sessions: &[crate::session_manager::SessionMetadata],
     workspace: &Path,
 ) -> Vec<String> {
+    // Empty auto-created placeholders are excluded *before* the candidate
+    // cap, the same definition `get_latest_session_for_workspace` applies:
+    // a run of fresh placeholders must not push a real session out of reach.
     let query = crate::session_projection::SessionQuery::default()
         .with_filter(crate::session_manager::SessionListFilter::ActiveOnly)
         .with_sort(crate::session_projection::SessionSortMode::Recent)
         .scoped_to(workspace)
+        .without_empty_auto_created()
         .with_limit(MAX_AUTO_RESUME_CANDIDATES);
     crate::session_projection::select_sessions(sessions, &query)
         .into_iter()
-        .filter(|metadata| !is_empty_placeholder(metadata))
         .map(|metadata| metadata.id.clone())
         .collect()
 }
 
-/// An auto-created, never-used session is not something to resume into.
-/// Mirrors the filter `get_latest_session_for_workspace` applies.
-fn is_empty_placeholder(metadata: &crate::session_manager::SessionMetadata) -> bool {
-    metadata.message_count == 0
-        && metadata
-            .title
-            .trim()
-            .eq_ignore_ascii_case(crate::session_manager::DEFAULT_SESSION_TITLE)
-}
-
 fn plural_sessions(count: usize) -> &'static str {
     if count == 1 { "session" } else { "sessions" }
 }
@@ -362,6 +355,45 @@ mod tests {
         assert_eq!(decision.status_message(), None);
     }
 
+    #[test]
+    fn fresh_placeholders_do_not_push_a_real_session_past_the_candidate_cap() {
+        let fx = fixture();
+        let mut real = saved("real", &fx.workspace, "Real work");
+        real.metadata.updated_at = chrono::Utc::now() - chrono::Duration::hours(1);
+        fx.manager.save_session(&real).expect("save real session");
+        for index in 0..MAX_AUTO_RESUME_CANDIDATES {
+            let mut placeholder = create_saved_session_with_id_and_mode(
+                format!("placeholder-{index}"),
+                &[],
+                "deepseek-chat",
+                &fx.workspace,
+                0,
+                None,
+                Some("agent"),
+            );
+            placeholder.metadata.title = crate::session_manager::DEFAULT_SESSION_TITLE.to_string();
+            placeholder.metadata.updated_at =
+                chrono::Utc::now() - chrono::Duration::seconds(index as i64);
+            fx.manager
+                .save_session(&placeholder)
+                .expect("save placeholder");
+        }
+        let listed = fx.manager.list_sessions().expect("list");
+        assert_eq!(
+            listed
+                .iter()
+                .filter(|session| crate::session_manager::is_empty_auto_created_session(session))
+                .count(),
+            MAX_AUTO_RESUME_CANDIDATES,
+            "fixture keeps a full cap of newer placeholders"
+        );
+
+        let decision =
+            decide_auto_resume(true, &ResumeRequest::default(), &fx.workspace, &fx.manager);
+
+        assert_eq!(decision.session_id(), Some("real"));
+    }
+
     #[test]
     fn enabled_resumes_the_newest_valid_session_for_this_workspace() {
         let fx = fixture();

crates/tui/src/work_graph/runtime.rs

@@ -894,21 +894,33 @@ fn operation_parent(
     session_id: &str,
     source: &str,
 ) -> Result<WorkNodeId, String> {
-    if let Some(parent) = graph
-        .snapshot()
+    let snapshot = graph.snapshot();
+    // Only a step the current plan or To-do projection still lists may adopt
+    // new work. Replacing a plan keeps the dropped steps as history (their
+    // ids are reused if the plan grows back, so they are not retired), but a
+    // step nobody can see must never become an operation's parent.
+    let projected_step = |node: &&WorkNode, state: NodeState| {
+        node.kind == NodeKind::PlanStep
+            && node.state == state
+            && (snapshot.compat.plan_order.contains(&node.id)
+                || snapshot
+                    .compat
+                    .todos
+                    .iter()
+                    .any(|binding| binding.node == node.id))
+    };
+    if let Some(parent) = snapshot
         .nodes
         .iter()
-        .find(|node| node.kind == NodeKind::PlanStep && node.state == NodeState::Active)
+        .find(|node| projected_step(node, NodeState::Active))
         .or_else(|| {
-            graph
-                .snapshot()
+            snapshot
                 .nodes
                 .iter()
-                .find(|node| node.kind == NodeKind::PlanStep && node.state == NodeState::Ready)
+                .find(|node| projected_step(node, NodeState::Ready))
         })
         .or_else(|| {
-            graph
-                .snapshot()
+            snapshot
                 .nodes
                 .iter()
                 .find(|node| node.kind == NodeKind::Objective)
@@ -1826,6 +1838,67 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn replaced_plan_steps_never_parent_new_operations() {
+        let runtime = new_shared_work_runtime(
+            crate::tools::todo::new_shared_todo_list(),
+            crate::tools::plan::new_shared_plan_state(),
+        );
+        let step = |step: &str, status: StepStatus| PlanItemArg {
+            step: step.to_string(),
+            status,
+        };
+        let three = PlanSnapshot {
+            items: vec![
+                step("first", StepStatus::Pending),
+                step("second", StepStatus::Pending),
+                step("third", StepStatus::InProgress),
+            ],
+            ..PlanSnapshot::default()
+        };
+        runtime
+            .apply_plan_update("session", "update_plan", &three)
+            .await
+            .expect("three-step plan");
+        let one = PlanSnapshot {
+            items: vec![step("only", StepStatus::Pending)],
+            ..PlanSnapshot::default()
+        };
+        runtime
+            .apply_plan_update("session", "update_plan", &one)
+            .await
+            .expect("replacement plan");
+
+        let operation = runtime
+            .register_operation(
+                "session",
+                OperationIntent::new(
+                    "shell:after",
+                    "after replacement",
+                    false,
+                    "exec_shell",
+                    "c1",
+                ),
+            )
+            .expect("register operation");
+        let graph = runtime
+            .capture(Some("session"))
+            .expect("capture")
+            .expect("graph")
+            .graph;
+        assert_eq!(graph.compat.plan_order.len(), 1);
+        let parent = graph
+            .edges
+            .iter()
+            .find(|edge| edge.kind == EdgeKind::Contains && edge.to == operation)
+            .map(|edge| edge.from.clone());
+        assert_eq!(
+            parent,
+            Some(graph.compat.plan_order[0].clone()),
+            "an orphaned step from the replaced plan must not adopt new work"
+        );
+    }
+
     #[test]
     fn legacy_restore_stays_pending_until_first_graph_bearing_write() {
         let todos = crate::tools::todo::new_shared_todo_list();