fix(runtime): keep a thread compacted across restart and engine reload

A Runtime thread (desktop and web) that compacted came back
uncompacted whenever its engine was rebuilt: after a restart, an LRU
eviction or a failure-path reload. The CompactionCompleted handler saved
only the summary text, into a delimited section of thread.system_prompt,
and dropped the SessionUpdated that carried the replacement history.
restore_thread_messages then replayed every turn record and SyncSession
appended the summary. The next request held everything compaction had
removed, plus the summary.

Reproduced live with a mock provider (artifacts/runtime-compact-20261019):
three turns, then POST /v1/threads/{id}/compact, then a fourth turn.
Without a restart the fourth request has 7 messages; after a real
restart it had 9, with both dropped assistant replies back.

Now:
- After TurnComplete, a turn in which a compaction completed reads the
  engine's history with get_session_snapshot (the engine is idle and
  starts no turn on its own). It masks tool results with the
  receipt_secrets every Runtime receipt already uses, and stores the
  list once under its SHA-256 as items/<digest>.history (item listings
  read only .json). The write runs in spawn_blocking. The completed
  ContextCompaction item gets metadata.settled_history = {sha256,
  messages}. Errors propagate through the monitor's existing failure
  path.
- RuntimeThreadStore::settled_history finds the newest reference among
  a turn slice, verifies the digest and the length, and fails loudly,
  naming the file, rather than replaying raw turns. history_from_records
  is that history plus the later turns rebuilt from records.
  restore_thread_messages applies it after any bound session prefix.
  Recovery of unbound threads, compacted native forks, fork-at-turn
  prefixes and session export (messages_from_thread_detail) use it, so
  a fork that keeps the compacting turn starts from what the source
  model saw, and one that cuts before it gets the raw kept turns.
- The thread-record carrier is gone: COMPACTION_SUMMARY_BEGIN/END,
  merge_summary_into_prompt, strip_summary_section and their two tests.
  synced_system_prompt strips a legacy carrier from the prompt sent to
  SyncSession whenever the history already holds its checkpoint. Both
  sync points use it, so an old summary can no longer replace a newer
  checkpoint.
- core/engine.rs: the tool-result redaction loop becomes
  redact_tool_result_blocks; the engine method now calls it.

Known limits:
- After a restart the agent-topology checkpoint appears in its
  "restored" form, and with the model-bound masking opt-out on, tool
  output is masked (as rebuilt turns already are).
- Threads compacted before this change keep the old shape until their
  next compaction.
- One history file per compacting turn, never removed, because forks
  share it.
- A turn still running at shutdown comes back uncompacted.

Found by GPT-6 Astra (codex). Design from a read-only Claude workflow:
four mappers, two designs, one judge.

Verification:
- Live, same repro, this source: after restart the fourth request has 7
  messages [system, u1, u2, u3, a3, checkpoint, u4], equal to the
  no-restart run. Two server starts, no errors, one .history file.
- RUST_MIN_STACK=8388608 cargo test -p codewhale-tui --lib -- compact
  fork restore messages_from_thread_detail export redact undo retry
  seed_thread: 838 passed, 1 failed. The failure was a cancelled
  compaction asking for a snapshot; capture is now limited to completed
  compactions. After that: -- compact settled restore_thread: 311
  passed, 0 failed.
- Updated tests: compaction_lifecycle_emits_item_events_with_compaction_counts
  (the mock answers the snapshot; restore equals the newest stored
  history; the record has no system prompt);
  compact_lifecycle_outlives_caller_and_preserves_concurrent_thread_updates
  (answers the snapshot); the runtime_api export tests (new signature).
- cargo clippy -p codewhale-tui --all-targets --all-features with CI's
  flags: clean. cargo fmt --all -- --check: clean.
  scripts/check-blocking-calls-budget.py: 540 sites, within budget.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01NBKbhx19BRbZtvWK8rVKjz

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- In the desktop and web apps, a conversation stays compacted after
+  Codewhale restarts or unloads it. Before, reloading replayed every earlier
+  turn and then added the summary, so the next request carried everything
+  compaction had removed. Each compacting turn now stores the history the
+  model continued from, and forks, `/undo` and session exports start from
+  it as well.
 - When a provider stream fails before any content and Codewhale quietly
   resends the request, the tokens the failed attempt reported are now added
   to the session, goal and sub-agent totals instead of being merged away.

crates/tui/CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- In the desktop and web apps, a conversation stays compacted after
+  Codewhale restarts or unloads it. Before, reloading replayed every earlier
+  turn and then added the summary, so the next request carried everything
+  compaction had removed. Each compacting turn now stores the history the
+  model continued from, and forks, `/undo` and session exports start from
+  it as well.
 - When a provider stream fails before any content and Codewhale quietly
   resends the request, the tokens the failed attempt reported are now added
   to the session, goal and sub-agent totals instead of being merged away.

crates/tui/src/core/engine.rs

@@ -917,6 +917,29 @@ enum EngineHostSetup {
     Rlm(rlm_host::RlmHostSetup),
 }
 
+/// Rewrite the text of every tool result in `message`, including text
+/// content blocks, with `redact`.
+pub(crate) fn redact_tool_result_blocks(message: &mut Message, redact: impl Fn(&str) -> String) {
+    for block in &mut message.content {
+        let ContentBlock::ToolResult {
+            content,
+            content_blocks,
+            ..
+        } = block
+        else {
+            continue;
+        };
+        *content = redact(content);
+        for value in content_blocks.iter_mut().flatten() {
+            if value.get("type").and_then(serde_json::Value::as_str) == Some("text")
+                && let Some(serde_json::Value::String(text)) = value.get_mut("text")
+            {
+                *text = redact(text);
+            }
+        }
+    }
+}
+
 pub(crate) fn compaction_checkpoint_for_restore(
     messages: &[codewhale_models::Message],
     system_prompt: Option<codewhale_models::SystemPrompt>,
@@ -4661,24 +4684,7 @@ impl Engine {
     /// written to disk live. Honors the confirmed `[redaction] model_bound`
     /// opt-out the same way the request boundary does.
     fn redact_tool_results_for_transcript(&self, message: &mut Message) {
-        for block in &mut message.content {
-            let ContentBlock::ToolResult {
-                content,
-                content_blocks,
-                ..
-            } = block
-            else {
-                continue;
-            };
-            *content = self.redact_tool_output_for_transcript(content);
-            for value in content_blocks.iter_mut().flatten() {
-                if value.get("type").and_then(serde_json::Value::as_str) == Some("text")
-                    && let Some(serde_json::Value::String(text)) = value.get_mut("text")
-                {
-                    *text = self.redact_tool_output_for_transcript(text);
-                }
-            }
-        }
+        redact_tool_result_blocks(message, |text| self.redact_tool_output_for_transcript(text));
     }
 
     fn redact_tool_output_for_transcript(&self, text: &str) -> String {

crates/tui/src/runtime_api/sessions.rs

@@ -527,9 +527,10 @@ pub(super) async fn create_session_from_thread(
         });
     }
 
-    let messages = messages_from_thread_detail(&detail).map_err(|error| {
-        ApiError::internal(format!("Failed to reconstruct thread history: {error}"))
-    })?;
+    let messages =
+        messages_from_thread_detail(&state.runtime_threads, &detail).map_err(|error| {
+            ApiError::internal(format!("Failed to reconstruct thread history: {error}"))
+        })?;
     if messages.is_empty() {
         return Err(ApiError::bad_request(format!(
             "Thread {thread_id} has no user or assistant messages to save"
@@ -685,7 +686,10 @@ pub(super) fn thread_detail_has_live_work(detail: &ThreadDetail) -> bool {
     })
 }
 
-pub(super) fn messages_from_thread_detail(detail: &ThreadDetail) -> anyhow::Result<Vec<Message>> {
+pub(super) fn messages_from_thread_detail(
+    runtime: &RuntimeThreadManager,
+    detail: &ThreadDetail,
+) -> anyhow::Result<Vec<Message>> {
     let mut items_by_turn = HashMap::new();
     for item in &detail.items {
         items_by_turn
@@ -693,7 +697,7 @@ pub(super) fn messages_from_thread_detail(detail: &ThreadDetail) -> anyhow::Resu
             .or_insert_with(Vec::new)
             .push(item.clone());
     }
-    RuntimeThreadManager::reconstruct_messages_from_turns_with(&detail.turns, &items_by_turn)
+    runtime.history_from_records(&detail.turns, &items_by_turn)
 }
 
 /// Merge the thread's authoritative cost into a session about to be saved.

crates/tui/src/runtime_api/tests.rs

@@ -592,7 +592,19 @@ fn messages_from_thread_detail_batches_tool_results() {
         approval_grants: Vec::new(),
     };
 
-    let messages = messages_from_thread_detail(&detail).expect("rebuild thread history");
+    let mut items_by_turn = std::collections::HashMap::new();
+    for item in &detail.items {
+        items_by_turn
+            .entry(item.turn_id.clone())
+            .or_insert_with(Vec::new)
+            .push(item.clone());
+    }
+    let messages =
+        crate::runtime_threads::RuntimeThreadManager::reconstruct_messages_from_turns_with(
+            &detail.turns,
+            &items_by_turn,
+        )
+        .expect("rebuild thread history");
     let roles = messages
         .iter()
         .map(|message| message.role.as_str())
@@ -25769,7 +25781,7 @@ async fn session_export_keeps_execution_identity_and_rejects_invalid_item_correl
         .seed_thread_from_messages(&thread.id, &original)
         .await?;
     let detail = runtime.get_thread_detail(&thread.id).await?;
-    assert_eq!(messages_from_thread_detail(&detail)?, original);
+    assert_eq!(messages_from_thread_detail(&runtime, &detail)?, original);
 
     // A completed live item contains both sides. It must not become only a
     // result when the same thread is saved through the API's projection.
@@ -25801,7 +25813,7 @@ async fn session_export_keeps_execution_identity_and_rejects_invalid_item_correl
     for turn in &mut coalesced.turns {
         turn.item_ids.retain(|id| id != &result_item.id);
     }
-    assert_eq!(messages_from_thread_detail(&coalesced)?, original);
+    assert_eq!(messages_from_thread_detail(&runtime, &coalesced)?, original);
 
     let client = crate::tls::reqwest_client();
     let exported = client

crates/tui/src/runtime_threads.rs

@@ -242,53 +242,16 @@ async fn coalesce_stream_delta(
     }
 }
 
-/// Sentinel delimiters wrapping the compaction summary section persisted in a
-/// thread record's `system_prompt`. The section carries the engine-rendered
-/// summary (which contains the compaction summary marker). On reload,
-/// `SyncSession` migrates that carrier into one ordinary history checkpoint
-/// and strips it from the model's standing system prompt. Delimiters make
-/// replacement idempotent: each completed
-/// compaction swaps the section in place instead of stacking duplicates.
-/// External `PATCH /v1/threads/{id}` callers that rewrite `system_prompt`
-/// should preserve this section verbatim or the summary is lost on reload.
-const COMPACTION_SUMMARY_BEGIN: &str = "<!-- compaction-summary:begin -->";
-const COMPACTION_SUMMARY_END: &str = "<!-- compaction-summary:end -->";
-
-/// Merge a rendered compaction summary into a thread record's system prompt,
-/// replacing any previously persisted summary section.
-fn merge_summary_into_prompt(base: Option<&str>, summary_text: &str) -> String {
-    let stripped = base.map(strip_summary_section).unwrap_or_default();
-    let mut out = stripped.trim_end().to_string();
-    if !out.is_empty() {
-        out.push_str("\n\n");
+/// The system prompt a sync installs. A history that already carries its
+/// compaction checkpoint outranks a legacy summary section persisted in the
+/// thread record's prompt, which would otherwise replace that checkpoint.
+fn synced_system_prompt(prompt: Option<&str>, messages: &[Message]) -> Option<SystemPrompt> {
+    let prompt = prompt.map(|text| SystemPrompt::Text(text.to_string()));
+    if crate::compaction::history_is_post_provenance(messages) {
+        crate::compaction::strip_compaction_summaries(prompt.as_ref())
+    } else {
+        prompt
     }
-    out.push_str(COMPACTION_SUMMARY_BEGIN);
-    out.push('\n');
-    out.push_str(summary_text.trim());
-    out.push('\n');
-    out.push_str(COMPACTION_SUMMARY_END);
-    out
-}
-
-/// Remove a previously persisted compaction summary section, if present.
-fn strip_summary_section(base: &str) -> String {
-    let Some(start) = base.find(COMPACTION_SUMMARY_BEGIN) else {
-        return base.to_string();
-    };
-    let end = base[start..]
-        .find(COMPACTION_SUMMARY_END)
-        .map(|rel| start + rel + COMPACTION_SUMMARY_END.len());
-    let mut out = base[..start].trim_end().to_string();
-    if let Some(end) = end {
-        let tail = base[end..].trim_start();
-        if !tail.is_empty() {
-            if !out.is_empty() {
-                out.push_str("\n\n");
-            }
-            out.push_str(tail);
-        }
-    }
-    out
 }
 
 /// A fresh durable record id: `<prefix>_<32 hex>`. Records live in flat
@@ -994,9 +957,8 @@ fn projected_user_text(message: &Message) -> Option<String> {
     // An engine-owned compaction checkpoint is not a turn's prompt. No turn
     // record can reproduce one — the compaction ran in the engine, between
     // records — so a document that carries it would read as drifted from the
-    // records it was rebuilt from. Every fork of a compacted source inherits
-    // the summary in its system prompt, and the engine installs the checkpoint
-    // into the synced history on the fork's first load, so leaving it in took
+    // records it was rebuilt from. A fork of a compacted source starts from
+    // the settled history that carries the checkpoint, so leaving it in took
     // the cut away from the fork itself: `/undo`, retry and fork-again each
     // refused with "cannot identify an exact saved-history boundary".
     if crate::compaction::is_wire_compaction_checkpoint_message(message) {
@@ -1174,9 +1136,9 @@ fn session_recovery_projection(messages: &[Message]) -> Vec<Value> {
 /// the kept turns hands every reader of that document — the session view,
 /// `restore_thread_messages` — a run of prompts with the agent's
 /// output missing, because nothing is left for it to rebuild the turns from.
-/// The turn store is not rewritten by compaction, so the records can: they are
-/// what the fork rebuilds the kept exchanges from
-/// (`reconstruct_messages_from_turns_with`).
+/// The records can: a compacting turn stores the history its engine settled
+/// on, and the fork rebuilds the kept exchanges from the newest one it keeps
+/// plus the raw turns after it (`RuntimeThreadStore::history_from_records`).
 ///
 /// The compaction names the turn it ran for, so the records answer the
 /// question directly. Any compaction in the thread is enough: the turns before
@@ -3178,6 +3140,116 @@ impl RuntimeThreadStore {
         Ok(())
     }
 
+    /// The engine history a compacting turn settled on, stored once under
+    /// its digest beside the items (listings read only `.json`). Forks share
+    /// it by reference, so it is never removed.
+    fn settled_history_path(&self, digest: &str) -> Result<PathBuf> {
+        Self::record_path(&self.items_dir, digest, "history", "settled history digest")
+    }
+
+    pub(crate) fn save_settled_history(&self, messages: &[Message]) -> Result<Value> {
+        let bytes = serde_json::to_vec(messages)?;
+        let digest = hex_digest(Sha256::digest(&bytes));
+        let path = self.settled_history_path(&digest)?;
+        reject_symlinked_store_file(&path)?;
+        crate::utils::write_atomic(&path, &bytes)
+            .with_context(|| format!("Failed to write settled history {}", path.display()))?;
+        Ok(json!({ "sha256": digest, "messages": messages.len() }))
+    }
+
+    /// The newest settled history among `turns`, and how many of them it
+    /// covers. A reference whose file is missing, altered or unreadable is an
+    /// error: replaying the raw turns instead would silently undo compaction.
+    fn settled_history(
+        &self,
+        turns: &[TurnRecord],
+        items_by_turn: &HashMap<String, Vec<TurnItemRecord>>,
+    ) -> Result<Option<(usize, Vec<Message>)>> {
+        for (index, turn) in turns.iter().enumerate().rev() {
+            let Some((item, reference)) = items_by_turn
+                .get(&turn.id)
+                .into_iter()
+                .flatten()
+                .rev()
+                .filter(|item| item.kind == TurnItemKind::ContextCompaction)
+                .find_map(|item| {
+                    let reference = item.metadata.as_ref()?.get("settled_history")?;
+                    Some((item, reference))
+                })
+            else {
+                continue;
+            };
+            let digest = reference
+                .get("sha256")
+                .and_then(Value::as_str)
+                .filter(|digest| {
+                    digest.len() == 64
+                        && digest
+                            .bytes()
+                            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+                })
+                .with_context(|| {
+                    format!(
+                        "Compaction item {} on turn {} names an invalid settled history",
+                        item.id, turn.id
+                    )
+                })?;
+            let expected = reference
+                .get("messages")
+                .and_then(Value::as_u64)
+                .and_then(|count| usize::try_from(count).ok())
+                .with_context(|| {
+                    format!(
+                        "Compaction item {} on turn {} names no settled history length",
+                        item.id, turn.id
+                    )
+                })?;
+            let path = self.settled_history_path(digest)?;
+            reject_symlinked_store_file(&path)?;
+            let bytes = fs::read(&path).with_context(|| {
+                format!(
+                    "Settled history for turn {} is unreadable: {}",
+                    turn.id,
+                    path.display()
+                )
+            })?;
+            anyhow::ensure!(
+                hex_digest(Sha256::digest(&bytes)) == digest,
+                "Settled history {} for turn {} does not match its digest",
+                path.display(),
+                turn.id
+            );
+            let messages: Vec<Message> = serde_json::from_slice(&bytes)
+                .with_context(|| format!("Failed to parse settled history {}", path.display()))?;
+            anyhow::ensure!(
+                messages.len() == expected,
+                "Settled history {} for turn {} holds {} messages, not {expected}",
+                path.display(),
+                turn.id,
+                messages.len()
+            );
+            return Ok(Some((index + 1, messages)));
+        }
+        Ok(None)
+    }
+
+    /// The model-visible history the records hold: the newest settled
+    /// history, then every later turn rebuilt from its items.
+    pub(crate) fn history_from_records(
+        &self,
+        turns: &[TurnRecord],
+        items_by_turn: &HashMap<String, Vec<TurnItemRecord>>,
+    ) -> Result<Vec<Message>> {
+        let (covered, mut messages) = self
+            .settled_history(turns, items_by_turn)?
+            .unwrap_or_default();
+        messages.extend(RuntimeThreadManager::reconstruct_messages_from_turns_with(
+            &turns[covered..],
+            items_by_turn,
+        )?);
+        Ok(messages)
+    }
+
     /// 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
@@ -6878,8 +6950,7 @@ impl HeldRuntimeStore {
             }
             let turn_ids: Vec<String> = turns.iter().map(|turn| turn.id.clone()).collect();
             let items = store.list_items_for_turns_map(&turn_ids)?;
-            let messages =
-                RuntimeThreadManager::reconstruct_messages_from_turns_with(&turns, &items)?;
+            let messages = store.history_from_records(&turns, &items)?;
             if messages.is_empty() {
                 continue;
             }
@@ -12012,7 +12083,8 @@ impl RuntimeThreadManager {
     ) -> Result<()> {
         let thread = self.get_thread(thread_id).await?;
         let engine = self.ensure_engine_loaded(&thread).await?;
-        let system_prompt = thread.system_prompt.clone().map(SystemPrompt::Text);
+        let system_prompt =
+            synced_system_prompt(thread.system_prompt.as_deref(), &session.messages);
         let expected = {
             let checkpoint = crate::core::engine::compaction_checkpoint_for_restore(
                 &session.messages,
@@ -12027,8 +12099,8 @@ impl RuntimeThreadManager {
             .send(Op::SyncSession {
                 session_id: Some(thread.id.clone()),
                 messages: session.messages.clone(),
+                system_prompt_override: system_prompt.is_some(),
                 system_prompt,
-                system_prompt_override: thread.system_prompt.is_some(),
                 model: thread.model.clone(),
                 workspace: thread.workspace.clone(),
                 mode: RuntimePolicyProjection::from_persisted(
@@ -12686,7 +12758,8 @@ impl RuntimeThreadManager {
             .await?;
         let fork_prefix = if saved_transcript_is_compacted(&source_turns, &items_by_turn) {
             (
-                Self::reconstruct_messages_from_turns_with(&source_turns, &items_by_turn)?,
+                self.store
+                    .history_from_records(&source_turns, &items_by_turn)?,
                 source_turns.len(),
             )
         } else {
@@ -13228,10 +13301,8 @@ impl RuntimeThreadManager {
             let (prefix, covered_turns) = match retained_messages {
                 Some(retained) => (messages[..retained].to_vec(), kept_turns),
                 None => (
-                    Self::reconstruct_messages_from_turns_with(
-                        &source_turns[..cutoff_turn_idx],
-                        &items_by_turn,
-                    )?,
+                    self.store
+                        .history_from_records(&source_turns[..cutoff_turn_idx], &items_by_turn)?,
                     cutoff_turn_idx,
                 ),
             };
@@ -13295,7 +13366,7 @@ impl RuntimeThreadManager {
         }
         if fork_prefix.is_none() {
             fork_prefix = Some((
-                Self::reconstruct_messages_from_turns_with(
+                self.store.history_from_records(
                     &source_turns[..cutoff_turn_idx],
                     &self.prepared_items_for_turns(&source_turns[..cutoff_turn_idx])?,
                 )?,
@@ -16223,10 +16294,8 @@ impl RuntimeThreadManager {
                 ));
             }
 
-            let sys_prompt = thread
-                .system_prompt
-                .as_ref()
-                .map(|s| SystemPrompt::Text(s.clone()));
+            let sys_prompt =
+                synced_system_prompt(thread.system_prompt.as_deref(), &session_messages);
             if !session_messages.is_empty() || sys_prompt.is_some() {
                 engine
                     .send(Op::SyncSession {
@@ -16234,9 +16303,9 @@ impl RuntimeThreadManager {
                         // re-sync of the same conversation, never a
                         // conversation boundary.
                         session_id: Some(thread.id.clone()),
+                        system_prompt_override: sys_prompt.is_some(),
                         messages: session_messages,
                         system_prompt: sys_prompt,
-                        system_prompt_override: thread.system_prompt.is_some(),
                         model: route_model.clone(),
                         workspace: thread.workspace.clone(),
                         mode: RuntimePolicyProjection::from_persisted(
@@ -16619,7 +16688,7 @@ impl RuntimeThreadManager {
 
     fn restore_thread_messages(&self, thread: &ThreadRecord) -> Result<Vec<Message>> {
         let turns = self.store.list_turns_for_thread(&thread.id)?;
-        let (mut messages, covered) = match self.saved_session_prefix(thread, &turns) {
+        let (mut messages, mut covered) = match self.saved_session_prefix(thread, &turns) {
             Ok(prefix) => prefix.unwrap_or_default(),
             Err(error) => {
                 let Some(stale) = error.downcast_ref::<StaleSessionBinding>() else {
@@ -16632,7 +16701,18 @@ impl RuntimeThreadManager {
                 (Vec::new(), 0)
             }
         };
-        messages.extend(self.reconstruct_messages_from_turns(&turns[covered..])?);
+        let items_by_turn = self.prepared_items_for_turns(&turns[covered..])?;
+        if let Some((settled, history)) = self
+            .store
+            .settled_history(&turns[covered..], &items_by_turn)?
+        {
+            messages = history;
+            covered += settled;
+        }
+        messages.extend(Self::reconstruct_messages_from_turns_with(
+            &turns[covered..],
+            &items_by_turn,
+        )?);
         Ok(messages)
     }
 
@@ -16865,6 +16945,7 @@ impl RuntimeThreadManager {
         Ok((messages, covered))
     }
 
+    #[cfg(test)]
     fn reconstruct_messages_from_turns(&self, turns: &[TurnRecord]) -> Result<Vec<Message>> {
         // One batch read for the whole set. `list_items_for_turn` scans the
         // store's entire items directory to answer for a single turn, so a
@@ -16874,6 +16955,14 @@ impl RuntimeThreadManager {
         Self::reconstruct_messages_from_turns_with(turns, &items_by_turn)
     }
 
+    pub(crate) fn history_from_records(
+        &self,
+        turns: &[TurnRecord],
+        items_by_turn: &HashMap<String, Vec<TurnItemRecord>>,
+    ) -> Result<Vec<Message>> {
+        self.store.history_from_records(turns, items_by_turn)
+    }
+
     pub(crate) fn reconstruct_messages_from_turns_with(
         turns: &[TurnRecord],
         items_by_turn: &HashMap<String, Vec<TurnItemRecord>>,
@@ -17207,6 +17296,8 @@ impl RuntimeThreadManager {
         let mut current_reasoning_item: Option<TurnItemRecord> = None;
         let mut tool_items: HashMap<String, String> = HashMap::new();
         let mut compaction_items: HashMap<String, String> = HashMap::new();
+        let mut completed_compaction_item: Option<String> = None;
+        let mut terminal_received = false;
         let mut turn_usage: Option<Usage> = None;
         let mut turn_effective_route_usage: Option<Usage> = None;
         let mut turn_routed_usage_dropped_records = 0_u64;
@@ -18022,36 +18113,11 @@ impl RuntimeThreadManager {
                     message,
                     messages_before,
                     messages_after,
-                    summary_prompt,
+                    summary_prompt: _,
                     post_input_tokens: _,
                 } => {
-                    // Persist the summary in the legacy thread-record carrier
-                    // so reloads survive LRU eviction/restart. SyncSession
-                    // migrates it into one ordinary history checkpoint and
-                    // strips the carrier from the standing system prompt.
-                    if let Some(summary) =
-                        summary_prompt.as_deref().filter(|s| !s.trim().is_empty())
-                    {
-                        let persist_summary = (|| -> Result<()> {
-                            let _thread_mutation = self.store.thread_mutation.lock();
-                            let mut thread = self.store.load_thread(&thread_id)?;
-                            let merged =
-                                merge_summary_into_prompt(thread.system_prompt.as_deref(), summary);
-                            if thread.system_prompt.as_deref() != Some(merged.as_str()) {
-                                thread.system_prompt = Some(merged);
-                                thread.updated_at = Utc::now();
-                                self.store.save_thread(&thread)?;
-                            }
-                            Ok(())
-                        })();
-                        if let Err(e) = persist_summary {
-                            tracing::warn!(
-                                thread_id = %thread_id,
-                                "Failed to persist compaction summary to thread record: {e}"
-                            );
-                        }
-                    }
                     if let Some(item_id) = compaction_items.remove(&id) {
+                        completed_compaction_item = Some(item_id.clone());
                         let mut item = self.store.load_item(&item_id)?;
                         item.status = TurnItemLifecycleStatus::Completed;
                         item.summary = summarize_text(&message, SUMMARY_LIMIT);
@@ -19016,6 +19082,7 @@ impl RuntimeThreadManager {
                     if let Some(err) = error {
                         turn_error = Some(err);
                     }
+                    terminal_received = true;
                     break;
                 }
                 EngineEvent::GoalUpdated { snapshot } => {
@@ -19173,6 +19240,31 @@ impl RuntimeThreadManager {
         self.settle_dynamic_tools_for_terminal_turn(&thread_id, &turn_id)
             .await?;
 
+        if terminal_received
+            && !self.cancel_token.is_cancelled()
+            && let Some(item_id) = completed_compaction_item.as_deref()
+        {
+            let mut messages = engine.get_session_snapshot().await?.messages;
+            for message in &mut messages {
+                crate::core::engine::redact_tool_result_blocks(message, |text| {
+                    crate::client::redact_model_bound_text(text, &receipt_secrets)
+                });
+            }
+            let store = self.store.clone();
+            let reference =
+                tokio::task::spawn_blocking(move || store.save_settled_history(&messages))
+                    .await
+                    .context("settled history write task failed")??;
+            let mut item = self.store.load_item(item_id)?;
+            let mut metadata = item.metadata.take().unwrap_or_else(|| json!({}));
+            let Some(fields) = metadata.as_object_mut() else {
+                bail!("Compaction item {item_id} has non-object metadata");
+            };
+            fields.insert("settled_history".to_string(), reference);
+            item.metadata = Some(metadata);
+            self.store.save_item(&item)?;
+        }
+
         // Publish the terminal projection as one snapshot boundary. The
         // duplicate scan is offloaded while this guard is held, so public
         // readers cannot observe a terminal record before its receipt and

crates/tui/src/runtime_threads/tests.rs

@@ -4357,6 +4357,16 @@ async fn compact_lifecycle_outlives_caller_and_preserves_concurrent_thread_updat
             base_url: None,
         })
         .await?;
+    loop {
+        match tokio::time::timeout(Duration::from_secs(5), harness.rx_op.recv()).await? {
+            Some(Op::GetSessionSnapshot { tx }) => {
+                answer_session_snapshot(&tx, "settled");
+                break;
+            }
+            Some(_) => {}
+            None => bail!("engine mailbox closed before the settled-history snapshot"),
+        }
+    }
     let terminal = wait_for_terminal_turn(&manager, &turn_id).await?;
     assert_eq!(terminal.status, RuntimeTurnStatus::Completed);
     assert_eq!(manager.active_turn_flags(&thread.id, &turn_id).await, None);
@@ -18287,6 +18297,31 @@ async fn failed_turn_cancels_pending_user_input_and_clears_snapshot() -> Result<
     Ok(())
 }
 
+fn answer_session_snapshot(
+    tx: &Arc<
+        std::sync::Mutex<Option<tokio::sync::oneshot::Sender<crate::core::ops::SessionSnapshot>>>,
+    >,
+    text: &str,
+) {
+    let messages: Vec<Message> = serde_json::from_value(json!([
+        {"role": "user", "content": [{"type": "text", "text": text}]}
+    ]))
+    .expect("snapshot message");
+    if let Some(tx) = tx.lock().expect("snapshot sender").take() {
+        let _ = tx.send(crate::core::ops::SessionSnapshot {
+            session_id: String::new(),
+            messages,
+            total_tokens: 0,
+            model: String::new(),
+            model_provider: String::new(),
+            model_provider_id: None,
+            workspace: PathBuf::new(),
+            system_prompt: None,
+            mode: String::new(),
+        });
+    }
+}
+
 #[tokio::test]
 async fn compaction_lifecycle_emits_item_events_with_compaction_counts() -> Result<()> {
     let manager = test_manager(test_runtime_dir())?;
@@ -18310,8 +18345,20 @@ async fn compaction_lifecycle_emits_item_events_with_compaction_counts() -> Resu
     let tx_event = harness.tx_event;
     tokio::spawn(async move {
         let mut op_count = 0usize;
+        let mut snapshots = 0usize;
         while let Some(op) = rx_op.recv().await {
             match op {
+                Op::GetSessionSnapshot { tx } => {
+                    snapshots = snapshots.saturating_add(1);
+                    answer_session_snapshot(
+                        &tx,
+                        if snapshots == 1 {
+                            "auto settled"
+                        } else {
+                            "manual settled"
+                        },
+                    );
+                }
                 Op::SendMessage(TurnSpec { .. }) => {
                     op_count = op_count.saturating_add(1);
                     let _ = tx_event
@@ -18411,7 +18458,7 @@ async fn compaction_lifecycle_emits_item_events_with_compaction_counts() -> Resu
                 }
                 _ => {}
             }
-            if op_count >= 2 {
+            if op_count >= 2 && snapshots >= 2 {
                 break;
             }
         }
@@ -18482,15 +18529,16 @@ async fn compaction_lifecycle_emits_item_events_with_compaction_counts() -> Resu
             && ev.payload.get("messages_after").and_then(Value::as_u64) == Some(2)
     }));
 
-    // The manual compact carried a summary_prompt → it must be persisted into
-    // the thread record so engine reloads restore it. The auto compact carried
-    // None → exactly one summary section, from the manual pass.
+    // Each compacting turn stores the history the engine settled on; a
+    // reload restores the newest one instead of replaying the turns it
+    // replaced, and the thread record carries no summary of its own.
     let record = manager.get_thread(&thread.id).await?;
-    let record_prompt = record.system_prompt.expect("record keeps a system prompt");
-    assert!(record_prompt.contains(COMPACTION_SUMMARY_BEGIN));
-    assert!(record_prompt.contains("Conversation Summary (Auto-Generated)"));
-    assert!(record_prompt.contains("key facts."));
-    assert_eq!(record_prompt.matches(COMPACTION_SUMMARY_BEGIN).count(), 1);
+    assert_eq!(record.system_prompt, None);
+    let restored = manager.restore_thread_messages(&record)?;
+    let expected: Vec<Message> = serde_json::from_value(json!([
+        {"role": "user", "content": [{"type": "text", "text": "manual settled"}]}
+    ]))?;
+    assert_eq!(restored, expected);
     Ok(())
 }
 
@@ -19169,73 +19217,6 @@ async fn fork_at_user_message_does_not_mutate_source() -> Result<()> {
     Ok(())
 }
 
-// ── compaction summary persistence (merge_summary_into_prompt) ──
-
-#[test]
-fn summary_merge_scenario() {
-    // Scenario consolidation of: summary_merge_appends_section_to_base_prompt, summary_merge_replaces_existing_section_idempotently, summary_merge_handles_missing_base
-    // from summary_merge_appends_section_to_base_prompt
-    {
-        let merged = merge_summary_into_prompt(
-            Some("You are a helpful agent."),
-            "## 📋 Conversation Summary (Auto-Generated)\n\nUser prefers lists.",
-        );
-        assert!(merged.starts_with("You are a helpful agent."));
-        assert!(merged.contains(COMPACTION_SUMMARY_BEGIN));
-        assert!(merged.contains("User prefers lists."));
-        assert!(merged.ends_with(COMPACTION_SUMMARY_END));
-        // Reload restore keys on the marker: SyncSession migrates this carrier
-        // into one ordinary history checkpoint before provider dispatch.
-        assert!(merged.contains("Conversation Summary (Auto-Generated)"));
-    }
-    // from summary_merge_replaces_existing_section_idempotently
-    {
-        let first = merge_summary_into_prompt(Some("Base prompt."), "summary v1");
-        let second = merge_summary_into_prompt(Some(&first), "summary v2");
-        assert!(second.contains("summary v2"));
-        assert!(!second.contains("summary v1"));
-        assert_eq!(
-            second.matches(COMPACTION_SUMMARY_BEGIN).count(),
-            1,
-            "repeated compactions must swap the section, not stack duplicates"
-        );
-        assert!(second.starts_with("Base prompt."));
-    }
-    // from summary_merge_handles_missing_base
-    {
-        let merged = merge_summary_into_prompt(None, "only summary");
-        assert!(merged.starts_with(COMPACTION_SUMMARY_BEGIN));
-        assert!(merged.contains("only summary"));
-        let empty_base = merge_summary_into_prompt(Some(""), "only summary");
-        assert!(empty_base.starts_with(COMPACTION_SUMMARY_BEGIN));
-    }
-}
-
-#[test]
-fn summary_strip_scenario() {
-    // Scenario consolidation of: summary_strip_preserves_text_after_section, summary_strip_handles_missing_end_sentinel
-    // from summary_strip_preserves_text_after_section
-    {
-        let with_tail = format!(
-            "Base.\n\n{COMPACTION_SUMMARY_BEGIN}\nold summary\n{COMPACTION_SUMMARY_END}\n\nTrailing rules."
-        );
-        let stripped = strip_summary_section(&with_tail);
-        assert!(stripped.contains("Base."));
-        assert!(stripped.contains("Trailing rules."));
-        assert!(!stripped.contains("old summary"));
-        // Re-merge keeps the tail intact.
-        let merged = merge_summary_into_prompt(Some(&with_tail), "new summary");
-        assert!(merged.contains("Trailing rules."));
-        assert!(merged.contains("new summary"));
-    }
-    // from summary_strip_handles_missing_end_sentinel
-    {
-        let broken = format!("Base.\n\n{COMPACTION_SUMMARY_BEGIN}\ntruncated…");
-        let stripped = strip_summary_section(&broken);
-        assert_eq!(stripped, "Base.");
-    }
-}
-
 /// Release acceptance: the full two-task Agent Mail matrix in one run, with
 /// evidence written to disk. Run explicitly:
 ///   CODEWHALE_MAIL_ACCEPTANCE_DIR=/Volumes/VIXinSSD/CW/artifacts/... \