diff --git a/apps/staged/src-tauri/src/lib.rs b/apps/staged/src-tauri/src/lib.rs
index ec2c14e27..ebcbc16ae 100644
--- a/apps/staged/src-tauri/src/lib.rs
+++ b/apps/staged/src-tauri/src/lib.rs
@@ -31,6 +31,7 @@ pub mod project_mcp;
 pub mod prs;
 pub mod review_commands;
 pub mod session_commands;
+pub mod session_completion;
 pub mod session_runner;
 pub mod shell_env;
 pub mod store;
@@ -2333,6 +2334,7 @@ pub fn run() {
             // Sessions
             session_commands::discover_acp_providers,
             session_commands::discover_acp_config,
+            session_commands::get_active_sessions,
             session_commands::get_session,
             session_commands::get_session_messages,
             session_commands::get_session_messages_since,
diff --git a/apps/staged/src-tauri/src/prs.rs b/apps/staged/src-tauri/src/prs.rs
index f747761b0..5976f9bb8 100644
--- a/apps/staged/src-tauri/src/prs.rs
+++ b/apps/staged/src-tauri/src/prs.rs
@@ -172,7 +172,11 @@ fn start_pipeline_for_branch(
 ) -> Result<String, String> {
     let pipeline = PipelineExecution::from_steps(&steps);
 
-    let mut session = store::Session::new_running(prompt, &ctx.working_dir);
+    // pr/push pipeline sessions link no artifact row, so the session row
+    // itself carries the branch linkage — busy-state snapshots and terminal
+    // completion side-effects resolve the branch from it.
+    let mut session =
+        store::Session::new_running(prompt, &ctx.working_dir).with_branch(&ctx.branch.id);
     if let Some(ref p) = provider {
         session = session.with_provider(p);
     }
@@ -1395,7 +1399,17 @@ pub async fn refresh_pr_status(
     branch_id: String,
 ) -> Result<(), String> {
     let store = get_store(&store)?;
+    refresh_pr_status_impl(store, app_handle, branch_id).await
+}
 
+/// Core implementation of the single-branch PR status refresh, shared by the
+/// `refresh_pr_status` command, the web-mode `dispatch()` arm, and the
+/// session-completion side effects that run after a PR session finishes.
+pub(crate) async fn refresh_pr_status_impl(
+    store: Arc<Store>,
+    app_handle: tauri::AppHandle,
+    branch_id: String,
+) -> Result<(), String> {
     let branch = store
         .get_branch(&branch_id)
         .map_err(|e| e.to_string())?
@@ -1675,12 +1689,22 @@ pub fn clear_branch_pr_status(
     branch_id: String,
 ) -> Result<(), String> {
     let store = get_store(&store)?;
+    clear_branch_pr_status_impl(&store, &app_handle, &branch_id)
+}
 
+/// Core implementation shared by the `clear_branch_pr_status` command, the
+/// web-mode `dispatch()` arm, and the session-completion side effects that
+/// run after a push session finishes.
+pub(crate) fn clear_branch_pr_status_impl(
+    store: &Store,
+    app_handle: &tauri::AppHandle,
+    branch_id: &str,
+) -> Result<(), String> {
     store
-        .update_branch_pr_status(&branch_id, None, None, None, None, None, None, None, None)
+        .update_branch_pr_status(branch_id, None, None, None, None, None, None, None, None)
         .map_err(|e| e.to_string())?;
 
-    crate::web_server::emit_to_all(&app_handle, "pr-status-cleared", &branch_id);
+    crate::web_server::emit_to_all(app_handle, "pr-status-cleared", branch_id);
 
     Ok(())
 }
diff --git a/apps/staged/src-tauri/src/session_commands.rs b/apps/staged/src-tauri/src/session_commands.rs
index b108a15e6..af0266512 100644
--- a/apps/staged/src-tauri/src/session_commands.rs
+++ b/apps/staged/src-tauri/src/session_commands.rs
@@ -901,6 +901,101 @@ pub async fn discover_acp_config(
 // Read-only queries (used by frontend polling)
 // =============================================================================
 
+/// A running or queued session projected to its branch/project context.
+///
+/// This is the busy-state snapshot clients hydrate from on load or reconnect.
+/// It carries the same discriminators as `SessionStatusEvent` so the snapshot
+/// and the `session-status-changed` delta stream describe sessions identically.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ActiveSessionInfo {
+    pub session_id: String,
+    pub project_id: Option<String>,
+    pub branch_id: Option<String>,
+    pub session_type: Option<String>,
+    pub status: String,
+    pub is_auto_review: bool,
+}
+
+/// Project one session to branch/project context and session type.
+///
+/// Keep the derivation aligned with `resume_session_for_store`, which stamps
+/// the same fields onto the `"running"` event for resumed sessions. Pipeline
+/// sessions (pr/push) link no artifact, so their branch comes from the
+/// session row's own `branch_id` and their type falls back to prompt
+/// inference.
+fn project_active_session(store: &Store, session: &store::Session) -> ActiveSessionInfo {
+    let project_note = store
+        .get_project_note_by_session(&session.id)
+        .ok()
+        .flatten();
+    let linked_commit = store.get_commit_by_session(&session.id).ok().flatten();
+    let linked_note = store.get_note_by_session(&session.id).ok().flatten();
+    let linked_review = store.get_review_by_session(&session.id).ok().flatten();
+
+    let session_type = if project_note.is_some() {
+        // Project notes and branch notes intentionally share the "note"
+        // session type because the frontend only needs a single "note work is
+        // running" signal for project-level activity indicators.
+        Some("note".to_string())
+    } else if linked_commit.is_some() {
+        Some("commit".to_string())
+    } else if linked_note.is_some() {
+        Some("note".to_string())
+    } else if linked_review.is_some() {
+        Some("review".to_string())
+    } else {
+        infer_branch_resume_session_type(&session.prompt).map(str::to_string)
+    };
+
+    let branch_id = linked_commit
+        .as_ref()
+        .map(|commit| commit.branch_id.clone())
+        .or_else(|| linked_note.as_ref().map(|note| note.branch_id.clone()))
+        .or_else(|| {
+            linked_review
+                .as_ref()
+                .map(|review| review.branch_id.clone())
+        })
+        .or_else(|| session.branch_id.clone());
+
+    let project_id = if let Some(note) = &project_note {
+        Some(note.project_id.clone())
+    } else {
+        branch_id
+            .as_deref()
+            .and_then(|bid| store.get_branch(bid).ok().flatten())
+            .map(|branch| branch.project_id)
+    };
+
+    ActiveSessionInfo {
+        session_id: session.id.clone(),
+        project_id,
+        branch_id,
+        session_type,
+        status: session.status.as_str().to_string(),
+        is_auto_review: linked_review.map(|review| review.is_auto).unwrap_or(false),
+    }
+}
+
+/// Synchronous body of [`get_active_sessions`], shared with the web-mode
+/// `dispatch()` arm.
+pub(crate) fn get_active_sessions_impl(store: &Store) -> Result<Vec<ActiveSessionInfo>, String> {
+    let sessions = store.get_active_sessions().map_err(|e| e.to_string())?;
+    Ok(sessions
+        .iter()
+        .map(|session| project_active_session(store, session))
+        .collect())
+}
+
+#[tauri::command]
+pub fn get_active_sessions(
+    store: tauri::State<'_, Mutex<Option<Arc<Store>>>>,
+) -> Result<Vec<ActiveSessionInfo>, String> {
+    let store = get_store(&store)?;
+    get_active_sessions_impl(&store)
+}
+
 #[tauri::command]
 pub fn get_session(
     store: tauri::State<'_, Mutex<Option<Arc<Store>>>>,
@@ -1131,6 +1226,10 @@ pub(crate) async fn resume_session_for_store(
         store.get_branch(&note.branch_id).ok().flatten()
     } else if let Some(review) = &linked_review {
         store.get_branch(&review.branch_id).ok().flatten()
+    } else if let Some(session_branch_id) = &session.branch_id {
+        // Pipeline (pr/push) sessions link no artifact; the session row
+        // carries their branch.
+        store.get_branch(session_branch_id).ok().flatten()
     } else {
         None
     };
@@ -1436,13 +1535,17 @@ pub(crate) async fn drain_queued_message_for_session(
 }
 
 pub(crate) fn infer_branch_resume_session_type(prompt: &str) -> Option<&'static str> {
-    // Keep these checks aligned with the action prompts built in `prs.rs`.
-    if prompt.contains("Create a draft pull request for the current branch.")
-        || prompt.contains("Create a pull request for the current branch.")
+    // Keep these checks aligned with the session prompts built in `prs.rs`.
+    // The needles are period-free prefixes so they match both the stored
+    // pipeline session prompts ("Create a pull request for the current
+    // branch") and the AI-handoff prompt sentences ("... for the current
+    // branch.").
+    if prompt.contains("Create a draft pull request for the current branch")
+        || prompt.contains("Create a pull request for the current branch")
     {
         Some("pr")
-    } else if prompt.contains("Push the current branch to the remote using force-with-lease.")
-        || prompt.contains("Push the current branch to the remote.")
+    } else if prompt.contains("Push the current branch to the remote")
+        || prompt.contains("Force push the current branch to the remote")
     {
         Some("push")
     } else {
@@ -5322,6 +5425,161 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn active_sessions_snapshot_projects_branch_sessions() {
+        let (store, branch) = setup_branch_store();
+        let commit_session =
+            create_branch_commit_session(&store, &branch.id, store::SessionStatus::Running);
+        let note_session =
+            create_branch_note_session(&store, &branch.id, store::SessionStatus::Queued);
+        let review_session =
+            create_branch_review_session(&store, &branch.id, store::SessionStatus::Running);
+
+        let snapshot = get_active_sessions_impl(&store).unwrap();
+        assert_eq!(snapshot.len(), 3);
+
+        let info_for = |session_id: &str| {
+            snapshot
+                .iter()
+                .find(|info| info.session_id == session_id)
+                .unwrap()
+        };
+
+        let commit_info = info_for(&commit_session.id);
+        assert_eq!(commit_info.branch_id.as_deref(), Some(branch.id.as_str()));
+        assert_eq!(
+            commit_info.project_id.as_deref(),
+            Some(branch.project_id.as_str())
+        );
+        assert_eq!(commit_info.session_type.as_deref(), Some("commit"));
+        assert_eq!(commit_info.status, "running");
+        assert!(!commit_info.is_auto_review);
+
+        let note_info = info_for(&note_session.id);
+        assert_eq!(note_info.session_type.as_deref(), Some("note"));
+        assert_eq!(note_info.status, "queued");
+
+        let review_info = info_for(&review_session.id);
+        assert_eq!(review_info.session_type.as_deref(), Some("review"));
+        assert!(!review_info.is_auto_review);
+    }
+
+    #[test]
+    fn active_sessions_snapshot_marks_auto_reviews() {
+        let (store, branch) = setup_branch_store();
+        let (session, _review) =
+            create_auto_review(&store, &branch.id, store::SessionStatus::Running);
+
+        let snapshot = get_active_sessions_impl(&store).unwrap();
+        assert_eq!(snapshot.len(), 1);
+        assert_eq!(snapshot[0].session_id, session.id);
+        assert_eq!(snapshot[0].session_type.as_deref(), Some("review"));
+        assert!(snapshot[0].is_auto_review);
+    }
+
+    #[test]
+    fn active_sessions_snapshot_projects_project_note_sessions() {
+        let store = Arc::new(Store::in_memory().unwrap());
+        let project = store::Project::new("test-owner/test-repo");
+        store.create_project(&project).unwrap();
+        let session =
+            create_session_with_status(&store, "project note", store::SessionStatus::Running);
+        let note = store::ProjectNote::new(&project.id, "note", "").with_session(&session.id);
+        store.create_project_note(&note).unwrap();
+
+        let snapshot = get_active_sessions_impl(&store).unwrap();
+        assert_eq!(snapshot.len(), 1);
+        assert_eq!(snapshot[0].project_id.as_deref(), Some(project.id.as_str()));
+        assert_eq!(snapshot[0].branch_id, None);
+        assert_eq!(snapshot[0].session_type.as_deref(), Some("note"));
+    }
+
+    #[test]
+    fn active_sessions_snapshot_infers_pipeline_session_type_from_prompt() {
+        let store = Arc::new(Store::in_memory().unwrap());
+        // pr/push pipeline sessions link no artifact, so the type falls back
+        // to the same prompt inference the resume path uses. Use the exact
+        // prompts prs.rs stores on pipeline sessions.
+        let pr_session = create_session_with_status(
+            &store,
+            "Create a pull request for the current branch",
+            store::SessionStatus::Running,
+        );
+        let push_session = create_session_with_status(
+            &store,
+            "Push the current branch to the remote with a normal push. If the push fails for a recoverable reason, diagnose and fix it, then retry with a normal push. Do not force push.",
+            store::SessionStatus::Running,
+        );
+        let force_push_session = create_session_with_status(
+            &store,
+            "Force push the current branch to the remote",
+            store::SessionStatus::Running,
+        );
+        let unknown_session =
+            create_session_with_status(&store, "anything else", store::SessionStatus::Running);
+
+        let snapshot = get_active_sessions_impl(&store).unwrap();
+        assert_eq!(snapshot.len(), 4);
+
+        let info_for = |session_id: &str| {
+            snapshot
+                .iter()
+                .find(|info| info.session_id == session_id)
+                .unwrap()
+        };
+
+        let pr_info = info_for(&pr_session.id);
+        assert_eq!(pr_info.session_type.as_deref(), Some("pr"));
+        assert_eq!(pr_info.branch_id, None);
+        assert_eq!(pr_info.project_id, None);
+
+        assert_eq!(
+            info_for(&push_session.id).session_type.as_deref(),
+            Some("push")
+        );
+        assert_eq!(
+            info_for(&force_push_session.id).session_type.as_deref(),
+            Some("push")
+        );
+        assert_eq!(info_for(&unknown_session.id).session_type, None);
+    }
+
+    #[test]
+    fn active_sessions_snapshot_resolves_pipeline_sessions_via_session_branch_id() {
+        let (store, branch) = setup_branch_store();
+        // Pipeline (pr/push) sessions carry their branch on the session row
+        // instead of an artifact; the snapshot must resolve branch and
+        // project from it.
+        let mut session = store::Session::new_running(
+            "Create a pull request for the current branch",
+            Path::new("/tmp"),
+        )
+        .with_branch(&branch.id);
+        session.pipeline = Some(store::PipelineExecution::from_steps(&[]));
+        store.create_session(&session).unwrap();
+
+        let snapshot = get_active_sessions_impl(&store).unwrap();
+        assert_eq!(snapshot.len(), 1);
+        assert_eq!(snapshot[0].branch_id.as_deref(), Some(branch.id.as_str()));
+        assert_eq!(
+            snapshot[0].project_id.as_deref(),
+            Some(branch.project_id.as_str())
+        );
+        assert_eq!(snapshot[0].session_type.as_deref(), Some("pr"));
+        assert!(!snapshot[0].is_auto_review);
+    }
+
+    #[test]
+    fn active_sessions_snapshot_excludes_terminal_sessions() {
+        let (store, branch) = setup_branch_store();
+        let session = create_branch_note_session(&store, &branch.id, store::SessionStatus::Running);
+        store
+            .update_session_status(&session.id, store::SessionStatus::Completed, None, None)
+            .unwrap();
+
+        assert!(get_active_sessions_impl(&store).unwrap().is_empty());
+    }
+
     #[test]
     fn branch_start_decision_queues_local_branch_without_workdir() {
         let (store, branch) = setup_branch_store();
@@ -6478,6 +6736,11 @@ mod tests {
             infer_branch_resume_session_type("Create a draft pull request for the current branch."),
             Some("pr")
         );
+        // The exact prompt prs.rs stores on PR pipeline sessions (no period).
+        assert_eq!(
+            infer_branch_resume_session_type("Create a pull request for the current branch"),
+            Some("pr")
+        );
     }
 
     #[test]
@@ -6492,6 +6755,17 @@ mod tests {
             ),
             Some("push")
         );
+        // The exact prompts prs.rs stores on push pipeline sessions.
+        assert_eq!(
+            infer_branch_resume_session_type(
+                "Push the current branch to the remote with a normal push. If the push fails for a recoverable reason, diagnose and fix it, then retry with a normal push. Do not force push."
+            ),
+            Some("push")
+        );
+        assert_eq!(
+            infer_branch_resume_session_type("Force push the current branch to the remote"),
+            Some("push")
+        );
     }
 
     #[test]
diff --git a/apps/staged/src-tauri/src/session_completion.rs b/apps/staged/src-tauri/src/session_completion.rs
new file mode 100644
index 000000000..b551a4ba4
--- /dev/null
+++ b/apps/staged/src-tauri/src/session_completion.rs
@@ -0,0 +1,752 @@
+//! Server-side completion side effects for pipeline (pr/push) sessions.
+//!
+//! When a PR or push session reaches its terminal transition, the session
+//! runner calls [`run_completion_side_effects`] from whichever thread won the
+//! `transition_from_running` write. It parses the session's outcome (PR URL
+//! from the transcript / pipeline step outputs, push result from the
+//! non-fast-forward markers), persists it (branch PR number, cleared PR
+//! status), and emits the `pr-created` / `push-completed` domain events —
+//! before the terminal `session-status-changed` event, so clients can render
+//! outcomes from the ordered event stream alone.
+//!
+//! This used to live in the frontend completion handlers, where every
+//! connected client raced to perform the same writes; frontends are now
+//! idempotent renderers of these events.
+
+use std::sync::Arc;
+
+use serde::Serialize;
+use tauri::AppHandle;
+
+use crate::store::{
+    MessageRole, PipelineExecution, Session, SessionMessage, SessionStatus, StepStatus, Store,
+};
+
+// =============================================================================
+// Domain events
+// =============================================================================
+
+/// Emitted when a completed PR session produced a pull request.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PrCreatedEvent {
+    pub branch_id: String,
+    pub session_id: String,
+    pub pr_url: String,
+    pub pr_number: u64,
+}
+
+/// Emitted when a push session completes, carrying the classified outcome.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PushCompletedEvent {
+    pub branch_id: String,
+    pub session_id: String,
+    pub outcome: PushOutcome,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub enum PushOutcome {
+    Succeeded,
+    RejectedNonFastForward,
+}
+
+/// What a completed pr/push session resolved to. Pure decision value so the
+/// parsing/classification can be tested without an app handle.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum CompletionEffect {
+    PrCreated { pr_url: String, pr_number: u64 },
+    PrUrlMissing,
+    PushCompleted { outcome: PushOutcome },
+}
+
+// =============================================================================
+// Entry point
+// =============================================================================
+
+/// Run completion side effects for a session that just reached a terminal
+/// state.
+///
+/// Callers must only invoke this after winning the `transition_from_running`
+/// write, so exactly one thread performs the side effects, and before emitting
+/// the terminal `session-status-changed` event, so the domain events reach
+/// clients first.
+///
+/// Only sessions launched via a pipeline are considered (pr/push sessions are
+/// always pipeline sessions); their kind is inferred from the stored prompt
+/// exactly like the resume path and busy-state snapshot do. Non-completed
+/// terminal states have no outcome to parse — clients render those directly
+/// from the terminal status event.
+///
+/// Outcomes are delivered at most once per session, recorded by the
+/// `completion_effects_at` marker: a resumed pr/push session completes again,
+/// but its pipeline and transcript still describe the *original* run, so
+/// re-evaluating them would re-emit `pr-created` or — destructively — re-clear
+/// the branch's PR status. See [`pending_completion_effect`] for what the
+/// marker does and doesn't claim.
+pub fn run_completion_side_effects(
+    store: &Arc<Store>,
+    app_handle: &AppHandle,
+    session_id: &str,
+    branch_id: Option<&str>,
+    status: &str,
+) {
+    // Cheap pre-check so ordinary terminal transitions don't read the session
+    // row; `pending_completion_effect` re-applies the gate as part of the
+    // decision it owns.
+    if status != "completed" {
+        return;
+    }
+
+    let session = match store.get_session(session_id) {
+        Ok(Some(session)) => session,
+        Ok(None) => return,
+        Err(e) => {
+            log::error!("Completion side effects: failed to load session {session_id}: {e}");
+            return;
+        }
+    };
+    let Some((kind, effect)) = pending_completion_effect(&session, status, || {
+        store.get_session_messages(session_id).unwrap_or_default()
+    }) else {
+        return;
+    };
+
+    let Some(branch_id) = branch_id
+        .map(str::to_string)
+        .or_else(|| session.branch_id.clone())
+    else {
+        log::warn!(
+            "Completion side effects: {kind} session {session_id} has no branch attribution"
+        );
+        return;
+    };
+
+    if should_record_effects(&effect) {
+        // Mark *before* emitting/persisting. Dying in between loses one
+        // emission, which existing recovery already covers (`recover_branch_pr`
+        // re-derives PR numbers, the PR refresh loop repopulates status,
+        // clients keep a read-only fallback classification). Dying the other
+        // way round would let a later resume replay the destructive push
+        // re-clear — the bug this marker exists to prevent. A failed marker
+        // write is logged but doesn't suppress the events: that leaves today's
+        // status quo rather than dropping a real outcome.
+        if let Err(e) = store.mark_completion_effects_ran(session_id) {
+            log::error!(
+                "Failed to mark completion effects for {kind} session {session_id} as delivered: {e}"
+            );
+        }
+    }
+
+    match effect {
+        CompletionEffect::PrCreated { pr_url, pr_number } => {
+            // Persist first so any refresh triggered by the event finds the
+            // number. A failed write is logged but doesn't suppress the event:
+            // the PR exists on GitHub, and `recover_branch_pr` re-derives the
+            // number later.
+            if let Err(e) = store.update_branch_pr_number(&branch_id, Some(pr_number)) {
+                log::error!(
+                    "Failed to persist PR #{pr_number} for branch {branch_id} after session {session_id}: {e}"
+                );
+            }
+            crate::web_server::emit_to_all(
+                app_handle,
+                "pr-created",
+                PrCreatedEvent {
+                    branch_id: branch_id.clone(),
+                    session_id: session_id.to_string(),
+                    pr_url,
+                    pr_number,
+                },
+            );
+
+            // Fetch checks/mergeability in the background; results arrive via
+            // the existing `pr-status-changed` event.
+            let store = Arc::clone(store);
+            let app_handle = app_handle.clone();
+            tauri::async_runtime::spawn(async move {
+                if let Err(e) =
+                    crate::prs::refresh_pr_status_impl(store, app_handle, branch_id.clone()).await
+                {
+                    log::warn!(
+                        "Failed to refresh PR status for branch {branch_id} after PR creation: {e}"
+                    );
+                }
+            });
+        }
+        CompletionEffect::PrUrlMissing => {
+            // No event: clients infer "completed but no PR URL" from a
+            // terminal status event that wasn't preceded by `pr-created`.
+            log::warn!("PR session {session_id} completed but no PR URL was found in the output");
+        }
+        CompletionEffect::PushCompleted { outcome } => {
+            if outcome == PushOutcome::Succeeded {
+                // The old PR head/checks no longer describe the branch.
+                if let Err(e) =
+                    crate::prs::clear_branch_pr_status_impl(store, app_handle, &branch_id)
+                {
+                    log::warn!("Failed to clear PR status for branch {branch_id} after push: {e}");
+                }
+            }
+            crate::web_server::emit_to_all(
+                app_handle,
+                "push-completed",
+                PushCompletedEvent {
+                    branch_id,
+                    session_id: session_id.to_string(),
+                    outcome,
+                },
+            );
+        }
+    }
+}
+
+/// Decide what outcome, if any, a session that just reached a terminal state
+/// still owes its clients.
+///
+/// Folds every gate — terminal status, the one-shot marker, pipeline
+/// provenance, pr/push kind inference — and the outcome evaluation into one
+/// pure decision, so the whole thing is testable without an `AppHandle`.
+///
+/// The marker means "outcome events for this session were delivered once", not
+/// "the session finished once": error and cancelled terminal states never reach
+/// evaluation (status gate) and never mark, so resuming a pipeline session that
+/// failed or was cancelled before it finished its work still fires its outcome
+/// on the first real completion.
+///
+/// `load_messages` is lazy because most completions are ordinary AI sessions
+/// that fail the pipeline gate, and there is no reason to read their transcript.
+fn pending_completion_effect(
+    session: &Session,
+    status: &str,
+    load_messages: impl FnOnce() -> Vec<SessionMessage>,
+) -> Option<(&'static str, CompletionEffect)> {
+    if status != SessionStatus::Completed.as_str()
+        || session.completion_effects_at.is_some()
+        || session.pipeline.is_none()
+    {
+        return None;
+    }
+    let kind = crate::session_commands::infer_branch_resume_session_type(&session.prompt)?;
+    let effect = evaluate_completed_session(kind, session, &load_messages())?;
+    Some((kind, effect))
+}
+
+/// Whether an effect counts as "outcomes delivered" for the one-shot marker.
+///
+/// `PrUrlMissing` emits and persists nothing, and leaving it unmarked preserves
+/// the recovery turn: a PR session that completed without producing a URL can
+/// be resumed ("you didn't create the PR — do it now"), and the next completion
+/// scans the new transcript and fires `pr-created` for real.
+fn should_record_effects(effect: &CompletionEffect) -> bool {
+    !matches!(effect, CompletionEffect::PrUrlMissing)
+}
+
+fn evaluate_completed_session(
+    kind: &str,
+    session: &Session,
+    messages: &[SessionMessage],
+) -> Option<CompletionEffect> {
+    match kind {
+        "pr" => Some(
+            extract_pr_url(messages)
+                .or_else(|| extract_pr_url_from_pipeline(session.pipeline.as_ref()?))
+                .map_or(CompletionEffect::PrUrlMissing, |(pr_url, pr_number)| {
+                    CompletionEffect::PrCreated { pr_url, pr_number }
+                }),
+        ),
+        "push" => Some(CompletionEffect::PushCompleted {
+            outcome: classify_completed_push_session(session.pipeline.as_ref(), messages),
+        }),
+        _ => None,
+    }
+}
+
+// =============================================================================
+// PR URL extraction
+// =============================================================================
+
+fn pr_url_regex() -> &'static regex::Regex {
+    static RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
+    RE.get_or_init(|| {
+        regex::Regex::new(
+            r"^https://github\.com/([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)/pull/(\d+)([/?#].*)?$",
+        )
+        .unwrap()
+    })
+}
+
+/// Canonicalize a PR URL candidate, stripping wrapping punctuation
+/// (`<url>`, `(url)`, trailing `.` etc.) and any query/fragment suffix.
+/// Returns the normalized URL and the PR number.
+fn normalize_pr_url(candidate: &str) -> Option<(String, u64)> {
+    let trimmed = candidate
+        .trim()
+        .trim_start_matches(['<', '`', '\'', '"', '[', '('])
+        .trim_end_matches(['>', '`', '\'', '"', ']', ')', ',', '.', '?', '!', ';', ':']);
+    let captures = pr_url_regex().captures(trimmed)?;
+    let number: u64 = captures[3].parse().ok()?;
+    Some((
+        format!(
+            "https://github.com/{}/{}/pull/{}",
+            &captures[1], &captures[2], number
+        ),
+        number,
+    ))
+}
+
+/// Find the PR URL in a session transcript.
+///
+/// First pass looks for the explicit `PR_URL: <url>` marker the PR session
+/// prompt asks the agent to output; the second pass falls back to any GitHub
+/// PR URL in the transcript. Both passes only consider assistant /
+/// tool-result messages: a URL in a user message (e.g. pasted into a queued
+/// follow-up) is not evidence the session created that PR.
+fn extract_pr_url(messages: &[SessionMessage]) -> Option<(String, u64)> {
+    static MARKER_RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
+    static URL_RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
+    let marker_re = MARKER_RE.get_or_init(|| regex::Regex::new(r"PR_URL:\s*(\S+)").unwrap());
+    let url_re = URL_RE.get_or_init(|| regex::Regex::new(r"https?://\S+").unwrap());
+
+    for msg in messages {
+        if !matches!(msg.role, MessageRole::Assistant | MessageRole::ToolResult) {
+            continue;
+        }
+        if let Some(captures) = marker_re.captures(&msg.content) {
+            if let Some(normalized) = normalize_pr_url(&captures[1]) {
+                return Some(normalized);
+            }
+        }
+    }
+
+    for msg in messages {
+        if !matches!(msg.role, MessageRole::Assistant | MessageRole::ToolResult) {
+            continue;
+        }
+        for candidate in url_re.find_iter(&msg.content) {
+            if let Some(normalized) = normalize_pr_url(candidate.as_str()) {
+                return Some(normalized);
+            }
+        }
+    }
+
+    None
+}
+
+/// Fallback for PR sessions whose pipeline steps produced the URL without an
+/// AI handoff (or where the transcript was lost): scan step outputs.
+fn extract_pr_url_from_pipeline(pipeline: &PipelineExecution) -> Option<(String, u64)> {
+    static STEP_URL_RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
+    let step_url_re = STEP_URL_RE.get_or_init(|| {
+        regex::Regex::new(r"https://github\.com/[^\s/]+/[^\s/]+/pull/\d+").unwrap()
+    });
+
+    pipeline
+        .steps
+        .iter()
+        .filter_map(|step| step.output.as_deref())
+        .find_map(|output| {
+            step_url_re
+                .find(output)
+                .and_then(|m| normalize_pr_url(m.as_str()))
+        })
+}
+
+// =============================================================================
+// Push outcome classification
+// =============================================================================
+
+fn contains_non_fast_forward_marker(content: &str) -> bool {
+    content.contains("PUSH_REJECTED: NON_FAST_FORWARD")
+        || content.to_lowercase().contains("non-fast-forward")
+}
+
+/// Classify a completed push session.
+///
+/// The deterministic pipeline is consulted first: a failed step whose output
+/// carries the non-fast-forward marker means the push was rejected — unless an
+/// AI turn ran afterwards (e.g. a failed `--force-with-lease` whose error
+/// happened to mention "non-fast-forward"), in which case the AI handled
+/// recovery. When the pipeline is inconclusive, the transcript markers decide;
+/// the default is success.
+fn classify_completed_push_session(
+    pipeline: Option<&PipelineExecution>,
+    messages: &[SessionMessage],
+) -> PushOutcome {
+    if let Some(pipeline) = pipeline {
+        let has_non_fast_forward = pipeline.steps.iter().any(|step| {
+            step.status == StepStatus::Failed
+                && step
+                    .output
+                    .as_deref()
+                    .is_some_and(contains_non_fast_forward_marker)
+        });
+        if has_non_fast_forward {
+            let ai_ran = messages
+                .iter()
+                .any(|msg| msg.role == MessageRole::Assistant);
+            return if ai_ran {
+                PushOutcome::Succeeded
+            } else {
+                PushOutcome::RejectedNonFastForward
+            };
+        }
+
+        let all_steps_passed_or_skipped = pipeline
+            .steps
+            .iter()
+            .all(|step| matches!(step.status, StepStatus::Succeeded | StepStatus::Skipped));
+        if pipeline.completed_without_ai || all_steps_passed_or_skipped {
+            return PushOutcome::Succeeded;
+        }
+    }
+
+    let rejected_in_transcript = messages.iter().any(|msg| {
+        matches!(msg.role, MessageRole::Assistant | MessageRole::ToolResult)
+            && contains_non_fast_forward_marker(&msg.content)
+    });
+    if rejected_in_transcript {
+        PushOutcome::RejectedNonFastForward
+    } else {
+        PushOutcome::Succeeded
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::store::{PipelineStepStatus, StepType};
+
+    fn message(role: MessageRole, content: &str) -> SessionMessage {
+        SessionMessage {
+            id: 0,
+            session_id: "session-1".to_string(),
+            role,
+            content: content.to_string(),
+            created_at: 0,
+            image_ids: vec![],
+            acp: Default::default(),
+        }
+    }
+
+    fn step(status: StepStatus, output: Option<&str>) -> PipelineStepStatus {
+        PipelineStepStatus {
+            label: "step".to_string(),
+            step_type: StepType::Command,
+            status,
+            output: output.map(str::to_string),
+            error: None,
+            started_at: None,
+            completed_at: None,
+        }
+    }
+
+    fn pipeline(steps: Vec<PipelineStepStatus>) -> PipelineExecution {
+        PipelineExecution {
+            kind: None,
+            rebase_target: None,
+            push_force: false,
+            steps,
+            current_step: 0,
+            completed_without_ai: false,
+        }
+    }
+
+    fn pr_session(pipeline: PipelineExecution) -> Session {
+        let mut session = Session::new_running(
+            "Create a pull request for the current branch",
+            std::path::Path::new("/tmp"),
+        )
+        .with_branch("branch-1");
+        session.pipeline = Some(pipeline);
+        session
+    }
+
+    fn push_session(pipeline: PipelineExecution) -> Session {
+        let mut session = Session::new_running(
+            "Push the current branch to the remote",
+            std::path::Path::new("/tmp"),
+        )
+        .with_branch("branch-1");
+        session.pipeline = Some(pipeline);
+        session
+    }
+
+    fn succeeded_push_pipeline() -> PipelineExecution {
+        pipeline(vec![step(StepStatus::Succeeded, Some("pushed"))])
+    }
+
+    #[test]
+    fn extracts_pr_url_from_marker() {
+        let messages = vec![
+            message(MessageRole::User, "Create a PR"),
+            message(
+                MessageRole::Assistant,
+                "Done!\nPR_URL: https://github.com/org/repo/pull/42",
+            ),
+        ];
+        assert_eq!(
+            extract_pr_url(&messages),
+            Some(("https://github.com/org/repo/pull/42".to_string(), 42))
+        );
+    }
+
+    #[test]
+    fn marker_pass_ignores_user_messages_and_prefers_marker_over_earlier_urls() {
+        let messages = vec![
+            message(
+                MessageRole::User,
+                "PR_URL: https://github.com/org/repo/pull/1",
+            ),
+            message(
+                MessageRole::Assistant,
+                "see https://github.com/org/repo/actions first",
+            ),
+            message(
+                MessageRole::ToolResult,
+                "PR_URL: <https://github.com/org/repo/pull/7>.",
+            ),
+        ];
+        assert_eq!(
+            extract_pr_url(&messages),
+            Some(("https://github.com/org/repo/pull/7".to_string(), 7))
+        );
+    }
+
+    #[test]
+    fn falls_back_to_any_pr_url_with_wrapping_punctuation() {
+        let messages = vec![message(
+            MessageRole::Assistant,
+            "Created the PR (https://github.com/org/repo/pull/123?diff=split).",
+        )];
+        assert_eq!(
+            extract_pr_url(&messages),
+            Some(("https://github.com/org/repo/pull/123".to_string(), 123))
+        );
+    }
+
+    #[test]
+    fn fallback_pass_ignores_user_messages() {
+        let messages = vec![
+            message(
+                MessageRole::User,
+                "see https://github.com/org/repo/pull/3 for prior art",
+            ),
+            message(MessageRole::Assistant, "Working on it."),
+        ];
+        assert_eq!(extract_pr_url(&messages), None);
+    }
+
+    #[test]
+    fn ignores_non_pr_github_urls() {
+        let messages = vec![message(
+            MessageRole::Assistant,
+            "See https://github.com/org/repo/issues/9 and https://example.com/pull/3",
+        )];
+        assert_eq!(extract_pr_url(&messages), None);
+    }
+
+    #[test]
+    fn extracts_pr_url_from_pipeline_step_output() {
+        let execution = pipeline(vec![
+            step(StepStatus::Succeeded, Some("pushed")),
+            step(
+                StepStatus::Succeeded,
+                Some("https://github.com/org/repo/pull/55\n"),
+            ),
+        ]);
+        assert_eq!(
+            extract_pr_url_from_pipeline(&execution),
+            Some(("https://github.com/org/repo/pull/55".to_string(), 55))
+        );
+    }
+
+    #[test]
+    fn evaluate_pr_session_without_url_reports_missing() {
+        let session = pr_session(pipeline(vec![step(StepStatus::Succeeded, Some("ok"))]));
+        assert_eq!(
+            evaluate_completed_session("pr", &session, &[]),
+            Some(CompletionEffect::PrUrlMissing)
+        );
+    }
+
+    #[test]
+    fn evaluate_pr_session_with_url_reports_created() {
+        let session = pr_session(pipeline(vec![step(StepStatus::Succeeded, Some("ok"))]));
+        let messages = vec![message(
+            MessageRole::Assistant,
+            "PR_URL: https://github.com/org/repo/pull/8",
+        )];
+        assert_eq!(
+            evaluate_completed_session("pr", &session, &messages),
+            Some(CompletionEffect::PrCreated {
+                pr_url: "https://github.com/org/repo/pull/8".to_string(),
+                pr_number: 8,
+            })
+        );
+    }
+
+    #[test]
+    fn push_rejected_when_failed_step_has_marker_and_no_ai_ran() {
+        let execution = pipeline(vec![step(
+            StepStatus::Failed,
+            Some("! [rejected] main -> main (non-fast-forward)"),
+        )]);
+        assert_eq!(
+            classify_completed_push_session(Some(&execution), &[]),
+            PushOutcome::RejectedNonFastForward
+        );
+    }
+
+    #[test]
+    fn push_succeeds_when_ai_recovered_after_marker() {
+        let execution = pipeline(vec![step(
+            StepStatus::Failed,
+            Some("error: failed to push (non-fast-forward)"),
+        )]);
+        let messages = vec![message(MessageRole::Assistant, "Retried and pushed.")];
+        assert_eq!(
+            classify_completed_push_session(Some(&execution), &messages),
+            PushOutcome::Succeeded
+        );
+    }
+
+    #[test]
+    fn push_succeeds_when_all_steps_passed_or_skipped() {
+        let execution = pipeline(vec![
+            step(StepStatus::Succeeded, Some("pushed")),
+            step(StepStatus::Skipped, None),
+        ]);
+        assert_eq!(
+            classify_completed_push_session(Some(&execution), &[]),
+            PushOutcome::Succeeded
+        );
+    }
+
+    #[test]
+    fn push_falls_back_to_transcript_markers_when_pipeline_inconclusive() {
+        let execution = pipeline(vec![
+            step(StepStatus::Failed, Some("some unrelated failure")),
+            step(StepStatus::Skipped, None),
+        ]);
+        let messages = vec![message(
+            MessageRole::ToolResult,
+            "PUSH_REJECTED: NON_FAST_FORWARD",
+        )];
+        assert_eq!(
+            classify_completed_push_session(Some(&execution), &messages),
+            PushOutcome::RejectedNonFastForward
+        );
+        assert_eq!(
+            classify_completed_push_session(None, &messages),
+            PushOutcome::RejectedNonFastForward
+        );
+    }
+
+    #[test]
+    fn push_defaults_to_succeeded_without_markers() {
+        assert_eq!(
+            classify_completed_push_session(None, &[message(MessageRole::Assistant, "pushed")]),
+            PushOutcome::Succeeded
+        );
+    }
+
+    #[test]
+    fn unmarked_completed_push_session_has_a_pending_effect() {
+        let session = push_session(succeeded_push_pipeline());
+        assert_eq!(
+            pending_completion_effect(&session, "completed", Vec::new),
+            Some((
+                "push",
+                CompletionEffect::PushCompleted {
+                    outcome: PushOutcome::Succeeded
+                }
+            ))
+        );
+    }
+
+    #[test]
+    fn marked_session_has_no_pending_effect() {
+        // A resumed push session: the old all-succeeded pipeline still
+        // classifies as a fresh success, so only the marker stops the
+        // destructive re-clear of the branch's PR status.
+        let mut session = push_session(succeeded_push_pipeline());
+        session.completion_effects_at = Some(1_700_000_000_000);
+        assert_eq!(
+            pending_completion_effect(&session, "completed", Vec::new),
+            None
+        );
+    }
+
+    #[test]
+    fn non_completed_terminal_states_have_no_pending_effect() {
+        let session = push_session(succeeded_push_pipeline());
+        for status in ["error", "cancelled", "running"] {
+            assert_eq!(
+                pending_completion_effect(&session, status, Vec::new),
+                None,
+                "status {status} should not evaluate an outcome"
+            );
+        }
+    }
+
+    #[test]
+    fn non_pipeline_and_non_pipeline_kind_sessions_have_no_pending_effect() {
+        let plain = Session::new_running("Fix the login flow", std::path::Path::new("/tmp"));
+        assert_eq!(
+            pending_completion_effect(&plain, "completed", Vec::new),
+            None
+        );
+
+        let mut ai_with_pipeline = plain.clone();
+        ai_with_pipeline.pipeline = Some(succeeded_push_pipeline());
+        assert_eq!(
+            pending_completion_effect(&ai_with_pipeline, "completed", Vec::new),
+            None
+        );
+    }
+
+    #[test]
+    fn only_pr_url_missing_skips_the_marker() {
+        assert!(should_record_effects(&CompletionEffect::PrCreated {
+            pr_url: "https://github.com/org/repo/pull/8".to_string(),
+            pr_number: 8,
+        }));
+        assert!(should_record_effects(&CompletionEffect::PushCompleted {
+            outcome: PushOutcome::Succeeded
+        }));
+        assert!(should_record_effects(&CompletionEffect::PushCompleted {
+            outcome: PushOutcome::RejectedNonFastForward
+        }));
+        assert!(!should_record_effects(&CompletionEffect::PrUrlMissing));
+    }
+
+    #[test]
+    fn unmarked_pr_url_missing_session_fires_on_a_later_completion() {
+        let session = pr_session(pipeline(vec![step(StepStatus::Succeeded, Some("ok"))]));
+
+        // First completion produced no URL, so nothing was delivered and the
+        // session stays eligible.
+        let first = pending_completion_effect(&session, "completed", Vec::new);
+        assert_eq!(first, Some(("pr", CompletionEffect::PrUrlMissing)));
+        assert!(!should_record_effects(&first.unwrap().1));
+
+        // Resumed with "you didn't create the PR — do it now": the next
+        // completion scans the new transcript and fires for real.
+        let messages = vec![message(
+            MessageRole::Assistant,
+            "PR_URL: https://github.com/org/repo/pull/9",
+        )];
+        assert_eq!(
+            pending_completion_effect(&session, "completed", || messages),
+            Some((
+                "pr",
+                CompletionEffect::PrCreated {
+                    pr_url: "https://github.com/org/repo/pull/9".to_string(),
+                    pr_number: 9,
+                }
+            ))
+        );
+    }
+}
diff --git a/apps/staged/src-tauri/src/session_runner.rs b/apps/staged/src-tauri/src/session_runner.rs
index c86c42055..46a7b69bb 100644
--- a/apps/staged/src-tauri/src/session_runner.rs
+++ b/apps/staged/src-tauri/src/session_runner.rs
@@ -8,7 +8,10 @@
 //! 2. Registers the session for cancellation
 //! 3. Spawns a background thread that runs the driver
 //! 4. On completion, atomically transitions the DB status
-//! 5. Emits a single `session-status-changed` event
+//! 5. Runs completion side effects for pr/push sessions (persisting the
+//!    parsed outcome and emitting `pr-created` / `push-completed`, see
+//!    [`crate::session_completion`])
+//! 6. Emits a single `session-status-changed` event
 //!
 //! The frontend never sees streaming events — it polls the DB.
 //!
@@ -849,6 +852,20 @@ pub fn start_session(
             )
             .unwrap_or(false);
 
+        // Parse and persist pr/push outcomes and emit the pr-created /
+        // push-completed domain events. Gated on winning the transition so
+        // exactly one writer runs them, and emitted before the terminal
+        // status event so clients see the outcome first.
+        if transitioned {
+            crate::session_completion::run_completion_side_effects(
+                &store_for_status,
+                &app_handle,
+                &session_id_for_status,
+                config.branch_id.as_deref(),
+                new_status,
+            );
+        }
+
         // Always emit the terminal status event, even if the DB row was already
         // deleted (e.g. user deleted the pending commit). This lets the frontend
         // clean up sidebar "running" state as a safety net.
@@ -1185,15 +1202,25 @@ pub fn start_pipeline_session(
                 // Pipeline completed successfully — transition session to completed.
                 resolve_pipeline_artifacts_without_ai(&config, &store_for_status, true);
                 let status_enum = SessionStatus::Completed;
+                let status_str = status_enum.as_str();
                 let reason = CompletionReason::TurnComplete;
                 registry.deregister(&session_id);
                 let transitioned = store_for_status
                     .transition_from_running(&session_id, status_enum, None, Some(&reason))
                     .unwrap_or(false);
+                if transitioned {
+                    crate::session_completion::run_completion_side_effects(
+                        &store_for_status,
+                        &app_handle,
+                        &session_id,
+                        config.branch_id.as_deref(),
+                        status_str,
+                    );
+                }
                 emit_status(
                     &app_handle,
                     &session_id,
-                    "completed",
+                    status_str,
                     None,
                     Some(&reason),
                     config.branch_id.clone(),
@@ -1326,23 +1353,36 @@ pub fn start_pipeline_session(
                 // The pipeline ran to its conclusion either way; only the
                 // outcome differs, so the completion reason stays the same.
                 let reason = CompletionReason::TurnComplete;
-                let status = if error.is_some() {
+                let status_enum = if error.is_some() {
                     SessionStatus::Error
                 } else {
                     SessionStatus::Completed
                 };
+                let status_str = status_enum.as_str();
                 registry.deregister(&session_id);
                 let transitioned = store_for_status
-                    .transition_from_running(&session_id, status, error.as_deref(), Some(&reason))
+                    .transition_from_running(
+                        &session_id,
+                        status_enum,
+                        error.as_deref(),
+                        Some(&reason),
+                    )
                     .unwrap_or(false);
+                if transitioned {
+                    // Classifies the abort (e.g. push rejected non-fast-forward)
+                    // and emits the push-completed domain event.
+                    crate::session_completion::run_completion_side_effects(
+                        &store_for_status,
+                        &app_handle,
+                        &session_id,
+                        config.branch_id.as_deref(),
+                        status_str,
+                    );
+                }
                 emit_status(
                     &app_handle,
                     &session_id,
-                    if error.is_some() {
-                        "error"
-                    } else {
-                        "completed"
-                    },
+                    status_str,
                     error,
                     Some(&reason),
                     config.branch_id.clone(),
diff --git a/apps/staged/src-tauri/src/store/migration_tests.rs b/apps/staged/src-tauri/src/store/migration_tests.rs
index e0083da1e..73e40c3fa 100644
--- a/apps/staged/src-tauri/src/store/migration_tests.rs
+++ b/apps/staged/src-tauri/src/store/migration_tests.rs
@@ -145,7 +145,7 @@ fn test_store_bootstraps_fresh_database_with_baseline_migration() {
         )
         .unwrap();
 
-    assert_eq!(version, 22);
+    assert_eq!(version, 23);
     assert_eq!(app_version, super::APP_VERSION);
     assert!(table_exists(&conn, "projects"));
     assert!(table_exists(&conn, "project_notes"));
@@ -161,6 +161,7 @@ fn test_store_bootstraps_fresh_database_with_baseline_migration() {
     assert!(column_exists(&conn, "sessions", "acp_config_selection"));
     assert!(column_exists(&conn, "sessions", "acp_title"));
     assert!(column_exists(&conn, "sessions", "branch_id"));
+    assert!(column_exists(&conn, "sessions", "completion_effects_at"));
 
     let trigger_count: i64 = conn
         .query_row(
@@ -189,7 +190,13 @@ fn test_store_repairs_github_comment_tracking_user_version() {
             app_version TEXT NOT NULL
         );
         INSERT INTO app_metadata (id, app_version) VALUES (1, '0.2.9');
-        CREATE TABLE sessions (id TEXT PRIMARY KEY);
+        -- `status` and `updated_at` predate every migration below; they are
+        -- spelled out because the 0023 backfill reads them.
+        CREATE TABLE sessions (
+            id         TEXT PRIMARY KEY,
+            status     TEXT NOT NULL,
+            updated_at INTEGER NOT NULL
+        );
         CREATE TABLE session_messages (
             id          INTEGER PRIMARY KEY AUTOINCREMENT,
             session_id  TEXT NOT NULL,
@@ -224,11 +231,12 @@ fn test_store_repairs_github_comment_tracking_user_version() {
     let version: i64 = conn
         .query_row("PRAGMA user_version", [], |row| row.get(0))
         .unwrap();
-    assert_eq!(version, 22);
+    assert_eq!(version, 23);
     assert!(column_exists(&conn, "sessions", "pipeline"));
     assert!(column_exists(&conn, "sessions", "acp_config_selection"));
     assert!(column_exists(&conn, "sessions", "acp_title"));
     assert!(column_exists(&conn, "sessions", "branch_id"));
+    assert!(column_exists(&conn, "sessions", "completion_effects_at"));
     assert!(column_exists(
         &conn,
         "session_messages",
@@ -252,8 +260,10 @@ fn test_store_repairs_pipeline_user_version() {
         );
         INSERT INTO app_metadata (id, app_version) VALUES (1, '0.2.9');
         CREATE TABLE sessions (
-            id       TEXT PRIMARY KEY,
-            pipeline TEXT
+            id         TEXT PRIMARY KEY,
+            status     TEXT NOT NULL,
+            pipeline   TEXT,
+            updated_at INTEGER NOT NULL
         );
         CREATE TABLE session_messages (
             id          INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -284,7 +294,7 @@ fn test_store_repairs_pipeline_user_version() {
     let version: i64 = conn
         .query_row("PRAGMA user_version", [], |row| row.get(0))
         .unwrap();
-    assert_eq!(version, 22);
+    assert_eq!(version, 23);
     assert!(column_exists(&conn, "comments", "github_comment_id"));
     assert!(column_exists(&conn, "comments", "github_comment_type"));
     assert!(column_exists(&conn, "comments", "github_comment_stale"));
@@ -297,6 +307,65 @@ fn test_store_repairs_pipeline_user_version() {
     assert!(column_exists(&conn, "sessions", "acp_config_selection"));
     assert!(column_exists(&conn, "sessions", "acp_title"));
     assert!(column_exists(&conn, "sessions", "branch_id"));
+    assert!(column_exists(&conn, "sessions", "completion_effects_at"));
+
+    cleanup_db(&path);
+}
+
+#[test]
+fn test_completion_effects_migration_backfills_finished_pipeline_sessions() {
+    let path = temp_db_path("completion-effects-backfill");
+    let conn = Connection::open(&path).unwrap();
+    conn.execute_batch(
+        "
+        PRAGMA user_version = 22;
+        CREATE TABLE app_metadata (
+            id          INTEGER PRIMARY KEY CHECK (id = 1),
+            app_version TEXT NOT NULL
+        );
+        INSERT INTO app_metadata (id, app_version) VALUES (1, '0.2.9');
+        CREATE TABLE sessions (
+            id         TEXT PRIMARY KEY,
+            status     TEXT NOT NULL,
+            pipeline   TEXT,
+            updated_at INTEGER NOT NULL
+        );
+        INSERT INTO sessions (id, status, pipeline, updated_at) VALUES
+            ('completed-pipeline', 'completed', '{}', 100),
+            ('running-pipeline',   'running',   '{}', 200),
+            ('error-pipeline',     'error',     '{}', 300),
+            ('completed-ai',       'completed', NULL, 400);
+        ",
+    )
+    .unwrap();
+    drop(conn);
+
+    let store = Store::new(&path).unwrap();
+    drop(store);
+
+    let conn = Connection::open(&path).unwrap();
+    let version: i64 = conn
+        .query_row("PRAGMA user_version", [], |row| row.get(0))
+        .unwrap();
+    assert_eq!(version, 23);
+    assert!(column_exists(&conn, "sessions", "completion_effects_at"));
+
+    let marker = |id: &str| -> Option<i64> {
+        conn.query_row(
+            "SELECT completion_effects_at FROM sessions WHERE id = ?1",
+            params![id],
+            |row| row.get(0),
+        )
+        .unwrap()
+    };
+    // Already-finished pipeline sessions are exactly the rows whose next resume
+    // would replay their outcome effects, so they start out marked.
+    assert_eq!(marker("completed-pipeline"), Some(100));
+    // Still-running and failed pipeline sessions have outcomes yet to deliver.
+    assert_eq!(marker("running-pipeline"), None);
+    assert_eq!(marker("error-pipeline"), None);
+    // Plain AI sessions never had completion side effects to begin with.
+    assert_eq!(marker("completed-ai"), None);
 
     cleanup_db(&path);
 }
diff --git a/apps/staged/src-tauri/src/store/migrations/0023-add-session-completion-effects-at/up.sql b/apps/staged/src-tauri/src/store/migrations/0023-add-session-completion-effects-at/up.sql
new file mode 100644
index 000000000..16477ff3f
--- /dev/null
+++ b/apps/staged/src-tauri/src/store/migrations/0023-add-session-completion-effects-at/up.sql
@@ -0,0 +1,13 @@
+-- One-shot marker recording that a pipeline (pr/push) session's completion
+-- outcome events were already delivered. Everything the completion hook gates
+-- on (pipeline JSON, prompt, branch_id) persists on the session row forever, so
+-- resuming a finished pr/push session re-ran the hook against the *old*
+-- pipeline and transcript on the follow-up turn's completion — re-emitting
+-- `pr-created`, or (destructively) re-clearing the branch's PR status because
+-- the old all-succeeded push pipeline still classifies as a fresh success.
+ALTER TABLE sessions ADD COLUMN completion_effects_at INTEGER DEFAULT NULL;
+
+-- Backfill the existing inventory of finished pipeline sessions: those are
+-- exactly the rows whose next resume would replay the effects.
+UPDATE sessions SET completion_effects_at = updated_at
+WHERE status = 'completed' AND pipeline IS NOT NULL;
diff --git a/apps/staged/src-tauri/src/store/models.rs b/apps/staged/src-tauri/src/store/models.rs
index 1325ae8fb..0c99affd0 100644
--- a/apps/staged/src-tauri/src/store/models.rs
+++ b/apps/staged/src-tauri/src/store/models.rs
@@ -534,11 +534,19 @@ pub struct Session {
     /// Branch this session belongs to, for sessions that create no artifact.
     ///
     /// Branch-scoped sessions are normally found through their commit, note, or
-    /// review row. Push pipelines have none of those, so they record the branch
-    /// here to stay visible to the branch queue. `None` for artifact-backed
-    /// sessions and for project-level sessions.
+    /// review row. Pipeline-launched (pr/push) sessions have none of those, so
+    /// they record the branch here to stay visible to the branch queue and to
+    /// let completion side-effects resolve the branch from the DB alone.
+    /// `None` for artifact-backed sessions and for project-level sessions.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub branch_id: Option<String>,
+    /// When this session's pipeline (pr/push) completion outcome events were
+    /// delivered. A one-shot marker: it means "outcome events for this session
+    /// were delivered once", *not* "the session finished once". Resuming a
+    /// finished pr/push session completes it again, and the stale pipeline plus
+    /// transcript would otherwise re-fire those effects.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub completion_effects_at: Option<i64>,
 }
 
 /// Persistent follow-up message waiting to be sent to an existing session.
@@ -630,6 +638,7 @@ impl Session {
             acp_config_selection: None,
             acp_title: None,
             branch_id: None,
+            completion_effects_at: None,
         }
     }
 
@@ -654,6 +663,7 @@ impl Session {
             acp_config_selection: None,
             acp_title: None,
             branch_id: None,
+            completion_effects_at: None,
         }
     }
 
diff --git a/apps/staged/src-tauri/src/store/sessions.rs b/apps/staged/src-tauri/src/store/sessions.rs
index 9581f7950..2e56064c3 100644
--- a/apps/staged/src-tauri/src/store/sessions.rs
+++ b/apps/staged/src-tauri/src/store/sessions.rs
@@ -19,8 +19,8 @@ impl Store {
         let acp_config_selection_json =
             serialize_acp_config_selection(session.acp_config_selection.as_ref())?;
         conn.execute(
-            "INSERT INTO sessions (id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id)
-             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
+            "INSERT INTO sessions (id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id, completion_effects_at)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)",
             params![
                 session.id,
                 session.prompt,
@@ -37,6 +37,7 @@ impl Store {
                 acp_config_selection_json,
                 session.acp_title,
                 session.branch_id,
+                session.completion_effects_at,
             ],
         )?;
         Ok(())
@@ -45,7 +46,7 @@ impl Store {
     pub fn get_session(&self, id: &str) -> Result<Option<Session>, StoreError> {
         let conn = self.conn.lock().unwrap();
         conn.query_row(
-            "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id
+            "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id, completion_effects_at
              FROM sessions WHERE id = ?1",
             params![id],
             Self::row_to_session,
@@ -224,7 +225,7 @@ impl Store {
     pub fn get_running_sessions(&self) -> Result<Vec<Session>, StoreError> {
         let conn = self.conn.lock().unwrap();
         let mut stmt = conn.prepare(
-            "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id
+            "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id, completion_effects_at
              FROM sessions WHERE status = 'running'",
         )?;
         let sessions = stmt
@@ -233,6 +234,24 @@ impl Store {
         Ok(sessions)
     }
 
+    /// Get all running and queued sessions, oldest first.
+    ///
+    /// Backs the `get_active_sessions` busy-state snapshot command, which
+    /// clients hydrate from on load or reconnect instead of relying solely on
+    /// accumulated `session-status-changed` events.
+    pub fn get_active_sessions(&self) -> Result<Vec<Session>, StoreError> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare(
+            "SELECT id, prompt, status, working_dir, provider, agent_id, error_message, completion_reason, created_at, updated_at, owner_pid, pipeline, acp_config_selection, acp_title, branch_id, completion_effects_at
+             FROM sessions WHERE status IN ('running', 'queued')
+             ORDER BY created_at ASC",
+        )?;
+        let sessions = stmt
+            .query_map([], Self::row_to_session)?
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(sessions)
+    }
+
     /// Store the ACP session ID returned by the agent after `new_session`.
     /// This is used by `load_session` to resume the conversation on follow-up turns.
     pub fn set_agent_session_id(&self, id: &str, agent_session_id: &str) -> Result<(), StoreError> {
@@ -298,7 +317,7 @@ impl Store {
     ) -> Result<Vec<Session>, StoreError> {
         let conn = self.conn.lock().unwrap();
         let mut stmt = conn.prepare(
-            "SELECT s.id, s.prompt, s.status, s.working_dir, s.provider, s.agent_id, s.error_message, s.completion_reason, s.created_at, s.updated_at, s.owner_pid, s.pipeline, s.acp_config_selection, s.acp_title, s.branch_id
+            "SELECT s.id, s.prompt, s.status, s.working_dir, s.provider, s.agent_id, s.error_message, s.completion_reason, s.created_at, s.updated_at, s.owner_pid, s.pipeline, s.acp_config_selection, s.acp_title, s.branch_id, s.completion_effects_at
              FROM sessions s
              WHERE s.status = 'queued'
                AND (
@@ -339,7 +358,8 @@ impl Store {
     }
 
     /// Resolve the branch that owns a session through its linked artifact, or
-    /// through `sessions.branch_id` for artifact-less branch work (pushes).
+    /// through `sessions.branch_id` for artifact-less branch work (pipeline
+    /// pr/push sessions).
     ///
     /// Project-note sessions do not belong to a branch and therefore return `None`.
     /// This assumes all branch-linked artifacts for a session point at the same
@@ -444,6 +464,21 @@ impl Store {
         Ok(())
     }
 
+    /// Record that this session's pipeline (pr/push) completion outcome events
+    /// have been delivered, so a later resume of the same session doesn't
+    /// re-fire them against its stale pipeline and transcript.
+    ///
+    /// One-shot: the completion hook refuses to run once this is set.
+    pub fn mark_completion_effects_ran(&self, id: &str) -> Result<(), StoreError> {
+        let conn = self.conn.lock().unwrap();
+        let now = now_timestamp();
+        conn.execute(
+            "UPDATE sessions SET completion_effects_at = ?1, updated_at = ?1 WHERE id = ?2",
+            params![now, id],
+        )?;
+        Ok(())
+    }
+
     /// Update the pipeline execution state for a session.
     pub fn update_session_pipeline(
         &self,
@@ -491,6 +526,7 @@ impl Store {
             acp_config_selection,
             acp_title: row.get(13)?,
             branch_id: row.get(14)?,
+            completion_effects_at: row.get(15)?,
         })
     }
 }
diff --git a/apps/staged/src-tauri/src/store/tests.rs b/apps/staged/src-tauri/src/store/tests.rs
index 0f4b36e11..5d2f88669 100644
--- a/apps/staged/src-tauri/src/store/tests.rs
+++ b/apps/staged/src-tauri/src/store/tests.rs
@@ -545,6 +545,33 @@ fn test_transition_queued_to_running_does_not_overwrite_cancelled_session() {
     assert_eq!(final_state.owner_pid, None);
 }
 
+#[test]
+fn test_get_active_sessions_returns_running_and_queued_only() {
+    let store = Store::in_memory().unwrap();
+
+    let running = Session::new_running("running", Path::new("/tmp"));
+    store.create_session(&running).unwrap();
+    let queued = Session::new_queued("queued");
+    store.create_session(&queued).unwrap();
+
+    for status in [
+        SessionStatus::Completed,
+        SessionStatus::Error,
+        SessionStatus::Cancelled,
+    ] {
+        let terminal = Session::new_running("terminal", Path::new("/tmp"));
+        store.create_session(&terminal).unwrap();
+        store
+            .update_session_status(&terminal.id, status, None, None)
+            .unwrap();
+    }
+
+    let active = store.get_active_sessions().unwrap();
+    assert_eq!(active.len(), 2);
+    assert!(active.iter().any(|s| s.id == running.id));
+    assert!(active.iter().any(|s| s.id == queued.id));
+}
+
 #[test]
 fn test_queued_session_messages_order_and_image_ids() {
     let store = Store::in_memory().unwrap();
@@ -1176,6 +1203,74 @@ fn test_running_pipeline_commit_marks_branch_busy_and_resolves_branch() {
     );
 }
 
+#[test]
+fn test_session_branch_id_round_trips_and_resolves_branch_and_project() {
+    let store = Store::in_memory().unwrap();
+    let project = Project::new("test-owner/test-repo");
+    store.create_project(&project).unwrap();
+    let branch = Branch::new(&project.id, "feature", "main");
+    store.create_branch(&branch).unwrap();
+
+    // A pr/push pipeline session: no artifact row, branch carried on the
+    // session row itself.
+    let session = Session::new_running(
+        "Create a pull request for the current branch",
+        Path::new("/tmp"),
+    )
+    .with_branch(&branch.id);
+    store.create_session(&session).unwrap();
+
+    let fetched = store.get_session(&session.id).unwrap().unwrap();
+    assert_eq!(fetched.branch_id.as_deref(), Some(branch.id.as_str()));
+
+    assert_eq!(
+        store
+            .get_branch_id_for_session(&session.id)
+            .unwrap()
+            .as_deref(),
+        Some(branch.id.as_str())
+    );
+    assert_eq!(
+        store
+            .get_project_id_for_session(&session.id)
+            .unwrap()
+            .as_deref(),
+        Some(project.id.as_str())
+    );
+}
+
+#[test]
+fn test_session_completion_effects_marker_round_trips() {
+    let store = Store::in_memory().unwrap();
+    let session = Session::new_running("Push the current branch to the remote", Path::new("/tmp"));
+    store.create_session(&session).unwrap();
+
+    let fetched = store.get_session(&session.id).unwrap().unwrap();
+    assert_eq!(fetched.completion_effects_at, None);
+
+    store.mark_completion_effects_ran(&session.id).unwrap();
+    let marked = store.get_session(&session.id).unwrap().unwrap();
+    let marked_at = marked
+        .completion_effects_at
+        .expect("marker should be set after delivering outcomes");
+    assert!(marked_at >= session.created_at);
+    assert_eq!(marked.updated_at, marked_at);
+}
+
+#[test]
+fn test_session_without_branch_or_artifact_resolves_to_none() {
+    let store = Store::in_memory().unwrap();
+    let session = Session::new_running("unattributed", Path::new("/tmp"));
+    store.create_session(&session).unwrap();
+
+    assert_eq!(
+        store.get_session(&session.id).unwrap().unwrap().branch_id,
+        None
+    );
+    assert_eq!(store.get_branch_id_for_session(&session.id).unwrap(), None);
+    assert_eq!(store.get_project_id_for_session(&session.id).unwrap(), None);
+}
+
 #[test]
 fn test_completion_reason_round_trips() {
     let store = Store::in_memory().unwrap();
diff --git a/apps/staged/src-tauri/src/web_server.rs b/apps/staged/src-tauri/src/web_server.rs
index d53454bbb..1bed48d69 100644
--- a/apps/staged/src-tauri/src/web_server.rs
+++ b/apps/staged/src-tauri/src/web_server.rs
@@ -2791,6 +2791,11 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result<Val
             .await?;
             Ok(serde_json::to_value(config).unwrap())
         }
+        "get_active_sessions" => {
+            let store = get_store(store_mutex)?;
+            let sessions = crate::session_commands::get_active_sessions_impl(&store)?;
+            Ok(serde_json::to_value(sessions).unwrap())
+        }
         "get_session" => {
             let store = get_store(store_mutex)?;
             let session_id: String = arg(&args, "sessionId")?;
@@ -3464,74 +3469,7 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result<Val
         "refresh_pr_status" => {
             let store = get_store(store_mutex)?;
             let branch_id: String = arg(&args, "branchId")?;
-
-            let branch = store
-                .get_branch(&branch_id)
-                .map_err(|e| e.to_string())?
-                .ok_or_else(|| format!("Branch not found: {branch_id}"))?;
-            let pr_number = branch
-                .pr_number
-                .ok_or_else(|| "Branch does not have an associated PR".to_string())?;
-            let project = store
-                .get_project(&branch.project_id)
-                .map_err(|e| e.to_string())?
-                .ok_or_else(|| format!("Project not found: {}", branch.project_id))?;
-            let (github_repo, _) =
-                crate::prs::resolve_branch_repo_and_subpath(&store, &project, &branch)?;
-
-            let pr_status = {
-                let github_repo = github_repo.clone();
-                tokio::task::spawn_blocking(move || {
-                    crate::git::fetch_pr_status_for_repo(&github_repo, pr_number)
-                })
-                .await
-                .map_err(|e| format!("refresh_pr_status task failed: {e}"))?
-            };
-            let pr_status = match pr_status {
-                Ok(status) => status,
-                Err(e) => {
-                    log::error!(
-                        "refresh_pr_status failed for branch_id={}, pr_number={}: {}",
-                        branch_id,
-                        pr_number,
-                        e
-                    );
-                    return Err(e.to_string());
-                }
-            };
-            let mergeable = pr_status.mergeable == "MERGEABLE";
-            let pr_fetched_at = store::now_timestamp();
-
-            store
-                .update_branch_pr_status(
-                    &branch_id,
-                    Some(pr_status.state.clone()),
-                    Some(pr_status.checks_summary.state.clone()),
-                    pr_status.review_decision.clone(),
-                    Some(mergeable),
-                    Some(pr_status.is_draft),
-                    None,
-                    None,
-                    pr_status.head_sha.clone(),
-                )
-                .map_err(|e| e.to_string())?;
-
-            emit_to_all(
-                app_handle,
-                "pr-status-changed",
-                crate::prs::PrStatusEvent {
-                    branch_id: branch_id.clone(),
-                    pr_state: pr_status.state,
-                    pr_checks_status: pr_status.checks_summary.state,
-                    pr_review_decision: pr_status.review_decision,
-                    pr_mergeable: mergeable,
-                    pr_draft: pr_status.is_draft,
-                    pr_head_sha: pr_status.head_sha,
-                    pr_fetched_at,
-                    failed_checks: pr_status.failed_checks,
-                },
-            );
-
+            crate::prs::refresh_pr_status_impl(store, app_handle.clone(), branch_id).await?;
             Ok(Value::Null)
         }
         "refresh_all_pr_statuses" => {
@@ -3641,9 +3579,7 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result<Val
         "clear_branch_pr_status" => {
             let store = get_store(store_mutex)?;
             let branch_id: String = arg(&args, "branchId")?;
-            store
-                .update_branch_pr_status(&branch_id, None, None, None, None, None, None, None, None)
-                .map_err(|e| e.to_string())?;
+            crate::prs::clear_branch_pr_status_impl(&store, app_handle, &branch_id)?;
             Ok(Value::Null)
         }
 
diff --git a/apps/staged/src/lib/commands.ts b/apps/staged/src/lib/commands.ts
index 81ba745d4..d06af66fd 100644
--- a/apps/staged/src/lib/commands.ts
+++ b/apps/staged/src/lib/commands.ts
@@ -26,6 +26,7 @@ import type {
   BranchSessionType,
   BranchSessionResponse,
   StoreIncompatibility,
+  ActiveSessionInfo,
   Session,
   SessionMessage,
   QueuedSessionMessage,
@@ -760,6 +761,15 @@ export function getSession(sessionId: string): Promise<Session | null> {
   return invokeCommand('get_session', { sessionId });
 }
 
+/**
+ * Busy-state snapshot: all running and queued sessions projected to their
+ * branch/project context. Clients hydrate from this on load or reconnect,
+ * then apply `session-status-changed` deltas on top.
+ */
+export function getActiveSessions(): Promise<ActiveSessionInfo[]> {
+  return invokeCommand('get_active_sessions');
+}
+
 export function getSessionMessages(sessionId: string): Promise<SwrResult<SessionMessage[]>> {
   return cachedCommand('get_session_messages', { sessionId }, { ttl: 5 * 60_000 });
 }
diff --git a/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte b/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte
index c433e36ef..9da135c89 100644
--- a/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte
+++ b/apps/staged/src/lib/features/branches/BranchCardPrButton.svelte
@@ -510,6 +510,10 @@
 
   let prCompletionInFlight = false;
 
+  // Renderer-only fallback for completions observed outside the event stream
+  // (e.g. found finished on session-modal close). The backend parses and
+  // persists the PR number at the terminal transition and emits `pr-created`;
+  // this must not write, only mirror the outcome locally.
   export async function handlePrSessionComplete(status: string) {
     if (prCompletionInFlight) return;
     const sid = prSessionId;
@@ -524,7 +528,6 @@
         if (foundUrl) {
           const prNumber = extractPrNumber(foundUrl);
           if (prNumber) {
-            await commands.updateBranchPr(branch.id, prNumber);
             branch.prNumber = prNumber;
             prPollingService.refreshNow(branch.projectId);
           }
@@ -635,6 +638,10 @@
     }
   }
 
+  // Renderer-only fallback, like handlePrSessionComplete above: the backend
+  // classifies the push at the terminal transition, clears the stale PR
+  // status, and emits `push-completed`; this only mirrors the outcome and
+  // refreshes read-side caches.
   export async function handlePushSessionComplete(status: string, completedSession?: Session) {
     if (pushCompletionInFlight) return;
     const sid = pushSessionId;
@@ -648,11 +655,6 @@
         if (outcome === 'rejected_non_fast_forward') {
           pushStateStore.setPushError(branch.id, '', true);
         } else {
-          try {
-            await commands.clearBranchPrStatus(branch.id);
-          } catch (e) {
-            console.warn('[Staged] Failed to clear PR status after push:', e);
-          }
           pushStateStore.setPushDone(branch.id);
           // Refresh local git state so `upstream.relation` settles back to
           // `inSync` (driving hasUnpushed to false) without optimistically
diff --git a/apps/staged/src/lib/listeners/sessionStatusListener.test.ts b/apps/staged/src/lib/listeners/sessionStatusListener.test.ts
new file mode 100644
index 000000000..1f0bbbba4
--- /dev/null
+++ b/apps/staged/src/lib/listeners/sessionStatusListener.test.ts
@@ -0,0 +1,841 @@
+// @vitest-environment jsdom
+import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest';
+
+/**
+ * Busy-state hydration tests: the stores are hydrated from the
+ * `get_active_sessions` snapshot (startup / reconnect / cache-stale), with
+ * `session-status-changed` deltas applied on top.
+ *
+ * The rune-based stores can't be imported under plain vitest, so the registry
+ * is replaced with a functional in-memory fake that mirrors the real
+ * register/cleanup semantics.
+ */
+
+interface FakeMetadata {
+  sessionId: string;
+  projectId: string;
+  branchId?: string;
+  type: string;
+  timestamp: number;
+}
+
+function createFakeRegistry(projectStateStore: {
+  removeRunningSession: (projectId: string, sessionId: string) => void;
+}) {
+  const sessions = new Map<string, FakeMetadata>();
+  return {
+    sessions,
+    register: vi.fn((sessionId: string, projectId: string, type: string, branchId?: string) => {
+      sessions.set(sessionId, { sessionId, projectId, branchId, type, timestamp: Date.now() });
+    }),
+    getMetadata: (sessionId: string) => sessions.get(sessionId) ?? null,
+    getAllSessionIds: () => Array.from(sessions.keys()),
+    getProjectId: (sessionId: string) => sessions.get(sessionId)?.projectId ?? null,
+    getBranchId: (sessionId: string) => sessions.get(sessionId)?.branchId ?? null,
+    getType: (sessionId: string) => sessions.get(sessionId)?.type ?? null,
+    unregister: vi.fn((sessionId: string) => {
+      sessions.delete(sessionId);
+    }),
+    cleanupSession: vi.fn((sessionId: string) => {
+      const projectId = sessions.get(sessionId)?.projectId;
+      if (projectId) projectStateStore.removeRunningSession(projectId, sessionId);
+      sessions.delete(sessionId);
+    }),
+    clear: () => sessions.clear(),
+  };
+}
+
+/**
+ * Workflow-store fake. The setters stay plain spies (tests assert on the
+ * calls), but the two lookups the sweep depends on are backed by a real state
+ * map so their validating semantics are exercised faithfully: a branch
+ * resolves only for its own session id while still in the in-progress state.
+ */
+function createFakeWorkflowStore(
+  registry: ReturnType<typeof createFakeRegistry>,
+  inProgressState: 'creating' | 'pushing'
+) {
+  const states = new Map<string, { state: string; sessionId: string | null }>();
+  return {
+    states,
+    getBranchIdForSession: vi.fn((sessionId: string) => {
+      const branchId = registry.getBranchId(sessionId);
+      if (!branchId) return null;
+      const entry = states.get(branchId);
+      return entry?.sessionId === sessionId && entry.state === inProgressState ? branchId : null;
+    }),
+    getSessionId: vi.fn((branchId: string) => states.get(branchId)?.sessionId ?? null),
+  };
+}
+
+describe('sessionStatusListener busy-state hydration', () => {
+  let getActiveSessions: ReturnType<typeof vi.fn>;
+  let getSession: ReturnType<typeof vi.fn>;
+  let invalidateBranchTimeline: ReturnType<typeof vi.fn>;
+  let getFreshSessionMessages: ReturnType<typeof vi.fn>;
+  let updateBranchPr: ReturnType<typeof vi.fn>;
+  let refreshPrStatus: ReturnType<typeof vi.fn>;
+  let clearBranchPrStatus: ReturnType<typeof vi.fn>;
+  let listenToEvent: ReturnType<typeof vi.fn>;
+  let unlistenEvents: ReturnType<typeof vi.fn>;
+  let eventCallbacks: Map<string, (payload: unknown) => void>;
+  let projectStateStore: {
+    addRunningSession: ReturnType<typeof vi.fn>;
+    removeRunningSession: Mock<(projectId: string, sessionId: string) => void>;
+    markAsUnread: ReturnType<typeof vi.fn>;
+  };
+  let prStateStore: ReturnType<typeof createFakeWorkflowStore> & {
+    clearSessionTracking: ReturnType<typeof vi.fn>;
+    setPrCreated: ReturnType<typeof vi.fn>;
+    setPrError: ReturnType<typeof vi.fn>;
+    clearPrState: ReturnType<typeof vi.fn>;
+    getPrState: ReturnType<typeof vi.fn>;
+  };
+  let pushStateStore: ReturnType<typeof createFakeWorkflowStore> & {
+    clearSessionTracking: ReturnType<typeof vi.fn>;
+    markQueuedPushStarted: ReturnType<typeof vi.fn>;
+    setPushDone: ReturnType<typeof vi.fn>;
+    setPushError: ReturnType<typeof vi.fn>;
+    clearPushState: ReturnType<typeof vi.fn>;
+    getPushState: ReturnType<typeof vi.fn>;
+  };
+  let sessionRegistry: ReturnType<typeof createFakeRegistry>;
+
+  beforeEach(() => {
+    vi.resetModules();
+    vi.useFakeTimers({ now: 1_000 });
+
+    getActiveSessions = vi.fn().mockResolvedValue([]);
+    getSession = vi.fn().mockResolvedValue(null);
+    invalidateBranchTimeline = vi.fn();
+    getFreshSessionMessages = vi.fn().mockResolvedValue([]);
+    updateBranchPr = vi.fn().mockResolvedValue(undefined);
+    refreshPrStatus = vi.fn().mockResolvedValue(undefined);
+    clearBranchPrStatus = vi.fn().mockResolvedValue(undefined);
+    unlistenEvents = vi.fn();
+    eventCallbacks = new Map();
+    listenToEvent = vi.fn((event: string, callback: (payload: unknown) => void) => {
+      eventCallbacks.set(event, callback);
+      return unlistenEvents;
+    });
+    projectStateStore = {
+      addRunningSession: vi.fn(),
+      removeRunningSession: vi.fn<(projectId: string, sessionId: string) => void>(),
+      markAsUnread: vi.fn(),
+    };
+    sessionRegistry = createFakeRegistry(projectStateStore);
+    prStateStore = {
+      ...createFakeWorkflowStore(sessionRegistry, 'creating'),
+      clearSessionTracking: vi.fn(),
+      setPrCreated: vi.fn(),
+      setPrError: vi.fn(),
+      clearPrState: vi.fn(),
+      getPrState: vi.fn().mockReturnValue(undefined),
+    };
+    pushStateStore = {
+      ...createFakeWorkflowStore(sessionRegistry, 'pushing'),
+      clearSessionTracking: vi.fn(),
+      markQueuedPushStarted: vi.fn(),
+      setPushDone: vi.fn(),
+      setPushError: vi.fn(),
+      clearPushState: vi.fn(),
+      getPushState: vi.fn().mockReturnValue(undefined),
+    };
+
+    vi.doMock('../transport', () => ({ isTauri: true, listenToEvent }));
+    vi.doMock('../commands', () => ({
+      getActiveSessions,
+      invalidateBranchTimeline,
+      getSession,
+      getFreshSessionMessages,
+      updateBranchPr,
+      refreshPrStatus,
+      clearBranchPrStatus,
+    }));
+    vi.doMock('../features/layout/navigation.svelte', () => ({
+      navigation: { selectedProjectId: null },
+    }));
+    vi.doMock('../stores/projectState.svelte', () => ({ projectStateStore }));
+    vi.doMock('../stores/prState.svelte', () => ({ prStateStore }));
+    vi.doMock('../stores/pushState.svelte', () => ({ pushStateStore }));
+    vi.doMock('../stores/pullState.svelte', () => ({
+      pullStateStore: {
+        markQueuedPullStarted: vi.fn(),
+        clearPullState: vi.fn(),
+      },
+    }));
+    vi.doMock('svelte-sonner', () => ({ toast: { error: vi.fn() } }));
+    vi.doMock('../stores/sessionRegistry.svelte', () => ({ sessionRegistry }));
+  });
+
+  afterEach(() => {
+    vi.doUnmock('../transport');
+    vi.doUnmock('../commands');
+    vi.doUnmock('../features/layout/navigation.svelte');
+    vi.doUnmock('../stores/projectState.svelte');
+    vi.doUnmock('../stores/prState.svelte');
+    vi.doUnmock('../stores/pushState.svelte');
+    vi.doUnmock('../stores/pullState.svelte');
+    vi.doUnmock('svelte-sonner');
+    vi.doUnmock('../stores/sessionRegistry.svelte');
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it('registers unseen running sessions and applies event-path gating to the snapshot', async () => {
+    getActiveSessions.mockResolvedValue([
+      {
+        sessionId: 'running-1',
+        projectId: 'project-1',
+        branchId: 'branch-1',
+        sessionType: 'commit',
+        status: 'running',
+        isAutoReview: false,
+      },
+      {
+        sessionId: 'queued-1',
+        projectId: 'project-1',
+        branchId: 'branch-1',
+        sessionType: 'commit',
+        status: 'queued',
+        isAutoReview: false,
+      },
+      {
+        sessionId: 'unresolved-1',
+        projectId: null,
+        branchId: null,
+        sessionType: 'pr',
+        status: 'running',
+        isAutoReview: false,
+      },
+      {
+        sessionId: 'auto-review-1',
+        projectId: 'project-2',
+        branchId: 'branch-2',
+        sessionType: 'review',
+        status: 'running',
+        isAutoReview: true,
+      },
+      {
+        sessionId: 'untyped-1',
+        projectId: 'project-2',
+        branchId: null,
+        sessionType: null,
+        status: 'running',
+        isAutoReview: false,
+      },
+    ]);
+
+    const { hydrateActiveSessions } = await import('./sessionStatusListener');
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.register.mock.calls).toEqual([
+      ['running-1', 'project-1', 'commit', 'branch-1'],
+      ['untyped-1', 'project-2', 'other', undefined],
+    ]);
+    expect(projectStateStore.addRunningSession.mock.calls).toEqual([
+      ['project-1', 'running-1'],
+      ['project-2', 'untyped-1'],
+    ]);
+  });
+
+  it('keeps existing local metadata for sessions the snapshot also reports', async () => {
+    // A pipeline (pr) session registered at launch with its real branch and
+    // project — the snapshot reports it running but with the same id; local
+    // knowledge must win.
+    sessionRegistry.sessions.set('pr-1', {
+      sessionId: 'pr-1',
+      projectId: 'project-1',
+      branchId: 'branch-1',
+      type: 'pr',
+      timestamp: 500,
+    });
+    getActiveSessions.mockResolvedValue([
+      {
+        sessionId: 'pr-1',
+        projectId: 'project-1',
+        branchId: null,
+        sessionType: 'other',
+        status: 'running',
+        isAutoReview: false,
+      },
+    ]);
+
+    const { hydrateActiveSessions } = await import('./sessionStatusListener');
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.register).not.toHaveBeenCalled();
+    expect(sessionRegistry.cleanupSession).not.toHaveBeenCalled();
+    expect(projectStateStore.addRunningSession).not.toHaveBeenCalled();
+    expect(sessionRegistry.getMetadata('pr-1')).toMatchObject({ branchId: 'branch-1', type: 'pr' });
+  });
+
+  it('sweeps local entries the backend no longer reports as active', async () => {
+    // A running entry whose terminal event this client missed — this is the
+    // stuck-spinner case the snapshot heals.
+    sessionRegistry.sessions.set('gone-1', {
+      sessionId: 'gone-1',
+      projectId: 'project-1',
+      type: 'commit',
+      timestamp: 500,
+    });
+    getActiveSessions.mockResolvedValue([]);
+
+    const { hydrateActiveSessions } = await import('./sessionStatusListener');
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.cleanupSession).toHaveBeenCalledWith('gone-1');
+    expect(projectStateStore.removeRunningSession).toHaveBeenCalledWith('project-1', 'gone-1');
+    expect(sessionRegistry.getMetadata('gone-1')).toBeNull();
+  });
+
+  it('does not sweep entries registered while the snapshot fetch was in flight', async () => {
+    sessionRegistry.sessions.set('stale-1', {
+      sessionId: 'stale-1',
+      projectId: 'project-1',
+      type: 'commit',
+      timestamp: 500,
+    });
+    vi.setSystemTime(2_000);
+    getActiveSessions.mockImplementation(async () => {
+      // An optimistic launch-site registration racing the fetch: the session
+      // is newer than the snapshot, so the sweep must keep it.
+      sessionRegistry.register('launched-mid-fetch', 'project-2', 'commit', 'branch-2');
+      return [];
+    });
+
+    const { hydrateActiveSessions } = await import('./sessionStatusListener');
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.cleanupSession).toHaveBeenCalledWith('stale-1');
+    expect(sessionRegistry.cleanupSession).not.toHaveBeenCalledWith('launched-mid-fetch');
+    expect(sessionRegistry.getMetadata('launched-mid-fetch')).not.toBeNull();
+  });
+
+  it('does not re-register sessions whose terminal delta arrived while the fetch was in flight', async () => {
+    const { listenForSessionStatus, hydrateActiveSessions } =
+      await import('./sessionStatusListener');
+    listenForSessionStatus();
+    await vi.waitFor(() => expect(getActiveSessions).toHaveBeenCalledTimes(1));
+
+    // A session running and registered via the delta path...
+    eventCallbacks.get('session-status-changed')?.({
+      sessionId: 'racy-1',
+      status: 'running',
+      projectId: 'project-1',
+      branchId: 'branch-1',
+      sessionType: 'commit',
+    });
+    expect(sessionRegistry.getMetadata('racy-1')).not.toBeNull();
+
+    vi.setSystemTime(2_000);
+    getActiveSessions.mockImplementation(async () => {
+      // ...whose terminal delta lands while a snapshot that still reports it
+      // running is in flight. The register loop must not resurrect it from
+      // the stale snapshot — that would recreate the stuck spinner.
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'racy-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+      return [
+        {
+          sessionId: 'racy-1',
+          projectId: 'project-1',
+          branchId: 'branch-1',
+          sessionType: 'commit',
+          status: 'running',
+          isAutoReview: false,
+        },
+      ];
+    });
+
+    sessionRegistry.register.mockClear();
+    projectStateStore.addRunningSession.mockClear();
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.register).not.toHaveBeenCalled();
+    expect(projectStateStore.addRunningSession).not.toHaveBeenCalled();
+    expect(sessionRegistry.getMetadata('racy-1')).toBeNull();
+  });
+
+  it('lets a later hydration register a session terminated during an earlier fetch', async () => {
+    const { listenForSessionStatus, hydrateActiveSessions } =
+      await import('./sessionStatusListener');
+    listenForSessionStatus();
+    await vi.waitFor(() => expect(getActiveSessions).toHaveBeenCalledTimes(1));
+
+    vi.setSystemTime(2_000);
+    getActiveSessions.mockImplementation(async () => {
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'racy-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+      return [];
+    });
+    await hydrateActiveSessions();
+
+    // The session was resumed backend-side after the terminal event: a fresh
+    // snapshot legitimately reports it running again, and the guard from the
+    // settled hydration must not suppress it.
+    vi.setSystemTime(3_000);
+    getActiveSessions.mockResolvedValue([
+      {
+        sessionId: 'racy-1',
+        projectId: 'project-1',
+        branchId: 'branch-1',
+        sessionType: 'commit',
+        status: 'running',
+        isAutoReview: false,
+      },
+    ]);
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.register).toHaveBeenCalledWith(
+      'racy-1',
+      'project-1',
+      'commit',
+      'branch-1'
+    );
+    expect(sessionRegistry.getMetadata('racy-1')).not.toBeNull();
+  });
+
+  it('leaves state untouched when the snapshot fetch fails', async () => {
+    sessionRegistry.sessions.set('running-1', {
+      sessionId: 'running-1',
+      projectId: 'project-1',
+      type: 'commit',
+      timestamp: 500,
+    });
+    getActiveSessions.mockRejectedValue(new Error('store not ready'));
+    const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
+
+    const { hydrateActiveSessions } = await import('./sessionStatusListener');
+    await hydrateActiveSessions();
+
+    expect(sessionRegistry.cleanupSession).not.toHaveBeenCalled();
+    expect(sessionRegistry.getMetadata('running-1')).not.toBeNull();
+    expect(consoleError).toHaveBeenCalled();
+  });
+
+  it('hydrates on start, re-hydrates on cache-stale, and detaches on unlisten', async () => {
+    const { listenForSessionStatus } = await import('./sessionStatusListener');
+
+    const unlisten = listenForSessionStatus();
+    expect(listenToEvent).toHaveBeenCalledWith('session-status-changed', expect.any(Function));
+    expect(listenToEvent).toHaveBeenCalledWith('pr-created', expect.any(Function));
+    expect(listenToEvent).toHaveBeenCalledWith('push-completed', expect.any(Function));
+    await vi.waitFor(() => expect(getActiveSessions).toHaveBeenCalledTimes(1));
+
+    window.dispatchEvent(new CustomEvent('cache-stale'));
+    await vi.waitFor(() => expect(getActiveSessions).toHaveBeenCalledTimes(2));
+
+    unlisten();
+    expect(unlistenEvents).toHaveBeenCalledTimes(3);
+    window.dispatchEvent(new CustomEvent('cache-stale'));
+    expect(getActiveSessions).toHaveBeenCalledTimes(2);
+  });
+
+  it('still applies session-status-changed deltas on top of the snapshot', async () => {
+    const { listenForSessionStatus } = await import('./sessionStatusListener');
+    listenForSessionStatus();
+    await vi.waitFor(() => expect(getActiveSessions).toHaveBeenCalledTimes(1));
+
+    eventCallbacks.get('session-status-changed')?.({
+      sessionId: 'delta-1',
+      status: 'running',
+      projectId: 'project-1',
+      branchId: 'branch-1',
+      sessionType: 'commit',
+    });
+    expect(sessionRegistry.register).toHaveBeenCalledWith(
+      'delta-1',
+      'project-1',
+      'commit',
+      'branch-1'
+    );
+    expect(projectStateStore.addRunningSession).toHaveBeenCalledWith('project-1', 'delta-1');
+
+    eventCallbacks.get('session-status-changed')?.({
+      sessionId: 'delta-1',
+      status: 'completed',
+      branchId: 'branch-1',
+    });
+    await vi.waitFor(() => expect(sessionRegistry.cleanupSession).toHaveBeenCalledWith('delta-1'));
+    expect(invalidateBranchTimeline).toHaveBeenCalledWith('branch-1');
+  });
+
+  // -------------------------------------------------------------------------
+  // Swept workflow chips: a client offline through a pipeline session's whole
+  // completion missed both the domain event and the terminal event, so the
+  // sweep reconciles its chip against the session's persisted status.
+  // -------------------------------------------------------------------------
+
+  describe('swept workflow reconciliation', () => {
+    /** A pipeline session the snapshot will no longer report, chip still in progress. */
+    function trackWorkflowSession(kind: 'pr' | 'push', sessionId: string, branchId = 'branch-1') {
+      sessionRegistry.sessions.set(sessionId, {
+        sessionId,
+        projectId: 'project-1',
+        branchId,
+        type: kind,
+        timestamp: 500,
+      });
+      const store = kind === 'pr' ? prStateStore : pushStateStore;
+      store.states.set(branchId, { state: kind === 'pr' ? 'creating' : 'pushing', sessionId });
+    }
+
+    async function hydrate() {
+      const { hydrateActiveSessions } = await import('./sessionStatusListener');
+      await hydrateActiveSessions();
+    }
+
+    it('clears a swept pr chip whose session completed, letting the branch row drive it', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      getSession.mockResolvedValue({ id: 'pr-1', status: 'completed' });
+
+      await hydrate();
+
+      expect(getSession).toHaveBeenCalledWith('pr-1');
+      // Not `setPrError`: the client missed `pr-created`, but the backend
+      // persisted the PR number — a false "no PR URL found" would stick.
+      expect(prStateStore.clearPrState).toHaveBeenCalledWith('branch-1');
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+      expect(sessionRegistry.cleanupSession).toHaveBeenCalledWith('pr-1');
+    });
+
+    it('renders the delta-path error copy for a swept pr session that failed', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      getSession.mockResolvedValue({ id: 'pr-1', status: 'error' });
+
+      await hydrate();
+
+      expect(prStateStore.setPrError).toHaveBeenCalledWith(
+        'branch-1',
+        'PR creation session failed.'
+      );
+      expect(prStateStore.clearSessionTracking).toHaveBeenCalledWith('branch-1');
+      expect(prStateStore.clearPrState).not.toHaveBeenCalled();
+    });
+
+    it('clears a swept push chip whose session completed without a done flash', async () => {
+      trackWorkflowSession('push', 'push-1');
+      getSession.mockResolvedValue({ id: 'push-1', status: 'completed' });
+
+      await hydrate();
+
+      expect(pushStateStore.clearPushState).toHaveBeenCalledWith('branch-1');
+      expect(pushStateStore.setPushDone).not.toHaveBeenCalled();
+      expect(pushStateStore.setPushError).not.toHaveBeenCalled();
+    });
+
+    it('renders the cancellation for a swept push session', async () => {
+      trackWorkflowSession('push', 'push-1');
+      getSession.mockResolvedValue({ id: 'push-1', status: 'cancelled' });
+
+      await hydrate();
+
+      expect(pushStateStore.setPushError).toHaveBeenCalledWith(
+        'branch-1',
+        'Push session was cancelled.'
+      );
+      expect(pushStateStore.clearSessionTracking).toHaveBeenCalledWith('branch-1');
+      expect(pushStateStore.clearPushState).not.toHaveBeenCalled();
+    });
+
+    it('errors out a swept pr chip when the session has vanished', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      getSession.mockResolvedValue(null);
+
+      await hydrate();
+
+      expect(prStateStore.setPrError).toHaveBeenCalledWith(
+        'branch-1',
+        'Lost track of PR creation session.'
+      );
+      expect(prStateStore.clearSessionTracking).toHaveBeenCalledWith('branch-1');
+    });
+
+    it('leaves a swept chip in progress when the session lookup throws', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      getSession.mockRejectedValue(new Error('socket not ready'));
+      const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
+
+      await hydrate();
+
+      // A throw is a transport failure, not proof the session is gone — most
+      // likely right after the WebSocket reconnect that triggered hydration.
+      // The sticky "Lost track of…" error would outlive a PR that actually
+      // succeeded, so keep the chip tracking for the card poller to heal.
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+      expect(prStateStore.clearPrState).not.toHaveBeenCalled();
+      expect(prStateStore.clearSessionTracking).not.toHaveBeenCalled();
+      expect(consoleError).toHaveBeenCalled();
+    });
+
+    it('errors out a swept push chip when the session has vanished', async () => {
+      trackWorkflowSession('push', 'push-1');
+      getSession.mockResolvedValue(null);
+
+      await hydrate();
+
+      expect(pushStateStore.setPushError).toHaveBeenCalledWith(
+        'branch-1',
+        'Lost track of push session.'
+      );
+      expect(pushStateStore.clearSessionTracking).toHaveBeenCalledWith('branch-1');
+    });
+
+    it('leaves the chip alone when the backend resumed the session', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      getSession.mockResolvedValue({ id: 'pr-1', status: 'running' });
+
+      await hydrate();
+
+      expect(prStateStore.clearPrState).not.toHaveBeenCalled();
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+      expect(prStateStore.clearSessionTracking).not.toHaveBeenCalled();
+    });
+
+    it('skips a chip already tracking a different session', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      // The user relaunched before hydration ran: the chip belongs to pr-2 now.
+      prStateStore.states.set('branch-1', { state: 'creating', sessionId: 'pr-2' });
+
+      await hydrate();
+
+      expect(getSession).not.toHaveBeenCalled();
+      expect(prStateStore.clearPrState).not.toHaveBeenCalled();
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+    });
+
+    it('skips a chip relaunched while the session lookup was in flight', async () => {
+      trackWorkflowSession('pr', 'pr-1');
+      getSession.mockImplementation(async () => {
+        prStateStore.states.set('branch-1', { state: 'creating', sessionId: 'pr-2' });
+        return { id: 'pr-1', status: 'completed' };
+      });
+
+      await hydrate();
+
+      expect(prStateStore.clearPrState).not.toHaveBeenCalled();
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+    });
+
+    it('ignores swept sessions no workflow chip is tracking', async () => {
+      sessionRegistry.sessions.set('commit-1', {
+        sessionId: 'commit-1',
+        projectId: 'project-1',
+        branchId: 'branch-1',
+        type: 'other',
+        timestamp: 500,
+      });
+
+      await hydrate();
+
+      expect(getSession).not.toHaveBeenCalled();
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+      expect(pushStateStore.setPushError).not.toHaveBeenCalled();
+      expect(sessionRegistry.cleanupSession).toHaveBeenCalledWith('commit-1');
+    });
+  });
+
+  // -------------------------------------------------------------------------
+  // Completion rendering: the backend parses/persists outcomes at the terminal
+  // transition and emits `pr-created` / `push-completed` before the terminal
+  // status event; these handlers only render and never write.
+  // -------------------------------------------------------------------------
+
+  describe('completion domain events', () => {
+    async function listen() {
+      const { listenForSessionStatus } = await import('./sessionStatusListener');
+      listenForSessionStatus();
+      await vi.waitFor(() => expect(getActiveSessions).toHaveBeenCalledTimes(1));
+    }
+
+    function registerSession(sessionId: string, type: string, branchId = 'branch-1') {
+      sessionRegistry.sessions.set(sessionId, {
+        sessionId,
+        projectId: 'project-1',
+        branchId,
+        type,
+        timestamp: 500,
+      });
+    }
+
+    it('renders pr-created by marking the branch created', async () => {
+      await listen();
+
+      eventCallbacks.get('pr-created')?.({
+        branchId: 'branch-1',
+        sessionId: 'pr-1',
+        prUrl: 'https://github.com/org/repo/pull/42',
+        prNumber: 42,
+      });
+
+      expect(prStateStore.setPrCreated).toHaveBeenCalledWith(
+        'branch-1',
+        'https://github.com/org/repo/pull/42'
+      );
+    });
+
+    it('renders push-completed success and clears it after the done flash', async () => {
+      await listen();
+
+      eventCallbacks.get('push-completed')?.({
+        branchId: 'branch-1',
+        sessionId: 'push-1',
+        outcome: 'succeeded',
+      });
+
+      expect(pushStateStore.setPushDone).toHaveBeenCalledWith('branch-1');
+      expect(pushStateStore.setPushError).not.toHaveBeenCalled();
+      vi.advanceTimersByTime(1_500);
+      expect(pushStateStore.clearPushState).toHaveBeenCalledWith('branch-1');
+    });
+
+    it('renders push-completed rejection as a non-fast-forward error', async () => {
+      await listen();
+
+      eventCallbacks.get('push-completed')?.({
+        branchId: 'branch-1',
+        sessionId: 'push-1',
+        outcome: 'rejectedNonFastForward',
+      });
+
+      expect(pushStateStore.setPushError).toHaveBeenCalledWith('branch-1', '', true);
+      expect(pushStateStore.setPushDone).not.toHaveBeenCalled();
+    });
+
+    it('leaves a created branch alone when the pr session completes', async () => {
+      await listen();
+      registerSession('pr-1', 'pr');
+      prStateStore.getPrState.mockReturnValue({ state: 'created' });
+
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'pr-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+
+      expect(prStateStore.setPrError).not.toHaveBeenCalled();
+      expect(prStateStore.clearSessionTracking).toHaveBeenCalledWith('branch-1');
+    });
+
+    it('reports a missing PR URL when completion arrives without pr-created', async () => {
+      await listen();
+      registerSession('pr-1', 'pr');
+
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'pr-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+
+      expect(prStateStore.setPrError).toHaveBeenCalledWith(
+        'branch-1',
+        'PR session completed but no PR URL was found in the output.'
+      );
+    });
+
+    it('renders terminal pr failures and cancellations', async () => {
+      await listen();
+      registerSession('pr-1', 'pr');
+      registerSession('pr-2', 'pr', 'branch-2');
+
+      eventCallbacks.get('session-status-changed')?.({ sessionId: 'pr-1', status: 'error' });
+      eventCallbacks.get('session-status-changed')?.({ sessionId: 'pr-2', status: 'cancelled' });
+
+      expect(prStateStore.setPrError).toHaveBeenCalledWith(
+        'branch-1',
+        'PR creation session failed.'
+      );
+      expect(prStateStore.setPrError).toHaveBeenCalledWith(
+        'branch-2',
+        'PR creation session was cancelled.'
+      );
+    });
+
+    it('falls back to success rendering when the push-completed event was missed', async () => {
+      await listen();
+      registerSession('push-1', 'push');
+      pushStateStore.getPushState.mockReturnValue({ state: 'pushing' });
+
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'push-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+
+      expect(pushStateStore.setPushDone).toHaveBeenCalledWith('branch-1');
+      vi.advanceTimersByTime(1_500);
+      expect(pushStateStore.clearPushState).toHaveBeenCalledWith('branch-1');
+    });
+
+    it('does not re-render a push completion already handled by push-completed', async () => {
+      await listen();
+      registerSession('push-1', 'push');
+      pushStateStore.getPushState.mockReturnValue({ state: 'done' });
+
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'push-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+
+      expect(pushStateStore.setPushDone).not.toHaveBeenCalled();
+      expect(pushStateStore.setPushError).not.toHaveBeenCalled();
+      expect(pushStateStore.clearSessionTracking).toHaveBeenCalledWith('branch-1');
+    });
+
+    it('renders terminal push failures and cancellations', async () => {
+      await listen();
+      registerSession('push-1', 'push');
+      registerSession('push-2', 'push', 'branch-2');
+
+      eventCallbacks.get('session-status-changed')?.({ sessionId: 'push-1', status: 'error' });
+      eventCallbacks.get('session-status-changed')?.({ sessionId: 'push-2', status: 'cancelled' });
+
+      expect(pushStateStore.setPushError).toHaveBeenCalledWith('branch-1', 'Push session failed.');
+      expect(pushStateStore.setPushError).toHaveBeenCalledWith(
+        'branch-2',
+        'Push session was cancelled.'
+      );
+    });
+
+    it('never performs authoritative writes from completion handling', async () => {
+      await listen();
+      registerSession('pr-1', 'pr');
+      registerSession('push-1', 'push', 'branch-2');
+      pushStateStore.getPushState.mockReturnValue({ state: 'pushing' });
+
+      eventCallbacks.get('pr-created')?.({
+        branchId: 'branch-1',
+        sessionId: 'pr-1',
+        prUrl: 'https://github.com/org/repo/pull/42',
+        prNumber: 42,
+      });
+      eventCallbacks.get('push-completed')?.({
+        branchId: 'branch-2',
+        sessionId: 'push-1',
+        outcome: 'succeeded',
+      });
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'pr-1',
+        status: 'completed',
+        branchId: 'branch-1',
+      });
+      eventCallbacks.get('session-status-changed')?.({
+        sessionId: 'push-1',
+        status: 'completed',
+        branchId: 'branch-2',
+      });
+
+      expect(updateBranchPr).not.toHaveBeenCalled();
+      expect(refreshPrStatus).not.toHaveBeenCalled();
+      expect(clearBranchPrStatus).not.toHaveBeenCalled();
+      expect(getFreshSessionMessages).not.toHaveBeenCalled();
+    });
+  });
+});
diff --git a/apps/staged/src/lib/listeners/sessionStatusListener.ts b/apps/staged/src/lib/listeners/sessionStatusListener.ts
index d6bb73047..844db0223 100644
--- a/apps/staged/src/lib/listeners/sessionStatusListener.ts
+++ b/apps/staged/src/lib/listeners/sessionStatusListener.ts
@@ -8,78 +8,323 @@
  * 4. pullState     — branch-specific queued-pull state (pull footer row)
  *
  * Session lookups are delegated to the unified sessionRegistry for consistency.
+ *
+ * Busy state follows a snapshot-then-deltas model: the backend is the source
+ * of truth, and this module hydrates the registry from the
+ * `get_active_sessions` snapshot (on startup, on WebSocket reconnect via
+ * transport.ts, and on the page-lifecycle `cache-stale` resume signal), then
+ * applies `session-status-changed` deltas on top. Sessions the snapshot
+ * proves dead are swept, and a swept session still rendering a pr/push
+ * workflow chip is reconciled against its persisted terminal status — a
+ * client that was offline through the whole completion missed both the
+ * domain event and the terminal event, so the chip would otherwise stay
+ * stuck in "Creating PR…" / "Pushing…". Unread state is the only client-local
+ * busy-state signal left untouched.
+ *
+ * Completion outcomes are parsed and persisted by the backend at the terminal
+ * transition (PR number, cleared PR status) and arrive as `pr-created` /
+ * `push-completed` domain events emitted *before* the terminal
+ * `session-status-changed` event. The handlers here are idempotent renderers
+ * — they perform no writes, so any number of connected clients can process
+ * the same events safely.
  */
 
 import { toast } from 'svelte-sonner';
 import { listenToEvent, type UnlistenFn } from '../transport';
 import { invalidateBranchTimeline } from '../commands';
 import * as commands from '../api/commands';
-import {
-  classifyCompletedPushSession,
-  extractPrUrl,
-  extractPrNumber,
-} from '../features/branches/branchCardHelpers';
 import { navigation } from '../features/layout/navigation.svelte';
 import { projectStateStore } from '../stores/projectState.svelte';
 import { prStateStore } from '../stores/prState.svelte';
 import { pullStateStore } from '../stores/pullState.svelte';
 import { pushStateStore } from '../stores/pushState.svelte';
 import { sessionRegistry, type SessionType } from '../stores/sessionRegistry.svelte';
-import type { SessionStatus, SessionStatusPayload } from '../types';
+import type {
+  ActiveSessionInfo,
+  PrCreatedPayload,
+  PushCompletedPayload,
+  SessionStatus,
+  SessionStatusPayload,
+} from '../types';
+
+// Terminal deltas processed while a snapshot fetch is in flight are newer
+// than that snapshot, which may still report the session as running — the
+// register loop in hydrateActiveSessions must not resurrect them. Entries are
+// keyed by arrival time so overlapping hydrations each compare against their
+// own fetch start; the map is cleared once no fetch is in flight.
+let hydrationFetchesInFlight = 0;
+const terminalWhileFetching = new Map<string, number>();
 
 export function listenForSessionStatus(): UnlistenFn {
-  return listenToEvent<SessionStatusPayload>('session-status-changed', async (payload) => {
-    const {
+  const unlistenEvents = listenToEvent<SessionStatusPayload>(
+    'session-status-changed',
+    handleSessionStatusChanged
+  );
+  const unlistenPrCreated = listenToEvent<PrCreatedPayload>('pr-created', handlePrCreated);
+  const unlistenPushCompleted = listenToEvent<PushCompletedPayload>(
+    'push-completed',
+    handlePushCompleted
+  );
+
+  // Snapshot-then-deltas: hydrate now (app/page startup) and again after a
+  // long hide (the page-lifecycle `cache-stale` resume signal). WebSocket
+  // reconnects re-hydrate from transport.ts, next to the PR-poll interest
+  // replay.
+  void hydrateActiveSessions();
+  const onCacheStale = () => void hydrateActiveSessions();
+  window.addEventListener('cache-stale', onCacheStale);
+
+  return () => {
+    window.removeEventListener('cache-stale', onCacheStale);
+    unlistenEvents();
+    unlistenPrCreated();
+    unlistenPushCompleted();
+  };
+}
+
+async function handleSessionStatusChanged(payload: SessionStatusPayload): Promise<void> {
+  const {
+    sessionId,
+    status,
+    errorMessage,
+    branchId: eventBranchId,
+    projectId: eventProjectId,
+    sessionType,
+    isAutoReview,
+  } = payload;
+
+  // Auto review sessions are handled by BranchCard — don't register them
+  // here so they don't cause the project list spinner. When the user
+  // adopts an auto review, BranchCard registers the session at that point.
+  if (status === 'running' && eventProjectId && !isAutoReview) {
+    sessionRegistry.register(
       sessionId,
-      status,
-      errorMessage,
-      branchId: eventBranchId,
-      projectId: eventProjectId,
-      sessionType,
-      isAutoReview,
-    } = payload;
-
-    // Auto review sessions are handled by BranchCard — don't register them
-    // here so they don't cause the project list spinner. When the user
-    // adopts an auto review, BranchCard registers the session at that point.
-    if (status === 'running' && eventProjectId && !isAutoReview) {
-      sessionRegistry.register(
-        sessionId,
-        eventProjectId,
-        (sessionType as SessionType) ?? 'other',
-        eventBranchId
-      );
-      projectStateStore.addRunningSession(eventProjectId, sessionId);
-      // A push or pull that was queued behind other branch work starts running
-      // when the branch queue drains it; this event is the only signal of that.
-      if (sessionType === 'push' && eventBranchId) {
-        pushStateStore.markQueuedPushStarted(eventBranchId, sessionId);
-      }
-      if (sessionType === 'pull' && eventBranchId) {
-        pullStateStore.markQueuedPullStarted(eventBranchId, sessionId);
-      }
+      eventProjectId,
+      (sessionType as SessionType) ?? 'other',
+      eventBranchId
+    );
+    projectStateStore.addRunningSession(eventProjectId, sessionId);
+    // A push or pull that was queued behind other branch work starts running
+    // when the branch queue drains it; this event is the only signal of that.
+    if (sessionType === 'push' && eventBranchId) {
+      pushStateStore.markQueuedPushStarted(eventBranchId, sessionId);
+    }
+    if (sessionType === 'pull' && eventBranchId) {
+      pullStateStore.markQueuedPullStarted(eventBranchId, sessionId);
+    }
+    return;
+  }
+
+  if (status === 'completed' || status === 'error' || status === 'cancelled') {
+    if (hydrationFetchesInFlight > 0) {
+      terminalWhileFetching.set(sessionId, Date.now());
+    }
+    // Invalidate cached timeline for the branch affected by this session
+    if (eventBranchId) {
+      invalidateBranchTimeline(eventBranchId);
+    }
+    handleSessionEnd(sessionId, status, errorMessage);
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Snapshot hydration
+// ---------------------------------------------------------------------------
+
+/**
+ * Hydrate the busy-state stores from the backend's `get_active_sessions`
+ * snapshot.
+ *
+ * The snapshot is authoritative for *which* sessions are active: local
+ * entries the backend no longer reports are swept, which is what heals a
+ * stuck spinner after a missed terminal event. Per-session metadata prefers
+ * what the client already knows: launch sites register pipeline (pr/push)
+ * sessions with their real branch/project, which the snapshot cannot resolve
+ * (they link no artifact), and BranchCard registers adopted auto reviews.
+ * Snapshot entries the client has never seen are applied with the same
+ * gating as the live `running` event: running, resolved project, not an
+ * auto review. Queued sessions register when their own running event
+ * arrives. Unread state is per-device UX state and is left untouched.
+ *
+ * Swept sessions that a workflow store is still rendering as in-progress are
+ * reconciled against their persisted status once the snapshot window closes
+ * (see `reconcileSweptWorkflowSession`).
+ */
+export async function hydrateActiveSessions(): Promise<void> {
+  // Anything that happens while the fetch is in flight is newer than the
+  // snapshot, in both directions: entries registered mid-fetch (optimistic
+  // launch-site registrations) must not be swept, and sessions whose terminal
+  // delta was processed mid-fetch must not be re-registered from the
+  // snapshot's stale "running" claim.
+  const fetchStartedAt = Date.now();
+  const sweptWorkflows: SweptWorkflowSession[] = [];
+  hydrationFetchesInFlight++;
+  try {
+    let active: ActiveSessionInfo[];
+    try {
+      active = await commands.getActiveSessions();
+    } catch (e) {
+      console.error('Failed to fetch active-sessions snapshot:', e);
       return;
     }
 
-    if (status === 'completed' || status === 'error' || status === 'cancelled') {
-      // Invalidate cached timeline for the branch affected by this session
-      if (eventBranchId) {
-        invalidateBranchTimeline(eventBranchId);
+    const activeIds = new Set(active.map((session) => session.sessionId));
+    for (const sessionId of sessionRegistry.getAllSessionIds()) {
+      const registeredAt = sessionRegistry.getMetadata(sessionId)?.timestamp ?? 0;
+      if (!activeIds.has(sessionId) && registeredAt < fetchStartedAt) {
+        // Collect before cleanupSession, which destroys the registry metadata
+        // the store lookups resolve the branch through. Both lookups only
+        // answer for their own session id in its in-progress state, so
+        // settled chips are never collected.
+        const prBranchId = prStateStore.getBranchIdForSession(sessionId);
+        if (prBranchId) sweptWorkflows.push({ sessionId, kind: 'pr', branchId: prBranchId });
+        const pushBranchId = pushStateStore.getBranchIdForSession(sessionId);
+        if (pushBranchId) sweptWorkflows.push({ sessionId, kind: 'push', branchId: pushBranchId });
+        sessionRegistry.cleanupSession(sessionId);
       }
-      handleSessionEnd(sessionId, status, errorMessage);
     }
-  });
+
+    for (const session of active) {
+      if (session.status !== 'running') continue;
+      if (!session.projectId || session.isAutoReview) continue;
+      if (sessionRegistry.getMetadata(session.sessionId)) continue;
+      if ((terminalWhileFetching.get(session.sessionId) ?? 0) >= fetchStartedAt) continue;
+      sessionRegistry.register(
+        session.sessionId,
+        session.projectId,
+        (session.sessionType as SessionType) ?? 'other',
+        session.branchId ?? undefined
+      );
+      projectStateStore.addRunningSession(session.projectId, session.sessionId);
+    }
+  } finally {
+    hydrationFetchesInFlight--;
+    if (hydrationFetchesInFlight === 0) {
+      terminalWhileFetching.clear();
+    }
+  }
+
+  // Deliberately outside the in-flight window: the counter and the
+  // `terminalWhileFetching` guard cover the snapshot fetch/apply only, and
+  // these lookups must not extend it. Awaited so callers (and tests) can
+  // observe the reconciliation.
+  await Promise.all(sweptWorkflows.map(reconcileSweptWorkflowSession));
+}
+
+interface SweptWorkflowSession {
+  sessionId: string;
+  kind: 'pr' | 'push';
+  branchId: string;
+}
+
+/**
+ * Heal a pr/push workflow chip whose session the sweep just proved dead.
+ *
+ * A client offline through a pipeline session's entire completion misses both
+ * the `pr-created` / `push-completed` domain event and the terminal
+ * `session-status-changed` event, so its chip is still rendering
+ * "Creating PR…" / "Pushing…" for a session that finished long ago. The
+ * delta-path reconcilers can't be reused here: they read the ordered event
+ * stream, so `handlePrCompletion` would render "no PR URL was found" for a PR
+ * that actually succeeded.
+ *
+ * Instead, look up the session's persisted status (one `getSession` per
+ * genuinely stuck chip — normally zero) and, on `completed`, drop the stale
+ * workflow state rather than re-deriving an outcome: the backend persisted
+ * the real one at the terminal transition, so the branch row drives the chip
+ * (PR number → created, none → idle) and the push chip returns to its
+ * git-state-derived affordance. `error` / `cancelled` are unambiguous and
+ * render the same copy as the delta path. Only a null row — the session is
+ * genuinely gone — errors out as "Lost track of …"; a *thrown* lookup is a
+ * transport failure and skips reconciliation entirely (see the catch below).
+ *
+ * Race safety comes from re-checking the tracked session id after the await:
+ * a terminal delta clears it, and a relaunch replaces it — either way this
+ * reconciliation is stale and skips. Overlapping hydrations can't
+ * double-reconcile, since the first sweep removes the registry entry the
+ * second's collection step resolves the branch through.
+ */
+async function reconcileSweptWorkflowSession({
+  sessionId,
+  kind,
+  branchId,
+}: SweptWorkflowSession): Promise<void> {
+  let status: SessionStatus | null;
+  try {
+    status = (await commands.getSession(sessionId))?.status ?? null;
+  } catch (e) {
+    // A thrown lookup is a transport failure, not evidence about the session
+    // — and this runs right after a WebSocket reconnect, when a hiccup is
+    // most likely. The registry entry is already swept, so no later hydration
+    // retries this branch; painting the sticky "Lost track of…" error here
+    // would turn one failed round-trip into a permanent false failure. Leave
+    // the chip in progress for the mounted card poller to heal instead, and
+    // reserve the hard error for a lookup that proves the row is gone.
+    console.error('Failed to look up swept workflow session:', sessionId, e);
+    return;
+  }
+
+  const store = kind === 'pr' ? prStateStore : pushStateStore;
+  if (store.getSessionId(branchId) !== sessionId) return;
+
+  // The backend resumed it between the snapshot and this lookup — leave the
+  // chip alone; its running event re-registers the session.
+  if (status === 'running' || status === 'queued') return;
+
+  if (kind === 'pr') {
+    if (status === 'completed') {
+      prStateStore.clearPrState(branchId);
+    } else if (status) {
+      handlePrCompletion(branchId, status);
+      prStateStore.clearSessionTracking(branchId);
+    } else {
+      prStateStore.setPrError(branchId, 'Lost track of PR creation session.');
+      prStateStore.clearSessionTracking(branchId);
+    }
+  } else {
+    if (status === 'completed') {
+      // No `done` flash: this completion may be arbitrarily old, and the
+      // outcome was already classified and persisted server-side.
+      pushStateStore.clearPushState(branchId);
+    } else if (status) {
+      pushStateStore.setPushError(
+        branchId,
+        `Push session ${status === 'error' ? 'failed' : 'was cancelled'}.`
+      );
+      pushStateStore.clearSessionTracking(branchId);
+    } else {
+      pushStateStore.setPushError(branchId, 'Lost track of push session.');
+      pushStateStore.clearSessionTracking(branchId);
+    }
+  }
 }
 
 // ---------------------------------------------------------------------------
-// Completion sub-handlers
+// Completion sub-handlers (idempotent renderers — the backend owns the writes)
 // ---------------------------------------------------------------------------
 
-async function handleSessionEnd(
-  sessionId: string,
-  status: SessionStatus,
-  errorMessage?: string | null
-) {
+/**
+ * Render a PR produced by a completed PR session. The backend already
+ * persisted the PR number and kicked off a status refresh; the fresh status
+ * arrives via the existing `pr-status-changed` event.
+ */
+function handlePrCreated(payload: PrCreatedPayload): void {
+  prStateStore.setPrCreated(payload.branchId, payload.prUrl);
+}
+
+/** Render the backend-classified outcome of a completed push session. */
+function handlePushCompleted(payload: PushCompletedPayload): void {
+  if (payload.outcome === 'rejectedNonFastForward') {
+    pushStateStore.setPushError(payload.branchId, '', true);
+  } else {
+    pushStateStore.setPushDone(payload.branchId);
+    setTimeout(() => {
+      pushStateStore.clearPushState(payload.branchId);
+    }, 1_500);
+  }
+}
+
+function handleSessionEnd(sessionId: string, status: SessionStatus, errorMessage?: string | null) {
   const sessionProjectId = sessionRegistry.getProjectId(sessionId);
   const sessionType = sessionRegistry.getType(sessionId);
   const branchId = sessionRegistry.getBranchId(sessionId);
@@ -95,12 +340,12 @@ async function handleSessionEnd(
   }
 
   if (sessionType === 'pr' && branchId) {
-    await handlePrCompletion(sessionId, branchId, status);
+    handlePrCompletion(branchId, status);
     prStateStore.clearSessionTracking(branchId);
   }
 
   if (sessionType === 'push' && branchId) {
-    await handlePushCompletion(sessionId, branchId, status);
+    handlePushCompletion(branchId, status);
     pushStateStore.clearSessionTracking(branchId);
   }
 
@@ -134,55 +379,21 @@ function handlePullCompletion(
   });
 }
 
-async function handlePrCompletion(sessionId: string, branchId: string, status: SessionStatus) {
+/**
+ * Reconcile PR workflow state with the terminal status event.
+ *
+ * The backend emits `pr-created` before the terminal event, so on a
+ * successful completion the branch is already marked created by the time
+ * this runs; a completed session whose branch is still not created means no
+ * PR URL was found in the output.
+ */
+function handlePrCompletion(branchId: string, status: SessionStatus) {
   if (status === 'completed') {
-    try {
-      // Try session messages first (AI session writes PR_URL: marker).
-      const messages = await commands.getFreshSessionMessages(sessionId);
-      let foundUrl = extractPrUrl(messages);
-
-      // Also check pipeline step outputs for older or partially migrated PR sessions.
-      if (!foundUrl) {
-        const session = await commands.getSession(sessionId);
-        if (session?.pipeline) {
-          for (const step of session.pipeline.steps) {
-            if (step.output) {
-              const match = step.output.match(/https:\/\/github\.com\/[^\s/]+\/[^\s/]+\/pull\/\d+/);
-              if (match) {
-                foundUrl = match[0];
-                break;
-              }
-            }
-          }
-        }
-      }
-
-      if (foundUrl) {
-        const prNumber = extractPrNumber(foundUrl);
-        if (prNumber) {
-          try {
-            await commands.updateBranchPr(branchId, prNumber);
-          } catch (storageError) {
-            console.error('Failed to persist PR state after creation:', storageError);
-            prStateStore.setPrError(branchId, 'Failed to save PR details after creation.');
-            return;
-          }
-
-          try {
-            await commands.refreshPrStatus(branchId);
-          } catch (refreshError) {
-            console.error('Failed to refresh PR state after creation:', refreshError);
-          }
-        }
-        prStateStore.setPrCreated(branchId, foundUrl);
-      } else {
-        prStateStore.setPrError(
-          branchId,
-          'PR session completed but no PR URL was found in the output.'
-        );
-      }
-    } catch (e) {
-      prStateStore.setPrError(branchId, e instanceof Error ? e.message : String(e));
+    if (prStateStore.getPrState(branchId)?.state !== 'created') {
+      prStateStore.setPrError(
+        branchId,
+        'PR session completed but no PR URL was found in the output.'
+      );
     }
   } else {
     prStateStore.setPrError(
@@ -192,29 +403,21 @@ async function handlePrCompletion(sessionId: string, branchId: string, status: S
   }
 }
 
-async function handlePushCompletion(sessionId: string, branchId: string, status: SessionStatus) {
+/**
+ * Reconcile push workflow state with the terminal status event.
+ *
+ * `push-completed` arrives before the terminal event and renders the
+ * classified outcome; a branch still marked `pushing` here means that event
+ * was missed, so fall back to the success rendering (the backend's default
+ * classification when no rejection markers are present).
+ */
+function handlePushCompletion(branchId: string, status: SessionStatus) {
   if (status === 'completed') {
-    try {
-      const session = await commands.getSession(sessionId);
-      const pipeline = session?.pipeline;
-      const messages = await commands.getFreshSessionMessages(sessionId);
-      const outcome = classifyCompletedPushSession(pipeline, messages);
-
-      if (outcome === 'rejected_non_fast_forward') {
-        pushStateStore.setPushError(branchId, '', true);
-      } else {
-        try {
-          await commands.clearBranchPrStatus(branchId);
-        } catch (e) {
-          console.warn('[Staged] Failed to clear PR status after push:', e);
-        }
-        pushStateStore.setPushDone(branchId);
-        setTimeout(() => {
-          pushStateStore.clearPushState(branchId);
-        }, 1_500);
-      }
-    } catch (e) {
-      pushStateStore.setPushError(branchId, e instanceof Error ? e.message : String(e));
+    if (pushStateStore.getPushState(branchId)?.state === 'pushing') {
+      pushStateStore.setPushDone(branchId);
+      setTimeout(() => {
+        pushStateStore.clearPushState(branchId);
+      }, 1_500);
     }
   } else {
     pushStateStore.setPushError(
diff --git a/apps/staged/src/lib/stores/sessionRegistry.svelte.ts b/apps/staged/src/lib/stores/sessionRegistry.svelte.ts
index 24346e1d9..2ff65c405 100644
--- a/apps/staged/src/lib/stores/sessionRegistry.svelte.ts
+++ b/apps/staged/src/lib/stores/sessionRegistry.svelte.ts
@@ -15,6 +15,14 @@
  * - prState: Branch-specific PR workflow state (creating/created/error, PR URL)
  *
  * But they delegate session metadata tracking to this central registry.
+ *
+ * The registry is a pure projection of backend busy state: entries are added
+ * by `session-status-changed` running events, launch-site registrations, and
+ * the `get_active_sessions` snapshot hydration, and removed only on terminal
+ * events or a hydration sweep. There is deliberately no client-side TTL or
+ * size eviction — the backend guarantees terminal events via its session
+ * state machine plus dead-session recovery, and local eviction is precisely
+ * what would turn a missed event into a permanent lie.
  */
 
 import { projectStateStore } from './projectState.svelte';
@@ -29,10 +37,6 @@ interface SessionMetadata {
   timestamp: number; // When the session was registered
 }
 
-const MAX_REGISTRY_SIZE = 200; // Maximum number of sessions to track
-const SESSION_TTL_MS = 48 * 60 * 60 * 1000; // 48 hours - keep longer than prState TTL
-const CLEANUP_THRESHOLD = 0.8; // Run cleanup when registry is 80% full
-
 class SessionRegistry {
   // Map from session ID to session metadata
   private sessions = $state<Map<string, SessionMetadata>>(new Map());
@@ -44,11 +48,6 @@ class SessionRegistry {
    * Register a new session with its metadata
    */
   register(sessionId: string, projectId: string, type: SessionType, branchId?: string): void {
-    // Only cleanup when we're approaching the size limit to avoid O(n) cost on every registration
-    if (this.sessions.size >= MAX_REGISTRY_SIZE * CLEANUP_THRESHOLD) {
-      this.cleanup();
-    }
-
     this.sessions.set(sessionId, {
       sessionId,
       projectId,
@@ -151,28 +150,11 @@ class SessionRegistry {
   }
 
   /**
-   * Clean up stale sessions to prevent memory leaks
-   * Removes sessions older than SESSION_TTL_MS or beyond MAX_REGISTRY_SIZE
+   * Get all tracked session IDs. Used by snapshot hydration to sweep entries
+   * the backend no longer reports as active.
    */
-  private cleanup(): void {
-    const now = Date.now();
-
-    // Remove stale entries (older than TTL)
-    for (const [sessionId, metadata] of this.sessions.entries()) {
-      if (now - metadata.timestamp > SESSION_TTL_MS) {
-        this.sessions.delete(sessionId);
-      }
-    }
-
-    // If still over limit, remove oldest entries
-    if (this.sessions.size > MAX_REGISTRY_SIZE) {
-      const entries = Array.from(this.sessions.entries());
-      entries.sort((a, b) => a[1].timestamp - b[1].timestamp);
-      const toRemove = entries.slice(0, entries.length - MAX_REGISTRY_SIZE);
-      for (const [sessionId] of toRemove) {
-        this.sessions.delete(sessionId);
-      }
-    }
+  getAllSessionIds(): string[] {
+    return Array.from(this.sessions.keys());
   }
 
   /**
diff --git a/apps/staged/src/lib/transport.test.ts b/apps/staged/src/lib/transport.test.ts
index 2c8b75859..fa1cf31f6 100644
--- a/apps/staged/src/lib/transport.test.ts
+++ b/apps/staged/src/lib/transport.test.ts
@@ -48,14 +48,21 @@ class MockWebSocket {
 let sockets: MockWebSocket[];
 
 describe('web transport', () => {
+  let hydrateActiveSessions: ReturnType<typeof vi.fn>;
+
   beforeEach(() => {
     vi.resetModules();
     vi.stubGlobal('crypto', { randomUUID: vi.fn(() => 'web-client-1') });
     sockets = [];
+    // The busy-state hydrator pulls in rune-based stores that plain vitest
+    // can't compile, so it is mocked for every socket-opening test.
+    hydrateActiveSessions = vi.fn().mockResolvedValue(undefined);
+    vi.doMock('./listeners/sessionStatusListener', () => ({ hydrateActiveSessions }));
   });
 
   afterEach(() => {
     vi.doUnmock('./services/prPollingService');
+    vi.doUnmock('./listeners/sessionStatusListener');
     vi.unstubAllGlobals();
     vi.useRealTimers();
   });
@@ -115,7 +122,7 @@ describe('web transport', () => {
     expect(socket.closed).toBe(true);
   });
 
-  it('replays PR polling interest when the browser event socket opens and reconnects', async () => {
+  it('replays PR polling interest and re-hydrates busy state when the browser event socket opens and reconnects', async () => {
     vi.useFakeTimers();
     vi.stubGlobal('WebSocket', MockWebSocket);
     const replayPrPollInterestHints = vi.fn().mockResolvedValue(undefined);
@@ -130,6 +137,7 @@ describe('web transport', () => {
 
     sockets[0].open();
     await vi.waitFor(() => expect(replayPrPollInterestHints).toHaveBeenCalledTimes(1));
+    await vi.waitFor(() => expect(hydrateActiveSessions).toHaveBeenCalledTimes(1));
 
     sockets[0].close();
     await vi.advanceTimersByTimeAsync(2000);
@@ -137,6 +145,7 @@ describe('web transport', () => {
 
     sockets[1].open();
     await vi.waitFor(() => expect(replayPrPollInterestHints).toHaveBeenCalledTimes(2));
+    await vi.waitFor(() => expect(hydrateActiveSessions).toHaveBeenCalledTimes(2));
 
     unlisten();
   });
diff --git a/apps/staged/src/lib/transport.ts b/apps/staged/src/lib/transport.ts
index 0f89441b8..df86ec1d2 100644
--- a/apps/staged/src/lib/transport.ts
+++ b/apps/staged/src/lib/transport.ts
@@ -146,6 +146,19 @@ function replayCurrentPrPollInterestHints(): void {
     });
 }
 
+/**
+ * Re-hydrate busy state from the backend snapshot after a (re)connect: any
+ * `session-status-changed` event emitted while the socket was down is gone
+ * for good, so the accumulated client state may be stale in either direction.
+ */
+function rehydrateBusyState(): void {
+  void import('./listeners/sessionStatusListener')
+    .then(({ hydrateActiveSessions }) => hydrateActiveSessions())
+    .catch((e) => {
+      console.error('[transport] Failed to hydrate busy-state snapshot:', e);
+    });
+}
+
 async function ensureWebSocket(): Promise<void> {
   if (ws && (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING)) {
     return;
@@ -166,6 +179,7 @@ async function ensureWebSocket(): Promise<void> {
     wsConnecting = false;
     startHeartbeat();
     replayCurrentPrPollInterestHints();
+    rehydrateBusyState();
     if (wsReconnectTimer) {
       clearTimeout(wsReconnectTimer);
       wsReconnectTimer = null;
diff --git a/apps/staged/src/lib/types.ts b/apps/staged/src/lib/types.ts
index 49e73b68a..d16d0154a 100644
--- a/apps/staged/src/lib/types.ts
+++ b/apps/staged/src/lib/types.ts
@@ -384,6 +384,12 @@ export interface Session {
   acpConfigSelection?: AcpConfigSelection | null;
   /** Latest session title pushed by the agent via ACP `session_info_update`. */
   acpTitle: string | null;
+  /**
+   * Branch this session runs for. Present on pipeline-launched (pr/push)
+   * sessions, which link no artifact row; artifact-linked sessions resolve
+   * their branch through the artifact instead.
+   */
+  branchId?: string | null;
 }
 
 export type QueuedSessionMessageStatus = 'queued' | 'sending' | 'sent';
@@ -523,6 +529,46 @@ export interface SessionStatusPayload {
   isAutoReview?: boolean;
 }
 
+/**
+ * One entry of the `get_active_sessions` busy-state snapshot: a running or
+ * queued session projected to its branch/project context. Carries the same
+ * discriminators as `SessionStatusPayload` so the snapshot and the
+ * `session-status-changed` delta stream describe sessions identically.
+ */
+export interface ActiveSessionInfo {
+  sessionId: string;
+  projectId: string | null;
+  branchId: string | null;
+  sessionType: string | null;
+  status: SessionStatus;
+  isAutoReview: boolean;
+}
+
+/**
+ * Payload emitted by the `pr-created` domain event when a completed PR
+ * session produced a pull request. The backend has already persisted the PR
+ * number and kicked off a status refresh; clients only render.
+ */
+export interface PrCreatedPayload {
+  branchId: string;
+  sessionId: string;
+  prUrl: string;
+  prNumber: number;
+}
+
+export type PushCompletedOutcome = 'succeeded' | 'rejectedNonFastForward';
+
+/**
+ * Payload emitted by the `push-completed` domain event when a push session
+ * completes. On success the backend has already cleared the stale PR status
+ * (and emitted `pr-status-cleared`); clients only render.
+ */
+export interface PushCompletedPayload {
+  branchId: string;
+  sessionId: string;
+  outcome: PushCompletedOutcome;
+}
+
 // =============================================================================
 // Store status
 // =============================================================================