feat: integrate durable run transitions

crates/protocol/src/lib.rs

@@ -919,6 +919,32 @@ pub struct SessionSnapshotEntry {
     pub data: SessionSnapshotEntryData,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[serde(rename_all = "snake_case")]
+pub enum RunYieldReason {
+    Compaction,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[serde(rename_all = "snake_case")]
+pub enum RunResumeSource {
+    Compaction,
+    Pause,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[serde(rename_all = "snake_case")]
+pub enum RunFailureKind {
+    Engine,
+    Compaction,
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
@@ -952,8 +978,26 @@ pub enum SessionSnapshotEntryData {
         #[cfg_attr(feature = "typescript", ts(type = "unknown"))]
         data: Option<serde_json::Value>,
     },
+    /// Durable, non-terminal control return from one logical Run. This is not a
+    /// completed run and must not be used to infer the current Worker state.
+    RunYielded {
+        reason: RunYieldReason,
+        active_run_turn_count: usize,
+    },
+    /// Durable boundary immediately before Engine execution resumes. The source
+    /// distinguishes automatic compaction continuation from an intentional
+    /// user-pause resume; it does not prove that later model output occurred.
+    RunResumed {
+        source: RunResumeSource,
+        active_run_turn_count: usize,
+    },
+    /// Durable terminal for an intentional cancellation. Consumers must not
+    /// present this lifecycle result as an execution failure.
+    RunCancelled,
     RunError {
         message: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        failure: Option<RunFailureKind>,
     },
 }
 
@@ -1696,6 +1740,9 @@ pub enum RunResult {
     Finished,
     Paused,
     LimitReached,
+    /// The logical Run ended because the user intentionally cancelled it after
+    /// observable work. Unlike `RolledBack`, committed history is retained.
+    Cancelled,
     /// The accepted Method::Submit produced no assistant/tool output before
     /// user interruption, so the Worker rolled the submit-time turn state back
     /// to its pre-submit snapshot. Clients should treat the Worker as Idle and

crates/protocol/src/typescript.rs

@@ -10,8 +10,9 @@ use crate::{
     InFlightCompaction, InFlightSnapshot, InFlightToolCallState, InternalWorkerKind,
     InternalWorkerRef, InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method,
     PasteArtifactAvailability, PasteArtifactMediaType, PasteArtifactRef, PendingSubmissionSummary,
-    PendingSubmissionsSnapshot, Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult,
+    PendingSubmissionsSnapshot, Permission, RewindSummary, RewindTarget, RewindTargetId,
-    ScopeRule, Segment, SessionContentPart, SessionConversationTurn, SessionEntryProvenance,
+    RunFailureKind, RunResult, RunResumeSource, RunYieldReason, ScopeRule, Segment,
+    SessionContentPart, SessionConversationTurn, SessionEntryProvenance,
     SessionHistoryLineageBoundary, SessionHistoryPage, SessionMessageRole, SessionSnapshot,
     SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment, SubmissionDisposition,
     SymlinkPolicy, ToolResultDisposition, TurnResult, UploadedFileAvailability, UploadedFileRef,
@@ -103,6 +104,9 @@ pub fn generated_protocol_types() -> String {
     push_decl::<SessionMessageRole>(&cfg, &mut output);
     push_decl::<SessionContentPart>(&cfg, &mut output);
     push_decl::<SessionToolAttachment>(&cfg, &mut output);
+    push_decl::<RunYieldReason>(&cfg, &mut output);
+    push_decl::<RunResumeSource>(&cfg, &mut output);
+    push_decl::<RunFailureKind>(&cfg, &mut output);
     push_decl::<SessionSnapshotEntryData>(&cfg, &mut output);
     push_decl::<SessionSnapshotEntry>(&cfg, &mut output);
     push_decl::<SessionConversationTurn>(&cfg, &mut output);

crates/session-store/src/lib.rs

@@ -107,7 +107,8 @@ pub use segment::{
     SegmentStartState, append_entry, append_system_item, classify_logged_history_entry,
     create_compacted_segment, create_segment, create_segment_with_ids, ensure_head_or_fork, fork,
     fork_at, restore, restore_by_segment, save_config_changed, save_delta, save_extension,
-    save_run_completed, save_run_errored, save_turn_end, save_usage, save_user_input,
+    save_run_completed, save_run_errored, save_run_resumed, save_run_yielded, save_turn_end,
+    save_usage, save_user_input,
 };
 pub use segment_log::{LogEntry, RestoredState, SegmentOrigin, SessionExtension, collect_state};
 pub use store::{Store, StoreError};

crates/session-store/src/public_snapshot.rs

@@ -985,6 +985,8 @@ fn project_session_snapshot_for_segment(
 ) -> SessionSnapshot {
     let mut session_key = session_id;
     let mut entries = Vec::new();
+    let mut total_turn_count = 0usize;
+    let mut active_run_turn_count = None;
 
     for (log_index, record) in log.iter().enumerate() {
         match record {
@@ -996,6 +998,8 @@ fn project_session_snapshot_for_segment(
             } => {
                 session_key = *session_id;
                 entries.clear();
+                total_turn_count = 0;
+                active_run_turn_count = None;
                 extend_history(&mut entries, history, None, *ts);
             }
             LogEntry::InputSegmentsCheckpoint { user_segments, .. } => {
@@ -1035,24 +1039,119 @@ fn project_session_snapshot_for_segment(
                 provenance(&entry.metadata.origin),
                 derivation_ids(entry),
             )),
-            LogEntry::RunErrored { ts, message, .. } => entries.push(legacy_entry(
+            LogEntry::RunYielded {
-                &session_key,
+                ts,
-                segment_id.as_ref(),
+                reason,
-                log_index,
+                active_run_turn_count: count,
-                0,
+            } => {
-                *ts,
+                active_run_turn_count = Some(*count);
-                SessionSnapshotEntryData::RunError {
+                entries.push(legacy_entry(
-                    message: message.clone(),
+                    &session_key,
-                },
+                    segment_id.as_ref(),
-            )),
+                    log_index,
-            // Run checkpoints, configuration, usage, and extension state are
+                    0,
-            // controller/storage authority rather than committed conversation.
+                    *ts,
-            LogEntry::Invoke { .. }
+                    SessionSnapshotEntryData::RunYielded {
-            | LogEntry::TurnEnd { .. }
+                        reason: *reason,
-            | LogEntry::RunCompleted { .. }
+                        active_run_turn_count: *count,
-            | LogEntry::ActiveRunCheckpoint { .. }
+                    },
-            | LogEntry::PausedTurnAbandoned { .. }
+                ));
-            | LogEntry::ConfigChanged { .. }
+            }
+            LogEntry::RunResumed {
+                ts,
+                source,
+                active_run_turn_count: count,
+            } => {
+                active_run_turn_count = Some(*count);
+                entries.push(legacy_entry(
+                    &session_key,
+                    segment_id.as_ref(),
+                    log_index,
+                    0,
+                    *ts,
+                    SessionSnapshotEntryData::RunResumed {
+                        source: *source,
+                        active_run_turn_count: *count,
+                    },
+                ));
+            }
+            LogEntry::RunCancelled { ts } => {
+                active_run_turn_count = None;
+                entries.push(legacy_entry(
+                    &session_key,
+                    segment_id.as_ref(),
+                    log_index,
+                    0,
+                    *ts,
+                    SessionSnapshotEntryData::RunCancelled,
+                ));
+            }
+            LogEntry::RunErrored {
+                ts,
+                message,
+                failure,
+                ..
+            } => {
+                active_run_turn_count = None;
+                entries.push(legacy_entry(
+                    &session_key,
+                    segment_id.as_ref(),
+                    log_index,
+                    0,
+                    *ts,
+                    SessionSnapshotEntryData::RunError {
+                        message: message.clone(),
+                        failure: *failure,
+                    },
+                ));
+            }
+            LogEntry::Invoke { .. } => active_run_turn_count = Some(0),
+            LogEntry::TurnEnd { turn_count, .. } => {
+                if let Some(active) = &mut active_run_turn_count {
+                    *active += turn_count.saturating_sub(total_turn_count);
+                }
+                total_turn_count = *turn_count;
+            }
+            LogEntry::RunCompleted {
+                ts,
+                interrupted,
+                result,
+                active_run_turn_count: persisted_count,
+            } => {
+                if *interrupted && matches!(result, agen::EngineResult::Yielded) {
+                    let count = (*persisted_count)
+                        .or(active_run_turn_count)
+                        .unwrap_or_default();
+                    active_run_turn_count = Some(count);
+                    entries.push(legacy_entry(
+                        &session_key,
+                        segment_id.as_ref(),
+                        log_index,
+                        0,
+                        *ts,
+                        SessionSnapshotEntryData::RunYielded {
+                            reason: protocol::RunYieldReason::Compaction,
+                            active_run_turn_count: count,
+                        },
+                    ));
+                } else if *interrupted && matches!(result, agen::EngineResult::Paused) {
+                    active_run_turn_count = (*persisted_count).or(active_run_turn_count);
+                } else {
+                    active_run_turn_count = None;
+                }
+            }
+            LogEntry::ActiveRunCheckpoint {
+                active_turn_count,
+                total_turn_count: persisted_total,
+                ..
+            } => {
+                active_run_turn_count = Some(*active_turn_count);
+                total_turn_count = *persisted_total;
+            }
+            LogEntry::PausedTurnAbandoned { .. } => active_run_turn_count = None,
+            // Configuration, usage, and extension state are controller/storage
+            // authority rather than committed conversation.
+            LogEntry::ConfigChanged { .. }
             | LogEntry::LlmUsage { .. }
             | LogEntry::Extension { .. } => {}
         }
@@ -2678,6 +2777,110 @@ mod tests {
         );
     }
 
+    #[test]
+    fn legacy_run_completed_yield_projects_as_nonterminal_run_yielded() {
+        let session_id = crate::new_session_id();
+        let legacy: LogEntry = serde_json::from_value(serde_json::json!({
+            "kind": "run_completed",
+            "ts": 30,
+            "interrupted": true,
+            "result": "yielded"
+        }))
+        .unwrap();
+        let snapshot = project_session_snapshot(
+            session_id,
+            &[
+                LogEntry::Invoke {
+                    ts: 10,
+                    trigger: protocol::InvokeKind::UserSend,
+                },
+                LogEntry::TurnEnd {
+                    ts: 20,
+                    turn_count: 4,
+                },
+                legacy,
+            ],
+        );
+
+        assert_eq!(snapshot.entries.len(), 1);
+        assert!(matches!(
+            snapshot.entries[0].data,
+            SessionSnapshotEntryData::RunYielded {
+                reason: protocol::RunYieldReason::Compaction,
+                active_run_turn_count: 4,
+            }
+        ));
+    }
+
+    #[test]
+    fn run_transitions_project_with_timestamps_and_typed_causes() {
+        let session_id = crate::new_session_id();
+        let log = vec![
+            LogEntry::RunYielded {
+                ts: 10,
+                reason: protocol::RunYieldReason::Compaction,
+                active_run_turn_count: 2,
+            },
+            LogEntry::RunResumed {
+                ts: 20,
+                source: protocol::RunResumeSource::Compaction,
+                active_run_turn_count: 2,
+            },
+            LogEntry::RunErrored {
+                ts: 30,
+                interrupted: false,
+                message: "mid-run compaction failed: unavailable".into(),
+                failure: Some(protocol::RunFailureKind::Compaction),
+            },
+        ];
+
+        let snapshot = project_session_snapshot(session_id, &log);
+        assert_eq!(snapshot.entries.len(), 3);
+        assert_eq!(snapshot.entries[0].timestamp, 10);
+        assert!(matches!(
+            snapshot.entries[0].data,
+            SessionSnapshotEntryData::RunYielded {
+                reason: protocol::RunYieldReason::Compaction,
+                active_run_turn_count: 2,
+            }
+        ));
+        assert!(matches!(
+            snapshot.entries[1].data,
+            SessionSnapshotEntryData::RunResumed {
+                source: protocol::RunResumeSource::Compaction,
+                active_run_turn_count: 2,
+            }
+        ));
+        assert!(matches!(
+            snapshot.entries[2].data,
+            SessionSnapshotEntryData::RunError {
+                failure: Some(protocol::RunFailureKind::Compaction),
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn cancelled_run_projects_as_a_non_failure_terminal() {
+        let snapshot = project_session_snapshot(
+            crate::new_session_id(),
+            &[
+                LogEntry::Invoke {
+                    ts: 10,
+                    trigger: protocol::InvokeKind::UserSend,
+                },
+                LogEntry::RunCancelled { ts: 20 },
+            ],
+        );
+
+        assert_eq!(snapshot.entries.len(), 1);
+        assert_eq!(snapshot.entries[0].timestamp, 20);
+        assert!(matches!(
+            snapshot.entries[0].data,
+            SessionSnapshotEntryData::RunCancelled
+        ));
+    }
+
     #[test]
     fn retained_snapshot_read_projects_active_segment_without_rewriting_files() {
         let root = tempfile::tempdir().unwrap();

crates/session-store/src/segment.rs

@@ -10,7 +10,7 @@ use crate::{LoggedHistoryEntry, LoggedSystemHistoryEntry, SegmentId, SessionId};
 use agen::EngineResult;
 use agen::llm_client::RequestConfig;
 use agen::llm_client::types::Item;
-use protocol::Segment;
+use protocol::{RunFailureKind, RunResumeSource, RunYieldReason, Segment};
 
 /// State snapshot for creating a SegmentStart entry.
 pub struct SegmentStartState<'a> {
@@ -332,6 +332,47 @@ pub fn save_run_completed(
     )
 }
 
+/// Log a non-terminal yield of one logical Run. New writers must use this
+/// instead of encoding `EngineResult::Yielded` in `RunCompleted`.
+pub fn save_run_yielded(
+    store: &impl Store,
+    session_id: SessionId,
+    segment_id: SegmentId,
+    reason: RunYieldReason,
+    active_run_turn_count: usize,
+) -> Result<(), StoreError> {
+    append_entry(
+        store,
+        session_id,
+        segment_id,
+        LogEntry::RunYielded {
+            ts: segment_log::now_millis(),
+            reason,
+            active_run_turn_count,
+        },
+    )
+}
+
+/// Log the boundary immediately before Engine resumes one logical Run.
+pub fn save_run_resumed(
+    store: &impl Store,
+    session_id: SessionId,
+    segment_id: SegmentId,
+    source: RunResumeSource,
+    active_run_turn_count: usize,
+) -> Result<(), StoreError> {
+    append_entry(
+        store,
+        session_id,
+        segment_id,
+        LogEntry::RunResumed {
+            ts: segment_log::now_millis(),
+            source,
+            active_run_turn_count,
+        },
+    )
+}
+
 /// Log a `RunErrored` entry — `run()` / `resume()` returned `Err(EngineError)`.
 ///
 /// `EngineError` is not `Serialize`, so the caller passes a lossy
@@ -351,6 +392,7 @@ pub fn save_run_errored(
             ts: segment_log::now_millis(),
             interrupted,
             message,
+            failure: Some(RunFailureKind::Engine),
         },
     )
 }

crates/session-store/src/segment_log.rs

@@ -11,7 +11,7 @@
 
 use agen::llm_client::types::{Item, RequestConfig};
 use agen::{EngineResult, UsageRecord};
-use protocol::{InvokeKind, Segment};
+use protocol::{InvokeKind, RunFailureKind, RunResumeSource, RunYieldReason, Segment};
 use serde::{Deserialize, Serialize};
 
 use crate::history::{LoggedHistoryEntry, LoggedSystemHistoryEntry};
@@ -28,7 +28,7 @@ use crate::logged_item::LoggedItem;
 ///   `run()`/`resume()` (current callers persist a single TurnEnd at
 ///   run completion); the fork-point seq for `at_turn_index` is the
 ///   preceding `Invoke` entry, not the TurnEnd.
-/// - `RunCompleted` / `RunErrored` — marks end of a `run()` or `resume()` call
+/// - `RunCompleted` / `RunCancelled` / `RunErrored` — terminal run outcomes
 /// - `PausedTurnAbandoned` — explicit abandon/cancel of a paused interrupted turn
 /// - `ConfigChanged` — `RequestConfig` mutation
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -86,9 +86,10 @@ pub enum LogEntry {
     /// serde tag already occupies `"kind"`.
     ///
     /// Replay marks the run interrupted until a terminal `RunCompleted`,
-    /// `RunErrored`, or `PausedTurnAbandoned` entry proves how it ended. This
+    /// `RunCancelled`, `RunErrored`, or `PausedTurnAbandoned` entry proves how
-    /// makes a process/disk failure between Invoke and its terminal record
+    /// it ended. This makes a process/disk failure between Invoke and its
-    /// restore conservatively instead of re-running a dangling tool call.
+    /// terminal record restore conservatively instead of re-running a dangling
+    /// tool call.
     Invoke { ts: u64, trigger: InvokeKind },
 
     /// Canonical user submission with its exact model-visible entries. Typed
@@ -126,12 +127,36 @@ pub enum LogEntry {
         ts: u64,
         interrupted: bool,
         result: EngineResult,
-        /// AgentTurns consumed by a paused/yielded logical run. Terminal
+        /// AgentTurns consumed by a paused legacy logical run. Terminal
-        /// outcomes persist `None`.
+        /// outcomes persist `None`. Legacy logs may also contain Yielded here;
+        /// new yields use `RunYielded` below.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         active_run_turn_count: Option<usize>,
     },
 
+    /// The current logical Run returned control so compaction can prepare a new
+    /// Segment. This is an intermediate transition, not a successful terminal
+    /// result. Ordering plus Session/Segment lineage correlate it with the
+    /// preceding Invoke and a later RunResumed or terminal record.
+    RunYielded {
+        ts: u64,
+        reason: RunYieldReason,
+        active_run_turn_count: usize,
+    },
+
+    /// Resume preparation has completed and Engine execution is about to begin.
+    /// A committed record is required before control is handed back to Engine.
+    /// It deliberately does not assert that a subsequent model call occurred.
+    RunResumed {
+        ts: u64,
+        source: RunResumeSource,
+        active_run_turn_count: usize,
+    },
+
+    /// The logical Run ended because the user intentionally cancelled it.
+    /// This is terminal, but is not an execution failure.
+    RunCancelled { ts: u64 },
+
     /// `run()` / `resume()` が `EngineError` で終了した。
     /// `EngineError` は `Serialize` 不可なので `message` のみ lossy 保持する。
     /// Audit-only metadata: replay は `interrupted` のみ反映する。
@@ -139,6 +164,10 @@ pub enum LogEntry {
         ts: u64,
         interrupted: bool,
         message: String,
+        /// Older records omit this field. New records distinguish failures in
+        /// Engine execution from failure to continue after a compaction yield.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        failure: Option<RunFailureKind>,
     },
 
     /// Restores an active logical-run budget at a segment boundary, notably
@@ -351,6 +380,24 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState {
                     state.active_run_turn_count = None;
                 }
             }
+            LogEntry::RunYielded {
+                active_run_turn_count,
+                ..
+            }
+            | LogEntry::RunResumed {
+                active_run_turn_count,
+                ..
+            } => {
+                // Both are resumable boundaries of the same logical Run. A log
+                // ending at RunResumed proves only that the resume boundary was
+                // committed, not that Engine produced later output.
+                state.last_run_interrupted = true;
+                state.active_run_turn_count = Some(*active_run_turn_count);
+            }
+            LogEntry::RunCancelled { .. } => {
+                state.last_run_interrupted = false;
+                state.active_run_turn_count = None;
+            }
             LogEntry::RunErrored { interrupted, .. } => {
                 state.last_run_interrupted = *interrupted;
                 state.active_run_turn_count = None;
@@ -818,6 +865,83 @@ mod tests {
         assert!(state.last_run_interrupted);
     }
 
+    #[test]
+    fn replay_new_yield_and_resume_keep_one_logical_run_budget() {
+        let state = collect_state(&[
+            LogEntry::Invoke {
+                ts: 100,
+                trigger: InvokeKind::UserSend,
+            },
+            LogEntry::TurnEnd {
+                ts: 200,
+                turn_count: 2,
+            },
+            LogEntry::RunYielded {
+                ts: 300,
+                reason: RunYieldReason::Compaction,
+                active_run_turn_count: 2,
+            },
+            LogEntry::ActiveRunCheckpoint {
+                ts: 400,
+                active_turn_count: 2,
+                total_turn_count: 2,
+            },
+            LogEntry::RunResumed {
+                ts: 500,
+                source: RunResumeSource::Compaction,
+                active_run_turn_count: 2,
+            },
+        ]);
+
+        assert!(state.last_run_interrupted);
+        assert_eq!(state.turn_count, 2);
+        assert_eq!(state.active_run_turn_count, Some(2));
+    }
+
+    #[test]
+    fn replay_cancelled_run_is_terminal_without_becoming_a_failure() {
+        let state = collect_state(&[
+            LogEntry::Invoke {
+                ts: 100,
+                trigger: InvokeKind::UserSend,
+            },
+            LogEntry::TurnEnd {
+                ts: 200,
+                turn_count: 1,
+            },
+            LogEntry::RunCancelled { ts: 300 },
+        ]);
+
+        assert!(!state.last_run_interrupted);
+        assert_eq!(state.active_run_turn_count, None);
+        assert_eq!(state.turn_count, 1);
+    }
+
+    #[test]
+    fn legacy_yielded_run_completed_remains_resumable_but_is_not_terminal_success() {
+        let entry: LogEntry = serde_json::from_value(serde_json::json!({
+            "kind": "run_completed",
+            "ts": 300,
+            "interrupted": true,
+            "result": "yielded"
+        }))
+        .expect("legacy yielded run-completed entry");
+        let state = collect_state(&[
+            LogEntry::Invoke {
+                ts: 100,
+                trigger: InvokeKind::UserSend,
+            },
+            LogEntry::TurnEnd {
+                ts: 200,
+                turn_count: 3,
+            },
+            entry,
+        ]);
+
+        assert!(state.last_run_interrupted);
+        assert_eq!(state.active_run_turn_count, Some(3));
+    }
+
     #[test]
     fn legacy_interrupted_run_derives_budget_from_invoke_and_turn_end() {
         let entry: LogEntry = serde_json::from_value(serde_json::json!({

crates/session-store/tests/session_test.rs

@@ -179,14 +179,13 @@ async fn run_and_persist(
     session_store::save_turn_end(store, session_id, segment_id, worker.turn_count()).unwrap();
 
     match &result {
-        agen::EngineRunExit::Finished
+        agen::EngineRunExit::Finished | agen::EngineRunExit::Paused => {
-        | agen::EngineRunExit::Paused
-        | agen::EngineRunExit::Yielded => {
             let (legacy_result, interrupted) = match &result {
                 agen::EngineRunExit::Finished => (agen::EngineResult::Finished, false),
                 agen::EngineRunExit::Paused => (agen::EngineResult::Paused, true),
-                agen::EngineRunExit::Yielded => (agen::EngineResult::Yielded, true),
+                agen::EngineRunExit::Yielded | agen::EngineRunExit::Interrupted(_) => {
-                agen::EngineRunExit::Interrupted(_) => unreachable!(),
+                    unreachable!()
+                }
             };
             session_store::save_run_completed(
                 store,
@@ -198,6 +197,18 @@ async fn run_and_persist(
             )
             .unwrap();
         }
+        agen::EngineRunExit::Yielded => {
+            session_store::save_run_yielded(
+                store,
+                session_id,
+                segment_id,
+                protocol::RunYieldReason::Compaction,
+                worker
+                    .active_run_turn_count()
+                    .expect("yielded test run retains its turn budget"),
+            )
+            .unwrap();
+        }
         agen::EngineRunExit::Interrupted(agen::RunInterruptionReason::LimitReached) => {
             session_store::save_run_completed(
                 store,

crates/tui/src/app.rs

@@ -961,6 +961,34 @@ impl App {
     }
 
     fn push_run_error(&mut self, message: String) {
+        if self
+            .run_error_messages
+            .iter()
+            .any(|existing| run_failure_messages_match(existing, &message))
+        {
+            return;
+        }
+        self.run_error_messages.push(message.clone());
+        self.blocks.push(Block::Alert {
+            level: AlertLevel::Error,
+            source: AlertSource::Worker,
+            message,
+        });
+    }
+
+    fn push_durable_run_error(&mut self, message: String) {
+        self.run_error_messages
+            .retain(|existing| !run_failure_messages_match(existing, &message));
+        self.blocks.retain(|block| {
+            !matches!(
+                block,
+                Block::Alert {
+                    level: AlertLevel::Error,
+                    message: existing,
+                    ..
+                } if run_failure_messages_match(existing, &message)
+            )
+        });
         self.run_error_messages.push(message.clone());
         self.blocks.push(Block::Alert {
             level: AlertLevel::Error,
@@ -970,6 +998,13 @@ impl App {
     }
 
     fn handle_error(&mut self, code: ErrorCode, message: String) {
+        if self
+            .run_error_messages
+            .iter()
+            .any(|existing| run_failure_messages_match(existing, &message))
+        {
+            return;
+        }
         let text = format!("[{code:?}] {message}");
         let was_applying = if let Some(picker) = self.rewind_picker.as_mut() {
             let applying = picker.applying;
@@ -1169,9 +1204,11 @@ impl App {
                 }
                 self.assistant_streaming = false;
             }
-            // The TUI already renders assistant/tool streaming events. This commit
+            Event::SessionEntryCommitted { entry } => {
-            // marker exists so identity-aware clients can reconcile durable history.
+                if let protocol::SessionSnapshotEntryData::RunError { message, .. } = entry.data {
-            Event::SessionEntryCommitted { .. } => {}
+                    self.push_durable_run_error(message);
+                }
+            }
             Event::SystemItem { item } => {
                 self.apply_system_item(&item);
                 self.assistant_streaming = false;
@@ -1495,6 +1532,14 @@ impl App {
                 }
             }
             Event::Alert(alert) => {
+                if alert.level == AlertLevel::Error
+                    && self
+                        .run_error_messages
+                        .iter()
+                        .any(|message| run_failure_messages_match(message, &alert.message))
+                {
+                    return None;
+                }
                 self.blocks.push(Block::Alert {
                     level: alert.level,
                     source: alert.source,
@@ -2499,7 +2544,13 @@ impl App {
                         self.apply_system_item(data);
                     }
                 }
-                SessionSnapshotEntryData::RunError { message } => {
+                SessionSnapshotEntryData::RunYielded { .. }
+                | SessionSnapshotEntryData::RunResumed { .. }
+                | SessionSnapshotEntryData::RunCancelled => {
+                    // Durable logical-Run transitions are intentionally not
+                    // presentation rows and never restore runtime progress.
+                }
+                SessionSnapshotEntryData::RunError { message, .. } => {
                     self.push_run_error(message.clone());
                 }
             }
@@ -2606,6 +2657,12 @@ pub fn fmt_tokens(n: u64) -> String {
     }
 }
 
+fn run_failure_messages_match(left: &str, right: &str) -> bool {
+    !left.is_empty()
+        && !right.is_empty()
+        && (left == right || left.ends_with(right) || right.ends_with(left))
+}
+
 fn fmt_millis(ms: u64) -> String {
     if ms >= 1_000 {
         format!("{:.1}s", ms as f64 / 1_000.0)
@@ -3480,7 +3537,7 @@ mod completion_flow_tests {
 
     #[test]
     fn non_rolled_back_run_end_keeps_submitted_blocks_and_does_not_restore_input() {
-        for result in [RunResult::Paused, RunResult::Finished] {
+        for result in [RunResult::Paused, RunResult::Finished, RunResult::Cancelled] {
             let mut app = App::new("test".into());
             let submitted = submit_text(&mut app, "normal run");
             app.handle_worker_event(Event::UserMessage {
@@ -3745,6 +3802,7 @@ mod completion_flow_tests {
             ts: 3,
             interrupted: false,
             message: "provider unavailable".into(),
+            failure: None,
         };
         app.handle_worker_event(Event::Snapshot {
             greeting: test_greeting(),
@@ -3769,6 +3827,48 @@ mod completion_flow_tests {
         assert_eq!(errors, ["provider unavailable"]);
     }
 
+    #[test]
+    fn live_durable_compaction_failure_reconciles_transient_alert_and_error() {
+        let mut app = App::new("test".into());
+        let message = "mid-run compaction failed: summary unavailable";
+        app.handle_worker_event(Event::Alert(protocol::Alert {
+            level: AlertLevel::Error,
+            source: AlertSource::Compactor,
+            message: message.into(),
+            timestamp_ms: 1,
+        }));
+        app.handle_worker_event(Event::SessionEntryCommitted {
+            entry: protocol::SessionSnapshotEntry {
+                entry_id: "run-failure".into(),
+                timestamp: 2,
+                provenance: protocol::SessionEntryProvenance::LegacyUnknown,
+                derived_from: Vec::new(),
+                data: protocol::SessionSnapshotEntryData::RunError {
+                    message: message.into(),
+                    failure: Some(protocol::RunFailureKind::Compaction),
+                },
+            },
+        });
+        app.handle_worker_event(Event::Error {
+            code: ErrorCode::Internal,
+            message: "summary unavailable".into(),
+        });
+
+        let errors = app
+            .blocks
+            .iter()
+            .filter_map(|block| match block {
+                Block::Alert {
+                    level: AlertLevel::Error,
+                    message,
+                    ..
+                } => Some(message.as_str()),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(errors, [message]);
+    }
+
     #[test]
     fn segment_rotation_retains_live_error_across_real_segment_start() {
         let mut app = App::new("test".into());

crates/worker-runtime/src/runtime.rs

@@ -8042,6 +8042,7 @@ mod tests {
                         derived_from: Vec::new(),
                         data: protocol::SessionSnapshotEntryData::RunError {
                             message: expected_entry.to_string(),
+                            failure: None,
                         },
                     }],
                 },

crates/worker/examples/worker_cli.rs

@@ -66,6 +66,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         WorkerRunResult::Finished => println!("(finished)"),
         WorkerRunResult::Paused => println!("(paused)"),
         WorkerRunResult::LimitReached => println!("(turn limit reached)"),
+        WorkerRunResult::Cancelled => println!("(cancelled)"),
         WorkerRunResult::Interrupted { message, .. } => println!("(interrupted: {message})"),
         WorkerRunResult::RolledBack => println!("(empty turn rolled back)"),
     }

crates/worker/src/controller.rs

@@ -2794,6 +2794,12 @@ where
                             WorkerRunResult::Paused => (WorkerStatus::Paused, RunResult::Paused),
                             WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached),
                             WorkerRunResult::RolledBack => (WorkerStatus::Idle, RunResult::RolledBack),
+                            WorkerRunResult::Cancelled if pause_requested => {
+                                (WorkerStatus::Paused, RunResult::Paused)
+                            }
+                            WorkerRunResult::Cancelled => {
+                                (WorkerStatus::Idle, RunResult::Cancelled)
+                            }
                             WorkerRunResult::Interrupted { .. } if pause_requested => {
                                 let _ = working_event_tx.send(Event::RunEnd { result: RunResult::Paused });
                                 return (WorkerStatus::Paused, shutdown_requested, false);

crates/worker/src/feature/builtin/memory_lifecycle.rs

@@ -632,7 +632,7 @@ fn extraction_disposition(
             ExtractionDisposition::Failed(error.source.to_string())
         }
         Ok(run) => match &run.lifecycle {
-            WorkerRunResult::RolledBack => {
+            WorkerRunResult::Cancelled | WorkerRunResult::RolledBack => {
                 ExtractionDisposition::Cancelled("memory-extract cancelled".to_string())
             }
             WorkerRunResult::Interrupted { message, .. } => {

crates/worker/src/internal_worker.rs

@@ -213,6 +213,7 @@ where
     let outcome = match worker.run_text(&input).await {
         Ok(lifecycle @ WorkerRunResult::Finished)
         | Ok(lifecycle @ WorkerRunResult::Paused)
+        | Ok(lifecycle @ WorkerRunResult::Cancelled)
         | Ok(lifecycle @ WorkerRunResult::RolledBack) => Ok(InternalWorkerResult {
             usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
             identity,
@@ -333,6 +334,10 @@ fn classify_internal_turn_result(
         Ok(WorkerRunResult::Interrupted { message, .. }) => {
             (InternalWorkerSessionStatus::Stopped, Some(message))
         }
+        Ok(WorkerRunResult::Cancelled) => (
+            InternalWorkerSessionStatus::Stopped,
+            Some("internal Worker run was cancelled".to_string()),
+        ),
         Ok(WorkerRunResult::RolledBack) => (
             InternalWorkerSessionStatus::Stopped,
             Some("internal Worker run was cancelled before AI output".to_string()),

crates/worker/src/ipc/protocol_session.rs

@@ -60,8 +60,12 @@ pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
                 .map(|entry| entry.entry_id.clone());
             Some(Event::UserMessage { entry_id, segments })
         }
-        entry
+        entry @ (LogEntry::AnnotatedAssistantItem { .. }
-        @ (LogEntry::AnnotatedAssistantItem { .. } | LogEntry::AnnotatedToolResult { .. }) => {
+        | LogEntry::AnnotatedToolResult { .. }
+        | LogEntry::RunYielded { .. }
+        | LogEntry::RunResumed { .. }
+        | LogEntry::RunCancelled { .. }
+        | LogEntry::RunErrored { .. }) => {
             let mut projected =
                 session_store::public_snapshot::project_current_session_snapshot(&[entry]);
             projected
@@ -114,6 +118,30 @@ pub async fn dispatch_worker_protocol_method(
 mod tests {
     use super::*;
 
+    #[test]
+    fn durable_run_transition_maps_to_session_entry_committed() {
+        let event = live_log_entry_event(LogEntry::RunYielded {
+            ts: 42,
+            reason: protocol::RunYieldReason::Compaction,
+            active_run_turn_count: 3,
+        })
+        .expect("RunYielded must be live-relevant");
+
+        match event {
+            Event::SessionEntryCommitted { entry } => {
+                assert_eq!(entry.timestamp, 42);
+                assert!(matches!(
+                    entry.data,
+                    protocol::SessionSnapshotEntryData::RunYielded {
+                        reason: protocol::RunYieldReason::Compaction,
+                        active_run_turn_count: 3,
+                    }
+                ));
+            }
+            other => panic!("expected SessionEntryCommitted, got {other:?}"),
+        }
+    }
+
     #[test]
     fn user_input_log_entry_maps_to_user_message_event() {
         let segments = vec![protocol::Segment::text("hello from log")];

crates/worker/src/segment_log_sink.rs

@@ -123,6 +123,10 @@ impl SegmentLogSink {
                 | LogEntry::AnnotatedToolResult { .. }
                 | LogEntry::AnnotatedSystemItem { .. }
                 | LogEntry::Invoke { .. }
+                | LogEntry::RunYielded { .. }
+                | LogEntry::RunResumed { .. }
+                | LogEntry::RunCancelled { .. }
+                | LogEntry::RunErrored { .. }
         )
     }
 
@@ -323,9 +327,17 @@ mod tests {
             other => panic!("expected SystemItem, got {other:?}"),
         }
 
+        // Durable logical-Run transitions share the committed-entry lane.
+        sink.publish(LogEntry::RunResumed {
+            ts: now_millis(),
+            source: protocol::RunResumeSource::Compaction,
+            active_run_turn_count: 1,
+        });
+        assert!(matches!(rx.try_recv(), Ok(LogEntry::RunResumed { .. })));
+
         // Mirror still grew with all entries (snapshot completeness).
         let (after_snapshot, _) = sink.subscribe_with_snapshot();
-        assert_eq!(after_snapshot.len(), 4);
+        assert_eq!(after_snapshot.len(), 5);
     }
 
     #[test]

crates/worker/src/session_capture.rs

@@ -285,6 +285,9 @@ impl SessionCapture {
                     // Observation deliberately excludes system items and
                     // controller errors from model-visible session evidence.
                     SessionSnapshotEntryData::SystemItem { .. }
+                    | SessionSnapshotEntryData::RunYielded { .. }
+                    | SessionSnapshotEntryData::RunResumed { .. }
+                    | SessionSnapshotEntryData::RunCancelled
                     | SessionSnapshotEntryData::RunError { .. } => return None,
                 };
                 Some(CapturedHistoryEntry {

crates/worker/src/spawn/comm_tools.rs

@@ -83,6 +83,7 @@ mod tests {
                         derived_from: Vec::new(),
                         data: protocol::SessionSnapshotEntryData::RunError {
                             message: value.to_string(),
+                            failure: None,
                         },
                     })
                     .collect(),

crates/worker/src/worker.rs

@@ -424,7 +424,7 @@ use async_trait::async_trait;
 use protocol::{
     AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, CompactionPhase,
     CompactionTrigger, ErrorCode, Event, InFlightCompaction, RewindSummary, RewindTarget,
-    RewindTargetId, Segment,
+    RewindTargetId, RunFailureKind, RunResumeSource, RunYieldReason, Segment,
 };
 use tokio::net::UnixStream;
 use tokio::sync::broadcast;
@@ -3471,8 +3471,12 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                         result: EngineResult::Finished,
                         ..
                     } => Some(CommittedRunExit::Finished),
-                    LogEntry::RunCompleted { .. } => Some(CommittedRunExit::NonFinal),
+                    LogEntry::RunCompleted { .. }
-                    LogEntry::RunErrored { .. } => Some(CommittedRunExit::Interrupted),
+                    | LogEntry::RunYielded { .. }
+                    | LogEntry::RunResumed { .. } => Some(CommittedRunExit::NonFinal),
+                    LogEntry::RunCancelled { .. } | LogEntry::RunErrored { .. } => {
+                        Some(CommittedRunExit::Interrupted)
+                    }
                     _ => None,
                 })
                 .unwrap_or(CommittedRunExit::NonFinal);
@@ -4660,12 +4664,40 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         self.handle_worker_result(result, history_before).await
     }
 
-    /// Resume from a paused state.
+    /// Resume from an intentional user pause. The durable transition is
+    /// committed only after preparation succeeds and immediately before Engine
+    /// receives control.
     pub async fn resume(&mut self) -> Result<WorkerRunResult, WorkerError>
+    where
+        St: Clone + 'static,
+    {
+        self.resume_from(RunResumeSource::Pause).await
+    }
+
+    async fn resume_from(&mut self, source: RunResumeSource) -> Result<WorkerRunResult, WorkerError>
     where
         St: Clone + 'static,
     {
         self.prepare_for_run().await?;
+        self.resume_prepared(source).await
+    }
+
+    async fn resume_prepared(
+        &mut self,
+        source: RunResumeSource,
+    ) -> Result<WorkerRunResult, WorkerError>
+    where
+        St: Clone + 'static,
+    {
+        let active_run_turn_count = self
+            .engine()
+            .active_run_turn_count()
+            .expect("pause/yield resume must retain the logical-run turn budget");
+        self.commit_entry(LogEntry::RunResumed {
+            ts: segment_log::now_millis(),
+            source,
+            active_run_turn_count,
+        })?;
 
         let history_before = self.session.history().len();
         let mut annotate = history_annotator(
@@ -4829,12 +4861,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         Ok(())
     }
 
-    /// Handle Engine result: always persist the turn first, then if
+    /// Handle one Engine execution boundary: persist its turn and transition,
-    /// `Yielded`, perform compaction and resume.
+    /// then compact and resume when Engine yielded for compaction.
     ///
-    /// Persisting before compaction ensures that if compact fails, the
+    /// A yield commits `RunYielded`, never `RunCompleted`. Successful preparation
-    /// turn is fully recorded in the old session (interrupted, outcome
+    /// later commits `RunResumed` on the active replacement Segment immediately
-    /// `Yielded`), so restore remains consistent.
+    /// before Engine regains control; a compaction failure instead commits the
+    /// same logical Run's terminal `RunErrored`.
     async fn handle_worker_result(
         &mut self,
         result: EngineRunExit,
@@ -4867,7 +4900,18 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         if matches!(&result, EngineRunExit::Interrupted(_)) {
             self.terminalize_orphan_tool_calls()?;
         }
-        self.persist_turn(history_before, &result).await?;
+        let request_block = self
+            .compact_state
+            .as_ref()
+            .and_then(|state| state.take_pending_request_block());
+        let request_block_message = request_block.map(|block| {
+            format!(
+                "mid-run compaction could not continue: {}",
+                automatic_compact_block_error(block)
+            )
+        });
+        self.persist_turn(history_before, &result, request_block_message.as_deref())
+            .await?;
         let committed_invocation = self.hook_invocation_context(Some(run_id));
         if let Some(hooks) = self.hook_registry.clone() {
             let context = RunCommittedContext {
@@ -4895,15 +4939,17 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
             tracing::warn!(error = %error, "run-committed background task start failed");
         }
 
-        let request_block = self
-            .compact_state
-            .as_ref()
-            .and_then(|state| state.take_pending_request_block());
         if let Some(block) = request_block {
+            let error = automatic_compact_block_error(block);
+            if matches!(result, EngineRunExit::Yielded) {
+                self.persist_compaction_run_failure(
+                    request_block_message.expect("blocked request has a terminal reason"),
+                );
+            }
             if let Some(state) = &self.compact_state {
                 state.finish_logical_run();
             }
-            return Err(automatic_compact_block_error(block));
+            return Err(error);
         }
 
         if matches!(result, EngineRunExit::Yielded) {
@@ -4930,6 +4976,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                 self.last_run_interrupted = false;
                 Ok(WorkerRunResult::LimitReached)
             }
+            EngineRunExit::Interrupted(RunInterruptionReason::Cancelled) => {
+                self.last_run_interrupted = false;
+                Ok(WorkerRunResult::Cancelled)
+            }
             EngineRunExit::Interrupted(reason) => {
                 self.last_run_interrupted = true;
                 Ok(WorkerRunResult::Interrupted {
@@ -4949,6 +4999,36 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         }
     }
 
+    fn persist_compaction_run_failure(&mut self, message: String) {
+        // Keep the original compaction/preparation error as controller authority.
+        // A terminal append failure is diagnosed separately and cannot publish a
+        // false persisted result through the session-log sink.
+        match self.commit_entry(LogEntry::RunErrored {
+            ts: segment_log::now_millis(),
+            interrupted: false,
+            message,
+            failure: Some(RunFailureKind::Compaction),
+        }) {
+            Ok(()) => {
+                self.last_run_interrupted = false;
+                self.engine_mut().set_active_run_turn_count(None);
+            }
+            Err(persist_error) => {
+                warn!(
+                    error = %persist_error,
+                    "failed to persist mid-run compaction terminal record"
+                );
+                self.alert(
+                    AlertLevel::Error,
+                    AlertSource::Worker,
+                    format!(
+                        "mid-run compaction failed and its Run terminal record could not be saved: {persist_error}"
+                    ),
+                );
+            }
+        }
+    }
+
     /// Perform compaction after a `compact_needed` abort and resume execution.
     ///
     /// Uses `Box::pin` for the recursive `resume()` call to break the
@@ -4992,15 +5072,21 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                             "automatic compaction must complete a claimed attempt"
                         );
                     }
-                    self.resume().await
+                    if let Err(error) = self.prepare_for_run().await {
+                        if let Some(state) = &state {
+                            state.finish_logical_run();
+                        }
+                        let message =
+                            format!("mid-run resume preparation failed after compaction: {error}");
+                        self.persist_compaction_run_failure(message.clone());
+                        self.alert(AlertLevel::Error, AlertSource::Compactor, message);
+                        return Err(error);
+                    }
+                    self.resume_prepared(RunResumeSource::Compaction).await
                 }
                 Err(e) => {
                     warn!(error = %e, "Compaction failed during run");
-                    self.alert(
+                    let message = format!("mid-run compaction failed: {e}");
-                        AlertLevel::Error,
-                        AlertSource::Compactor,
-                        format!("mid-run compaction failed: {e}"),
-                    );
                     if let Some(state) = &state {
                         let outcome = if matches!(e, WorkerError::CompactCancelled) {
                             CompactionOutcome::Cancelled
@@ -5012,7 +5098,17 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                             completed,
                             "automatic compaction must complete a claimed attempt"
                         );
+                        state.finish_logical_run();
                     }
+
+                    // The yielded logical Run is no longer resumable. Commit its
+                    // durable failure before any log-derived live event can claim
+                    // the terminal transition. If storage itself is unavailable,
+                    // the helper retains the original compaction error (and its
+                    // fail-closed controller classification) while diagnosing the
+                    // missing terminal record separately.
+                    self.persist_compaction_run_failure(message.clone());
+                    self.alert(AlertLevel::Error, AlertSource::Compactor, message);
                     Err(e)
                 }
             }
@@ -5209,6 +5305,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         &mut self,
         history_before: usize,
         result: &EngineRunExit,
+        compaction_failure: Option<&str>,
     ) -> Result<(), StoreError> {
         // Per-item commits for AssistantItem / ToolResult / SystemItem
         // entries are expected to have landed synchronously: the
@@ -5314,13 +5411,23 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                 | EngineRunExit::Interrupted(RunInterruptionReason::Unexpected(_))
         );
         let active_run_turn_count = self.engine.as_ref().unwrap().active_run_turn_count();
+        if let Some(message) = compaction_failure
+            && !matches!(result, EngineRunExit::Yielded)
+        {
+            self.commit_entry(LogEntry::RunErrored {
+                ts: segment_log::now_millis(),
+                interrupted: false,
+                message: message.to_string(),
+                failure: Some(RunFailureKind::Compaction),
+            })?;
+            return Ok(());
+        }
         match result {
-            EngineRunExit::Finished | EngineRunExit::Paused | EngineRunExit::Yielded => {
+            EngineRunExit::Finished | EngineRunExit::Paused => {
                 let result = match result {
                     EngineRunExit::Finished => EngineResult::Finished,
                     EngineRunExit::Paused => EngineResult::Paused,
-                    EngineRunExit::Yielded => EngineResult::Yielded,
+                    EngineRunExit::Yielded | EngineRunExit::Interrupted(_) => unreachable!(),
-                    EngineRunExit::Interrupted(_) => unreachable!(),
                 };
                 self.commit_entry(LogEntry::RunCompleted {
                     ts: segment_log::now_millis(),
@@ -5329,6 +5436,14 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                     active_run_turn_count,
                 })?;
             }
+            EngineRunExit::Yielded => {
+                self.commit_entry(LogEntry::RunYielded {
+                    ts: segment_log::now_millis(),
+                    reason: RunYieldReason::Compaction,
+                    active_run_turn_count: active_run_turn_count
+                        .expect("yielded logical run must retain its turn budget"),
+                })?;
+            }
             EngineRunExit::Interrupted(RunInterruptionReason::LimitReached) => {
                 self.commit_entry(LogEntry::RunCompleted {
                     ts: segment_log::now_millis(),
@@ -5337,11 +5452,17 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                     active_run_turn_count,
                 })?;
             }
+            EngineRunExit::Interrupted(RunInterruptionReason::Cancelled) => {
+                self.commit_entry(LogEntry::RunCancelled {
+                    ts: segment_log::now_millis(),
+                })?;
+            }
             EngineRunExit::Interrupted(reason) => {
                 self.commit_entry(LogEntry::RunErrored {
                     ts: segment_log::now_millis(),
                     interrupted,
                     message: run_interruption_reason_message(reason),
+                    failure: Some(RunFailureKind::Engine),
                 })?;
             }
         }
@@ -7266,6 +7387,8 @@ pub enum WorkerRunResult {
     Paused,
     /// The worker reached its configured max_turns limit.
     LimitReached,
+    /// The logical Run was intentionally cancelled after observable work.
+    Cancelled,
     /// The run was interrupted by a known or unexpected terminal cause.
     Interrupted { code: ErrorCode, message: String },
     /// The submit-time user turn was rolled back because no AI output was materialized.
@@ -9127,6 +9250,146 @@ mod build_summary_prompt_tests {
         );
     }
 
+    #[tokio::test]
+    async fn yielded_engine_exit_commits_intermediate_transition_not_run_completed() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
+        let mut worker = Worker::new(
+            minimal_manifest(),
+            Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient),
+            store.clone(),
+            WorkerWorkspaceContext::no_workspace(),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        worker.ensure_segment_head().await.unwrap();
+        worker.engine_mut().set_active_run_turn_count(Some(2));
+
+        worker
+            .persist_turn(0, &EngineRunExit::Yielded, None)
+            .await
+            .unwrap();
+
+        let entries = store
+            .read_all(worker.session_id(), worker.segment_id())
+            .unwrap();
+        assert!(entries.iter().any(|entry| matches!(
+            entry,
+            LogEntry::RunYielded {
+                reason: RunYieldReason::Compaction,
+                active_run_turn_count: 2,
+                ..
+            }
+        )));
+        assert!(!entries.iter().any(|entry| matches!(
+            entry,
+            LogEntry::RunCompleted {
+                result: EngineResult::Yielded,
+                ..
+            }
+        )));
+    }
+
+    #[tokio::test]
+    async fn failed_compaction_terminalizes_the_yielded_run_without_resume() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
+        let mut worker = Worker::new(
+            minimal_manifest(),
+            Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient),
+            store.clone(),
+            WorkerWorkspaceContext::no_workspace(),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        worker.ensure_segment_head().await.unwrap();
+        worker.engine_mut().set_active_run_turn_count(Some(2));
+        worker
+            .persist_turn(0, &EngineRunExit::Yielded, None)
+            .await
+            .unwrap();
+        worker.last_run_interrupted = true;
+
+        worker.persist_compaction_run_failure(
+            "mid-run compaction failed: summary unavailable".into(),
+        );
+
+        let entries = store
+            .read_all(worker.session_id(), worker.segment_id())
+            .unwrap();
+        let yielded = entries
+            .iter()
+            .position(|entry| matches!(entry, LogEntry::RunYielded { .. }))
+            .unwrap();
+        let failed = entries
+            .iter()
+            .position(|entry| {
+                matches!(
+                    entry,
+                    LogEntry::RunErrored {
+                        interrupted: false,
+                        failure: Some(RunFailureKind::Compaction),
+                        ..
+                    }
+                )
+            })
+            .unwrap();
+        assert!(yielded < failed);
+        assert!(
+            !entries
+                .iter()
+                .any(|entry| matches!(entry, LogEntry::RunResumed { .. }))
+        );
+        assert!(!worker.last_run_interrupted);
+        assert_eq!(worker.engine().active_run_turn_count(), None);
+    }
+
+    #[tokio::test]
+    async fn intentional_cancel_commits_a_non_failure_terminal() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
+        let mut worker = Worker::new(
+            minimal_manifest(),
+            Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient),
+            store.clone(),
+            WorkerWorkspaceContext::no_workspace(),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        worker.ensure_segment_head().await.unwrap();
+        worker.engine_mut().set_active_run_turn_count(Some(1));
+
+        let result = worker
+            .handle_worker_result(
+                EngineRunExit::Interrupted(RunInterruptionReason::Cancelled),
+                0,
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(result, WorkerRunResult::Cancelled);
+        let entries = store
+            .read_all(worker.session_id(), worker.segment_id())
+            .unwrap();
+        assert!(
+            entries
+                .iter()
+                .any(|entry| matches!(entry, LogEntry::RunCancelled { .. }))
+        );
+        assert!(
+            !entries
+                .iter()
+                .any(|entry| matches!(entry, LogEntry::RunErrored { .. }))
+        );
+        assert!(!worker.last_run_interrupted);
+    }
+
     #[tokio::test]
     async fn fresh_run_clears_interrupted_budget_before_pre_run_compaction() {
         let dir = tempfile::tempdir().unwrap();
@@ -10242,7 +10505,7 @@ mod build_summary_prompt_tests {
         let mut worker = Worker::new(
             minimal_manifest(),
             engine,
-            store,
+            store.clone(),
             WorkerWorkspaceContext::no_workspace(),
             WorkerFilesystemAuthority::None,
             Scope::empty(),
@@ -10265,6 +10528,14 @@ mod build_summary_prompt_tests {
             item,
             Item::ToolResult { call_id, .. } if call_id == "call_pending"
         )));
+        let paused_entries = store
+            .read_all(worker.session_id(), worker.segment_id())
+            .unwrap();
+        assert!(
+            !paused_entries
+                .iter()
+                .any(|entry| matches!(entry, LogEntry::RunResumed { .. }))
+        );
 
         assert_eq!(worker.resume().await.unwrap(), WorkerRunResult::Finished);
         assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
@@ -10283,10 +10554,30 @@ mod build_summary_prompt_tests {
                 .count(),
             1
         );
+        let resumed_entries = store
+            .read_all(worker.session_id(), worker.segment_id())
+            .unwrap();
+        let resume_index = resumed_entries
+            .iter()
+            .position(|entry| {
+                matches!(
+                    entry,
+                    LogEntry::RunResumed {
+                        source: RunResumeSource::Pause,
+                        ..
+                    }
+                )
+            })
+            .expect("pause resume transition");
+        let terminal_index = resumed_entries
+            .iter()
+            .rposition(|entry| matches!(entry, LogEntry::RunCompleted { .. }))
+            .expect("resumed run terminal");
+        assert!(resume_index < terminal_index);
     }
 
     #[tokio::test]
-    async fn interrupted_result_terminalizes_orphan_before_run_completed() {
+    async fn cancelled_result_terminalizes_orphan_before_run_terminal() {
         let dir = tempfile::tempdir().unwrap();
         let manifest = minimal_manifest();
         let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
@@ -10395,7 +10686,9 @@ mod build_summary_prompt_tests {
             .position(|entry| {
                 matches!(
                     entry,
-                    LogEntry::RunCompleted { .. } | LogEntry::RunErrored { .. }
+                    LogEntry::RunCompleted { .. }
+                        | LogEntry::RunCancelled { .. }
+                        | LogEntry::RunErrored { .. }
                 )
             })
             .expect("durable final run status");
@@ -10961,7 +11254,8 @@ mod build_summary_prompt_tests {
             .commit_entry(LogEntry::RunErrored {
                 ts: segment_log::now_millis(),
                 interrupted: true,
-                message: "cancelled".to_string(),
+                message: "provider failed".to_string(),
+                failure: Some(RunFailureKind::Engine),
             })
             .unwrap();
         assert_eq!(
@@ -10980,6 +11274,33 @@ mod build_summary_prompt_tests {
             capture_handle.capture().unwrap().run_exit,
             CommittedRunExit::Finished
         );
+        worker
+            .commit_entry(LogEntry::Invoke {
+                ts: segment_log::now_millis(),
+                trigger: protocol::InvokeKind::UserSend,
+            })
+            .unwrap();
+        worker
+            .commit_entry(LogEntry::RunCancelled {
+                ts: segment_log::now_millis(),
+            })
+            .unwrap();
+        assert_eq!(
+            capture_handle.capture().unwrap().run_exit,
+            CommittedRunExit::Interrupted,
+            "a current cancellation must not inherit an earlier finished run"
+        );
+        worker
+            .commit_entry(LogEntry::RunResumed {
+                ts: segment_log::now_millis(),
+                source: RunResumeSource::Compaction,
+                active_run_turn_count: 1,
+            })
+            .unwrap();
+        assert_eq!(
+            capture_handle.capture().unwrap().run_exit,
+            CommittedRunExit::NonFinal
+        );
     }
 
     #[test]

crates/worker/tests/compact_events_test.rs

@@ -1123,7 +1123,55 @@ async fn request_threshold_compact_publishes_runtime_progress() {
 
     // Second run: pre_llm_request yields immediately, Engine returns
     // Yielded, handle_worker_result routes into do_compact_and_resume.
+    let source_segment = worker.segment_id();
+    let store = worker.store().clone();
     worker.run_text("second").await.unwrap();
+    let replacement_segment = worker.segment_id();
+    assert_ne!(replacement_segment, source_segment);
+
+    let source_entries = store.read_all(worker.session_id(), source_segment).unwrap();
+    assert!(source_entries.iter().any(|entry| matches!(
+        entry,
+        LogEntry::RunYielded {
+            reason: protocol::RunYieldReason::Compaction,
+            ..
+        }
+    )));
+    assert!(!source_entries.iter().any(|entry| matches!(
+        entry,
+        LogEntry::RunCompleted {
+            result: agen::EngineResult::Yielded,
+            ..
+        }
+    )));
+    let replacement_entries = store
+        .read_all(worker.session_id(), replacement_segment)
+        .unwrap();
+    let resumed = replacement_entries
+        .iter()
+        .position(|entry| {
+            matches!(
+                entry,
+                LogEntry::RunResumed {
+                    source: protocol::RunResumeSource::Compaction,
+                    ..
+                }
+            )
+        })
+        .expect("compaction resume transition");
+    let completed = replacement_entries
+        .iter()
+        .position(|entry| {
+            matches!(
+                entry,
+                LogEntry::RunCompleted {
+                    result: agen::EngineResult::Finished,
+                    ..
+                }
+            )
+        })
+        .expect("logical run completion");
+    assert!(resumed < completed);
 
     let events = drain(&mut rx);
     assert!(events.iter().any(|event| matches!(
@@ -1461,6 +1509,21 @@ async fn compacted_context_above_request_threshold_fails_before_provider_request
         3,
         "the provider must receive only the seed and compaction requests"
     );
+    let entries = worker
+        .store()
+        .read_all(worker.session_id(), worker.segment_id())
+        .unwrap();
+    assert!(entries.iter().any(|entry| {
+        matches!(
+            entry,
+            LogEntry::RunErrored {
+                interrupted: false,
+                failure: Some(protocol::RunFailureKind::Compaction),
+                message,
+                ..
+            } if message.contains("mid-run compaction could not continue")
+        )
+    }));
 }
 
 #[tokio::test]

crates/worker/tests/controller_test.rs

@@ -3693,6 +3693,71 @@ async fn empty_turn_rollback_removes_only_the_most_recent_turn() {
     );
 }
 
+#[tokio::test]
+async fn cancel_after_assistant_output_is_a_non_failure_terminal() {
+    let client = MockClient::sequential(vec![MockResponse::Hang(vec![
+        LlmEvent::text_block_start(0),
+        LlmEvent::text_delta(0, "committed before cancel"),
+        LlmEvent::text_block_stop(0, None),
+    ])]);
+    let worker = make_worker(client).await;
+    let handle = spawn_controller(worker).await;
+    let mut rx = handle.subscribe();
+
+    handle
+        .send(Method::submit_text(
+            protocol::new_submission_request_id(),
+            "cancel this active turn",
+        ))
+        .await
+        .unwrap();
+    assert!(
+        drain_until(
+            &mut rx,
+            std::time::Duration::from_secs(2),
+            |event| matches!(event, Event::TextDone { .. })
+        )
+        .await,
+        "assistant output should be durable before cancellation"
+    );
+    handle
+        .send(Method::Cancel {
+            command: worker_command(&handle),
+        })
+        .await
+        .unwrap();
+
+    assert!(
+        drain_until(
+            &mut rx,
+            std::time::Duration::from_secs(2),
+            |event| matches!(
+                event,
+                Event::RunEnd {
+                    result: protocol::RunResult::Cancelled
+                }
+            )
+        )
+        .await,
+        "intentional cancellation must emit RunEnd::Cancelled, not Event::Error"
+    );
+    wait_for_status(&handle, WorkerStatus::Idle).await;
+
+    let (entries, _) = handle.sink.subscribe_with_snapshot();
+    assert!(
+        entries
+            .iter()
+            .any(|entry| matches!(entry, LogEntry::RunCancelled { .. })),
+        "cancelled run must keep a durable non-failure terminal: {entries:?}"
+    );
+    assert!(
+        !entries
+            .iter()
+            .any(|entry| matches!(entry, LogEntry::RunErrored { .. })),
+        "intentional cancellation must not be restored as a run error: {entries:?}"
+    );
+}
+
 #[tokio::test]
 async fn pause_after_assistant_token_does_not_rollback() {
     let client = MockClient::sequential(vec![MockResponse::Hang(vec![

web/workspace/src/lib/generated/protocol.ts

@@ -40,7 +40,7 @@ export type TurnResult = "finished" | "paused";
 
 export type InvokeKind = "user_send" | "notify" | "worker_event" | "system_reminder" | "wakeup";
 
-export type RunResult = "finished" | "paused" | "limit_reached" | "rolled_back";
+export type RunResult = "finished" | "paused" | "limit_reached" | "cancelled" | "rolled_back";
 
 export type ToolResultDisposition = "success" | "error" | "interrupted" | "cancelled" | "outcome_unknown";
 
@@ -136,7 +136,13 @@ export type SessionToolAttachment = { media_type: string,
  */
 data_base64: string, };
 
-export type SessionSnapshotEntryData = { "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_error", message: string, };
+export type RunYieldReason = "compaction";
+
+export type RunResumeSource = "compaction" | "pause";
+
+export type RunFailureKind = "engine" | "compaction";
+
+export type SessionSnapshotEntryData = { "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_yielded", reason: RunYieldReason, active_run_turn_count: number, } | { "kind": "run_resumed", source: RunResumeSource, active_run_turn_count: number, } | { "kind": "run_cancelled" } | { "kind": "run_error", message: string, failure?: RunFailureKind | null, };
 
 export type SessionSnapshotEntry = {
 /**
@@ -147,7 +153,7 @@ entry_id: string,
 /**
  * Timestamp copied from the durable log record that commits this entry.
  */
-timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array<string>, } & ({ "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_error", message: string, });
+timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array<string>, } & ({ "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_yielded", reason: RunYieldReason, active_run_turn_count: number, } | { "kind": "run_resumed", source: RunResumeSource, active_run_turn_count: number, } | { "kind": "run_cancelled" } | { "kind": "run_error", message: string, failure?: RunFailureKind | null, });
 
 export type SessionConversationTurn = { turn_id: string, entries: Array<SessionSnapshotEntry>, };
 
@@ -340,7 +346,7 @@ export type SubscriptionFrame = { protocol_version: number, } & ({ "frame": "req
 
 export type Method = { "method": "submit", "params": { submission_request_id: string, input: Array<Segment>, } } | { "method": "notify", "params": { notification_request_id: string, message: string, } } | { "method": "worker_event", "params": WorkerEvent } | { "method": "list_pending_submissions" } | { "method": "cancel_pending_submission", "params": { submission_id: string, expected_revision: number, } } | { "method": "clear_pending_submissions", "params": { expected_revision: number, } } | { "method": "continue_pending", "params": { expected_revision: number, expected_head_id: string, } } | { "method": "resume", "params": { command: WorkerCommandEnvelope, } } | { "method": "cancel", "params": { command: WorkerCommandEnvelope, } } | { "method": "pause", "params": { command: WorkerCommandEnvelope, } } | { "method": "compact", "params": { command: WorkerCommandEnvelope, } } | { "method": "list_rewind_targets" } | { "method": "rewind_to", "params": { target: RewindTargetId, expected_head_entries: number, } } | { "method": "shutdown", "params": { command: WorkerCommandEnvelope, } } | { "method": "list_completions", "params": { kind: CompletionKind, prefix: string, } } | { "method": "list_workers" } | { "method": "restore_worker", "params": { name: string, } } | { "method": "register_peer", "params": { name: string, } };
 
-export type Event = { "event": "submission_accepted", "data": { submission_request_id: string, submission_id: string, disposition: SubmissionDisposition, } } | { "event": "submission_rejected", "data": { submission_request_id: string, message: string, } } | { "event": "pending_submissions_changed", "data": { pending: PendingSubmissionsSnapshot, } } | { "event": "user_message", "data": {
+export type Event = { "event": "submission_accepted", "data": { submission_request_id: string, submission_id: string, disposition: SubmissionDisposition, } } | { "event": "submission_rejected", "data": { submission_request_id: string, message: string, } } | { "event": "notification_accepted", "data": { notification_request_id: string, } } | { "event": "notification_rejected", "data": { notification_request_id: string, message: string, } } | { "event": "pending_submissions_changed", "data": { pending: PendingSubmissionsSnapshot, } } | { "event": "user_message", "data": {
 /**
  * Stable durable history identity. Older mixed-version producers omit it.
  */

web/workspace/src/lib/workspace/console/model.test.ts

@@ -132,12 +132,22 @@ function canonicalSession(logEntries: unknown[]): Event extends {
         });
         break;
       }
+      case "run_yielded":
+      case "run_resumed":
+      case "run_cancelled":
+        entries.push({
+          entry_id: `legacy-test-${sequence++}`,
+          provenance: "legacy_unknown",
+          ...entry,
+        });
+        break;
       case "run_errored":
         entries.push({
           entry_id: `legacy-test-${sequence++}`,
           provenance: "legacy_unknown",
           kind: "run_error",
           message: entry["message"] ?? "Worker run failed.",
+          failure: entry["failure"],
         });
         break;
     }
@@ -291,6 +301,138 @@ Deno.test("Worker state events and acknowledgements replace the full state", ()
   );
 });
 
+Deno.test("durable compaction failure reconciles live alert and error in both views", () => {
+  const message = "mid-run compaction failed: summary unavailable";
+  const transition = (
+    entryId: string,
+    data: Record<string, unknown>,
+  ): Event => ({
+    event: "session_entry_committed",
+    data: {
+      entry: {
+        entry_id: entryId,
+        timestamp: 10,
+        provenance: "legacy_unknown",
+        ...data,
+      },
+    },
+  } as Event);
+  const projection = projectConsole([
+    {
+      eventId: "yielded",
+      event: transition("yielded-entry", {
+        kind: "run_yielded",
+        reason: "compaction",
+        active_run_turn_count: 2,
+      }),
+    },
+    {
+      eventId: "alert",
+      event: {
+        event: "alert",
+        data: {
+          level: "error",
+          source: "compactor",
+          message,
+          timestamp_ms: 10,
+        },
+      },
+    },
+    {
+      eventId: "failed",
+      event: transition("failed-entry", {
+        kind: "run_error",
+        message,
+        failure: "compaction",
+      }),
+    },
+    {
+      eventId: "controller-error",
+      event: {
+        event: "error",
+        data: { code: "internal", message: "summary unavailable" },
+      },
+    },
+  ]);
+
+  const errors = projection.lines.filter((line) => line.kind === "error");
+  assertEquals(errors.length, 1);
+  assertEquals(errors[0].title, "Compaction failed");
+  assertEquals(errors[0].body, message);
+  assertEquals(errors[0].entryId, "failed-entry");
+  assertEquals(projectConsoleLines(projection.lines, "normal"), errors);
+  assertEquals(projectConsoleLines(projection.lines, "overview"), errors);
+});
+
+Deno.test("snapshot restores compaction terminal without reviving transition progress", () => {
+  const projection = projectConsole([{
+    eventId: "snapshot",
+    event: snapshotEvent("/repo", [
+      {
+        kind: "run_yielded",
+        timestamp: 1,
+        reason: "compaction",
+        active_run_turn_count: 2,
+      },
+      {
+        kind: "run_resumed",
+        timestamp: 2,
+        source: "compaction",
+        active_run_turn_count: 2,
+      },
+      {
+        kind: "run_errored",
+        timestamp: 3,
+        interrupted: false,
+        message: "mid-run compaction failed: summary unavailable",
+        failure: "compaction",
+      },
+    ]),
+  }]);
+
+  assertEquals(projection.compaction, null);
+  assertEquals(projection.lines.length, 1);
+  assertEquals(projection.lines[0].title, "Compaction failed");
+  assertEquals(projection.lines[0].kind, "error");
+});
+
+Deno.test("intentional cancellation stays non-error in live and snapshot projections", () => {
+  const live = projectConsole([
+    {
+      eventId: "cancelled-terminal",
+      event: {
+        event: "session_entry_committed",
+        data: {
+          entry: {
+            entry_id: "cancelled-entry",
+            timestamp: 10,
+            provenance: "legacy_unknown",
+            kind: "run_cancelled",
+          },
+        },
+      } as Event,
+    },
+    {
+      eventId: "cancelled-run-end",
+      event: {
+        event: "run_end",
+        data: { result: "cancelled" },
+      } as Event,
+    },
+  ]);
+  assertEquals(live.lines.some((line) => line.kind === "error"), false);
+
+  const restored = projectConsole([{
+    eventId: "snapshot",
+    event: snapshotEvent("/repo", [{
+      kind: "run_cancelled",
+      timestamp: 10,
+    }]),
+  }]);
+  assertEquals(restored.lines.some((line) => line.kind === "error"), false);
+  assertEquals(restored.compaction, null);
+});
+
 Deno.test("snapshot replaces a live error with one durable run_errored row", () => {
   const projector = createConsoleProjector();
   let projection = projector.append([

web/workspace/src/lib/workspace/console/model.ts

@@ -1002,6 +1002,7 @@ export function applyProtocolEvent(
       next.usage = usageText(event.data);
       break;
     case "error":
+      if (hasDurableRunFailure(next, event.data.message)) break;
       next.lines.push(
         line(
           envelope.eventId,
@@ -1264,12 +1265,55 @@ function upsertStatusLine(
   }
 }
 
+function runFailureMessagesMatch(left: string, right: string): boolean {
+  return left.length > 0 && right.length > 0 &&
+    (left === right || left.endsWith(right) || right.endsWith(left));
+}
+
+function isDurableRunFailure(line: ConsoleLine): boolean {
+  return line.kind === "error" &&
+    (line.title === "Run error" || line.title === "Compaction failed");
+}
+
+function hasDurableRunFailure(
+  projection: ConsoleProjection,
+  message: string,
+): boolean {
+  return projection.lines.some((line) =>
+    isDurableRunFailure(line) && runFailureMessagesMatch(line.body, message)
+  );
+}
+
+function appendDurableRunFailure(
+  projection: ConsoleProjection,
+  eventId: string,
+  message: string,
+  failure: string | undefined,
+): void {
+  projection.lines = projection.lines.filter((line) => {
+    if (isDurableRunFailure(line)) return true;
+    return !(line.kind === "error" && runFailureMessagesMatch(line.body, message));
+  });
+  projection.lines.push(
+    line(
+      eventId,
+      "error",
+      failure === "compaction" ? "Compaction failed" : "Run error",
+      message,
+      undefined,
+      false,
+      true,
+    ),
+  );
+}
+
 function appendAlertLine(
   projection: ConsoleProjection,
   eventId: string,
   alert: Alert,
 ): void {
   const isError = alert.level === "error";
+  if (isError && hasDurableRunFailure(projection, alert.message)) return;
   projection.lines.push(
     line(
       eventId,
@@ -2144,17 +2188,18 @@ function applySessionEntry(
       applyTaskSystemItem(projection, item);
       break;
     }
+    case "run_yielded":
+    case "run_resumed":
+    case "run_cancelled":
+      // Durable logical-Run transitions and intentional cancellation are
+      // replayable lifecycle state, not errors or Console progress rows.
+      break;
     case "run_error":
-      projection.lines.push(
+      appendDurableRunFailure(
-        line(
+        projection,
-          eventId,
+        eventId,
-          "error",
+        stringField(value, "message") ?? "Worker run failed.",
-          "Run error",
+        stringField(value, "failure"),
-          stringField(value, "message") ?? "Worker run failed.",
-          undefined,
-          false,
-          true,
-        ),
       );
       break;
     default: