forge_server/

live.rs

//! What a view has learned from the stream, which its first render and every
//! later one both read.
//!
//! Three of the facts here are announced on the wire once and retained by
//! nobody: whether a completion has been shown, which seats a view is
//! showing, and whether a seat's failed turn has been shown since it failed.
//! A view that attaches to an already-running session cannot reconstruct
//! them, so they are folded here rather than in the view.

use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard};
use std::time::SystemTime;

use forge_primitives::Message;
use forge_primitives::SessionSlot;
use forge_primitives::runtime::RuntimeSessionState;

use crate::SessionUpdate;
use crate::composer::Composer;
use crate::surface::is_success_result;
use crate::translate::state_parsing::parse_runtime_session_state;
use crate::unseen::Unseen;

/// What the view has learned from the stream, which the first render and
/// every later one both read.
#[derive(Default)]
pub struct Live {
    unseen: Unseen,
    /// The seats a page is open on, by how many connections are showing
    /// them: a turn finishing on one of those is a turn the reader watched.
    attached: HashMap<SessionSlot, usize>,
    /// When each seat was last shown, by any connection. A failure newer
    /// than the stamp is one the reader has not seen, which is what the
    /// rail's failure mark reads (#1612): one `Live` is folded per server
    /// and shared by every connection, so showing a seat spends its mark
    /// for all of them.
    seen_failed: HashMap<SessionSlot, SystemTime>,
    composer: Composer,
}

/// What the stream has said, as one render reads it. A render takes this
/// rather than the lock: the guard is not `Send`, and a handler that held
/// it across its own awaits could not be one.
#[derive(Default)]
pub struct LiveState {
    pub unseen: Unseen,
    pub composer: Composer,
    /// The seats a page is open on, by how many connections are showing
    /// them, and when each seat was last shown: together these answer
    /// whether a failure on it is the reader's to see, which the home read
    /// filters rows with.
    attached: HashMap<SessionSlot, usize>,
    seen_failed: HashMap<SessionSlot, SystemTime>,
}

impl LiveState {
    /// The failure instant the rail should draw for `slot`, or `None`
    /// when there is no failure to mark: the seat is being shown (some
    /// connection is watching it fail), or it has been shown since the
    /// failure landed.
    pub fn failed_mark(&self, slot: &SessionSlot, at: SystemTime) -> Option<SystemTime> {
        mark_showing(&self.attached, &self.seen_failed, slot, at).then_some(at)
    }
}

/// Whether a failure at `at` is still the reader's to see.
///
/// One predicate for its two readers - the home's rows, and the attach that
/// spends the mark - so the test the home draws by and the test that decides
/// whether a refresh is owed cannot drift apart.
fn mark_showing(
    attached: &HashMap<SessionSlot, usize>,
    seen_failed: &HashMap<SessionSlot, SystemTime>,
    slot: &SessionSlot,
    at: SystemTime,
) -> bool {
    if attached.contains_key(slot) {
        return false;
    }
    !matches!(seen_failed.get(slot), Some(shown) if shown >= &at)
}

impl Live {
    pub fn new() -> Self {
        Self::default()
    }

    /// A panicking task must not take the view's live state with it.
    pub fn lock(live: &Mutex<Self>) -> MutexGuard<'_, Self> {
        live.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    pub fn snapshot(&self) -> LiveState {
        LiveState {
            unseen: self.unseen.clone(),
            composer: self.composer.clone(),
            attached: self.attached.clone(),
            seen_failed: self.seen_failed.clone(),
        }
    }

    /// This view has shown `slot`, so nothing about it is unseen.
    pub fn seen(&mut self, slot: &SessionSlot) {
        self.unseen.clear(slot);
        self.seen_failed.insert(slot.clone(), SystemTime::now());
    }

    /// A page is open on `slot`, which is a connection showing it, so a mark
    /// armed before the page opened goes with it. Counted, because two tabs
    /// on one seat are one seat still being shown.
    ///
    /// **Answers whether a mark was actually spent** - the diamond this view
    /// was carrying, or a failure at `failed_at` still showing for it. A
    /// connection that holds the home is owed a fresh one exactly then: a
    /// re-encode with nothing to carry is 448 KB and 113-263 ms of server
    /// time for a home the client already has (measured 2026-10-09).
    pub fn attach(&mut self, slot: &SessionSlot, failed_at: Option<SystemTime>) -> bool {
        let spent = self.unseen.is_unseen(slot)
            || failed_at
                .is_some_and(|at| mark_showing(&self.attached, &self.seen_failed, slot, at));
        *self.attached.entry(slot.clone()).or_default() += 1;
        self.unseen.clear(slot);
        self.seen_failed.insert(slot.clone(), SystemTime::now());
        spent
    }

    /// One page on `slot` has gone. The seat is let go with the last of them.
    pub fn detach(&mut self, slot: &SessionSlot) {
        let Some(count) = self.attached.get_mut(slot) else {
            return;
        };
        *count -= 1;
        if *count == 0 {
            self.attached.remove(slot);
            // The seat is no longer shown, and it was shown up to now: a
            // failure that landed while the page was open is not news.
            self.seen_failed.insert(slot.clone(), SystemTime::now());
        }
    }

    /// Whether a page is holding `slot`.
    pub fn is_attached(&self, slot: &SessionSlot) -> bool {
        self.attached.contains_key(slot)
    }

    /// A write sent the seat's words down a path the update stream does not
    /// carry; the composer's own rule decides whether they were its landed
    /// take's (#2283).
    pub fn composer_sent(&mut self, slot: &SessionSlot, words: &str) {
        self.composer.sent(slot, words);
    }

    /// Fold one update in: the marks it moves, and the composer's own record.
    ///
    /// The classification is what keeps a busy turn from mattering here for
    /// every token of it: only the updates that can change what a row draws
    /// move a mark.
    pub fn apply(&mut self, update: &SessionUpdate) {
        self.composer.apply(update);
        match fleet_news(update) {
            // A turn finished on a session this page is not showing, so the
            // row earns its diamond until the session is opened. A seat a
            // page is open on has already shown it, so it earns nothing,
            // and the row settles out of running like any other.
            FleetNews::Completed(key) => {
                if !self.attached.contains_key(key) {
                    self.unseen.mark_completed(key);
                }
            }
            // Work started again, which supersedes the completion the
            // diamond marks, and a fresh occupant whose history is not a
            // completion this page failed to show. Serving the seat's page
            // clears it too, so this is the clear for a seat nobody opened.
            FleetNews::Running(key) | FleetNews::Occupant(key) => {
                self.unseen.clear(key);
            }
            FleetNews::Nothing | FleetNews::Redraw => {}
        }
    }
}

/// What one update asks of the fleet region: the rows, and the marks on them.
///
/// One classification for its two readers - the view folding the stream, and
/// the socket deciding which subscribers an update belongs to - because a
/// second table of the variants would drift from this one.
pub enum FleetNews<'a> {
    /// The region draws nothing of it. Most of the stream: a turn's own words
    /// are the bulk of it, and no row shows one.
    Nothing,
    /// A row or a mark changed.
    Redraw,
    /// A turn finished on the seat.
    Completed(&'a SessionSlot),
    /// Work started again on the seat.
    Running(&'a SessionSlot),
    /// A fresh occupant took the seat.
    Occupant(&'a SessionSlot),
}

impl FleetNews<'_> {
    /// Whether the region draws anything of this update at all.
    pub fn any(&self) -> bool {
        !matches!(self, Self::Nothing)
    }
}

/// Classify one update for the fleet region.
///
/// The filter is what keeps a busy turn from re-sending the fleet for every
/// token of it: only the updates that can change what a row draws belong here.
pub fn fleet_news(update: &SessionUpdate) -> FleetNews<'_> {
    match update {
        // A prompt frame is a user turn: neither arm below draws anything of
        // it, so the origin does not change what the fleet folds.
        SessionUpdate::ChatAppended { key, msg, .. } => match msg {
            Message::Result { is_error, subtype, .. } => {
                if is_success_result(*is_error, subtype) {
                    FleetNews::Completed(key)
                } else {
                    // A failed turn can arm the row's failure mark, so the
                    // rows are re-read; a cancelled turn only redraws the
                    // rows it left.
                    FleetNews::Redraw
                }
            }
            Message::System { subtype, data, .. } if subtype == "session_state_changed" => {
                if parse_runtime_session_state(data.get("state")) == Some(RuntimeSessionState::Running)
                {
                    FleetNews::Running(key)
                } else {
                    FleetNews::Redraw
                }
            }
            Message::BackgroundTasksChanged { .. } => FleetNews::Redraw,
            _ => FleetNews::Nothing,
        },
        // The row set, and what each row is.
        SessionUpdate::Spawning { key, .. }
        | SessionUpdate::Connected { key, .. }
        | SessionUpdate::SessionReplaced { key, .. } => FleetNews::Occupant(key),
        // Everything else that changes what a row or a card says. The
        // catalog, the dictation snapshot and the claude version all arrive
        // after the listener binds: a page opened in those first seconds
        // would otherwise keep the empty answer it painted until the next
        // tick.
        SessionUpdate::CatalogLoaded
        | SessionUpdate::CliVersionChanged
        // The account pool, which the band's own card draws and no row does.
        // A page that read it once drew `0 ready, probing` until the next
        // unrelated redraw.
        | SessionUpdate::AccountsChanged { .. }
        | SessionUpdate::DictateAvailability
        | SessionUpdate::ConnectionFailed { .. }
        | SessionUpdate::TurnError { .. }
        | SessionUpdate::TurnCancelled { .. }
        // The turn's start, and the earliest frame that says a turn was
        // accepted: `turn_pending` is stamped when the prompt is routed and
        // cleared only on the turn's result, so the read this asks for is
        // answered with the row already running. Every state is taken,
        // `state` being free-form by design - a state this build has not
        // seen has to move a row rather than be dropped. `queued` and
        // `started` are the start; `completed` sits at the boundary the
        // result frame also announces; `cancelled`, `discarded` and
        // `refused` are a prompt that will not run, where a row read at the
        // start has no other frame guaranteed to say so. Without this a row
        // kept the idle it last read until an unrelated frame happened to be
        // news (#1887, measured at ~40s while an agent ran underneath).
        | SessionUpdate::PromptLifecycle { .. }
        | SessionUpdate::PermissionRequest { .. }
        | SessionUpdate::QuestionRequest { .. }
        // Answering moves the seat out of the rail's needs-you group and
        // drops the inspector's pending row, so it redraws a row even though
        // it is the composer that asked for it.
        | SessionUpdate::PendingInteractionResolved { .. }
        // A held draft is the third kind of ask: its seat moves into the
        // needs-you group while it waits and back out when it resolves, so a
        // home-only subscriber has to be sent the pair or its row reads as
        // it stood before the draft (#1758). A browser hand-off is the same
        // shape on the same grounds.
        | SessionUpdate::SlackPostPending { .. }
        | SessionUpdate::SlackDraftResolved { .. }
        | SessionUpdate::BrowserHandOffPending { .. }
        | SessionUpdate::BrowserHandOffResolved { .. }
        // A terminal hand-off is the same shape again, and a terminal that
        // opened, exited or was closed moves the row's strip.
        | SessionUpdate::TerminalHandOffPending { .. }
        | SessionUpdate::TerminalHandOffResolved { .. }
        | SessionUpdate::TerminalChanged { .. }
        // **A hand-off has to reach the client wherever the reader is.** The
        // viewer opens on the raise, and the home subscription is the one
        // every client holds - a seat-keyed raise a reader sitting on the
        // home never heard drew no viewer at all (measured 2026-10-12, the
        // e2e). It redraws no row; the arm is what carries it there.
        | SessionUpdate::DocumentHandOffPending { .. }
        | SessionUpdate::DocumentHandOffResolved { .. }
        // A held configuration write is the same shape again: the seat
        // moves into the needs-you group while the dock waits on the
        // reader's answer and back out when it resolves.
        | SessionUpdate::SettingsWritePending { .. }
        | SessionUpdate::SettingsWriteResolved { .. }
        // A held vault call is the same shape again: the seat moves into
        // the needs-you group while its card waits on the reader's answer
        // and back out when it resolves.
        | SessionUpdate::VaultAskPending { .. }
        | SessionUpdate::VaultAskResolved { .. }
        | SessionUpdate::WorkerStatusChanged { .. }
        // The seat's release has begun: its row says where it is going until
        // the roster lands it asleep or gone. Announced at the start of the
        // teardown rather than at the end, because the teardown is what takes
        // the seconds - and a close made from another view reaches this page
        // through this arm alone (#1930).
        | SessionUpdate::Releasing { .. }
        // The project's task set, its schedules and its connector
        // subscriptions moved - the three sections the board's project row
        // draws from this stream. The home hears the task write too, because
        // its fleet counts are derived from the same tasks (#2517).
        | SessionUpdate::TasksChanged { .. }
        | SessionUpdate::CronSchedulesChanged { .. }
        | SessionUpdate::ConnectorSubscriptionsChanged { .. } => FleetNews::Redraw,
        // Everything else is the conversation, which no row shows.
        _ => FleetNews::Nothing,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The pool's frame with an empty pool, which is all these classifications
    /// read of it.
    fn accounts_changed() -> SessionUpdate {
        SessionUpdate::AccountsChanged {
            accounts: forge_workspace::AccountPoolView {
                loading: Vec::new(),
                all_loaded: false,
                gateway: forge_workspace::AccountGatewayView {
                    ready: false,
                    port: 0,
                    bind_error: None,
                },
                usage: Vec::new(),
            },
        }
    }

    fn result_message(subtype: &str, is_error: bool) -> Message {
        serde_json::from_value(serde_json::json!({
            "type": "result",
            "subtype": subtype,
            "duration_ms": 1,
            "duration_api_ms": 1,
            "is_error": is_error,
            "num_turns": 1,
            "session_id": "s",
        }))
        .expect("parse a result message")
    }

    fn appended(key: &SessionSlot, msg: Message) -> SessionUpdate {
        SessionUpdate::ChatAppended { key: key.clone(), msg, source: None }
    }

    /// The failure mark is the view's own: a failure on a seat this page
    /// has not shown marks; showing the seat clears it; a failure after
    /// that marks again; and one that lands while the page is showing the
    /// seat marks nothing, because the reader is watching it happen.
    #[test]
    fn a_failure_marks_until_the_view_shows_the_seat() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        let first = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(100);

        assert_eq!(
            live.snapshot().failed_mark(&slot, first),
            Some(first),
            "a failure nobody has shown marks",
        );

        live.attach(&slot, None);
        assert_eq!(
            live.snapshot().failed_mark(&slot, first),
            None,
            "showing the seat clears the mark",
        );
        live.detach(&slot);
        assert_eq!(
            live.snapshot().failed_mark(&slot, first),
            None,
            "and it stays clear once the page leaves - the reader saw it",
        );

        // The two stamps are taken from the clock, so each instant below is
        // given a gap to sit strictly after the one before it: on a coarse
        // clock, `now()` twice in a row reads equal.
        std::thread::sleep(std::time::Duration::from_millis(2));
        let second = std::time::SystemTime::now();
        assert_eq!(
            live.snapshot().failed_mark(&slot, second),
            Some(second),
            "a failure after the showing is news again",
        );

        live.attach(&slot, None);
        std::thread::sleep(std::time::Duration::from_millis(2));
        let third = std::time::SystemTime::now();
        assert_eq!(
            live.snapshot().failed_mark(&slot, third),
            None,
            "a failure landing while the page is open marks nothing",
        );
        live.detach(&slot);
        assert_eq!(
            live.snapshot().failed_mark(&slot, third),
            None,
            "and leaving does not resurrect it",
        );
    }

    /// **An attach answers whether it spent anything**, which is what a home
    /// refresh after it is owed for: the diamond it was carrying, or a failure
    /// mark still showing. Nothing spent means the home the client holds is
    /// the home it would be sent - and re-encoding it anyway is 448 KB and
    /// 113-263 ms of server time for no change (measured 2026-10-09).
    #[test]
    fn an_attach_answers_whether_a_mark_was_spent() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();

        let first = live.attach(&slot, None);
        assert!(!first, "a seat carrying no marks spends nothing");
        let second = live.attach(&slot, None);
        assert!(!second, "and a second page on it spends nothing either");
        live.detach(&slot);
        live.detach(&slot);

        // The diamond: a completion on a seat nobody is showing arms it.
        live.apply(&appended(&slot, result_message("success", false)));
        assert!(
            live.attach(&slot, None),
            "a completion nobody has shown is spent by showing the seat",
        );

        // A failure mark: a failure instant after the last time the seat was
        // shown. The stamps come from the clock, so each instant is given a
        // gap to sit strictly after the one before it.
        live.detach(&slot);
        std::thread::sleep(std::time::Duration::from_millis(2));
        let failed = std::time::SystemTime::now();
        assert!(
            live.attach(&slot, Some(failed)),
            "a failure nobody has shown is spent by showing the seat",
        );
        live.detach(&slot);
        let again = live.attach(&slot, Some(failed));
        assert!(!again, "a failure already shown is not spent twice");
    }

    fn session_state(state: &str) -> Message {
        serde_json::from_value(serde_json::json!({
            "type": "system",
            "subtype": "session_state_changed",
            "session_id": "s",
            "state": state,
        }))
        .expect("parse a state message")
    }

    /// A diamond goes when its seat moves on without being looked at: work
    /// started again, or a fresh occupant took the slot. The other clear,
    /// the seat's page being served, is pinned through the route.
    #[test]
    fn starting_work_again_clears_the_diamond() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        live.apply(&appended(&slot, result_message("success", false)));
        assert!(live.snapshot().unseen.is_unseen(&slot), "precondition: the turn armed it");

        let starting = appended(&slot, session_state("running"));
        assert!(fleet_news(&starting).any(), "a turn starting redraws the page");
        live.apply(&starting);
        assert!(
            !live.snapshot().unseen.is_unseen(&slot),
            "and clears a completion nobody looked at",
        );

        // The same for a slot a new occupant took: its history is not a
        // completion this page failed to show.
        let taken = SessionSlot::lead("Org", "other");
        live.apply(&appended(&taken, result_message("success", false)));
        assert!(live.snapshot().unseen.is_unseen(&taken), "precondition: armed");

        live.apply(&SessionUpdate::Connected {
            key: taken.clone(),
            session_id: forge_primitives::SessionId::new("new"),
            cwd: "/proj".to_owned(),
            current_model: forge_primitives::CurrentModel {
                resolved_id: "claude".to_owned(),
                display_name_short: "claude".to_owned(),
                display_name_long: "claude".to_owned(),
                requested_id: None,
                catalog_id: None,
                supports_effort: false,
                supported_effort_levels: Vec::new(),
                supports_auto_mode: None,
                supports_adaptive_thinking: None,
                is_authoritative: true,
            },
            available_models: Vec::new(),
            mode: None,
            history: Vec::new(),
            dropped: 0,
            compaction_count: 0,
        });

        assert!(
            !live.snapshot().unseen.is_unseen(&taken),
            "a replaced occupant starts from a clean row",
        );

        // And clearing one slot leaves the rest alone, which is the
        // property the fold has to keep: a new turn somewhere is not news
        // about somewhere else.
        let untouched = SessionSlot::lead("Org", "untouched");
        live.apply(&appended(&slot, result_message("success", false)));
        live.apply(&appended(&untouched, result_message("success", false)));
        live.apply(&appended(&slot, session_state("running")));
        let unseen = live.snapshot().unseen;
        assert!(!unseen.is_unseen(&slot), "the slot that started again is cleared");
        assert!(unseen.is_unseen(&untouched), "and the slot that did not keeps its diamond");
    }

    /// Catches a diamond armed by the wrong result, and a failed turn that
    /// redraws the rows without arming one - the mark it arms is the
    /// failure mark, not the diamond.
    #[test]
    fn only_a_finished_turn_arms_the_diamond() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();

        let failed = appended(&slot, result_message("error_during_execution", true));
        assert!(fleet_news(&failed).any(), "a failed turn redraws the rows the failure mark rides");
        live.apply(&failed);
        assert!(
            !live.snapshot().unseen.is_unseen(&slot),
            "but a turn that failed is not a turn that finished, so no diamond",
        );

        let finished = appended(&slot, result_message("success", false));
        assert!(fleet_news(&finished).any(), "a finished turn redraws the page");
        live.apply(&finished);
        assert!(live.snapshot().unseen.is_unseen(&slot), "and leaves the diamond");

        let other = SessionSlot::lead("Org", "other");
        let elsewhere = appended(&other, result_message("success", false));
        assert!(fleet_news(&elsewhere).any(), "a second slot's finish is the same kind of event");
        live.apply(&elsewhere);
        let unseen = live.snapshot().unseen;
        assert!(unseen.is_unseen(&other), "and earns its own diamond");
        assert!(unseen.is_unseen(&slot), "without clearing the first slot's");
    }

    /// A turn that finishes on a seat whose page is open is a turn the reader
    /// watched, so it earns no diamond - and the row still redraws, because
    /// the turn that was running has ended.
    #[test]
    fn an_attached_seat_earns_no_diamond_and_still_settles() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        live.attach(&slot, None);

        let finished = appended(&slot, result_message("success", false));
        assert!(fleet_news(&finished).any(), "the row settles out of running, so the page redraws");
        live.apply(&finished);
        assert!(
            !live.snapshot().unseen.is_unseen(&slot),
            "and the page that is open on it has shown the turn",
        );
    }

    /// The other half: with the page gone, a completion is unlooked again.
    #[test]
    fn a_seat_arms_again_once_its_page_closes() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        live.attach(&slot, None);
        live.detach(&slot);

        let finished = appended(&slot, result_message("success", false));
        assert!(fleet_news(&finished).any(), "the diamond is the home's news again");
        live.apply(&finished);
        assert!(
            live.snapshot().unseen.is_unseen(&slot),
            "so a seat nobody is showing earns its diamond",
        );
    }

    /// Two tabs on one seat are two connections and one seat still being
    /// shown. Catches holding the attachment as a flag, where closing either
    /// tab re-arms a mark the other tab is still displaying.
    #[test]
    fn two_pages_on_one_seat_hold_it_until_both_close() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        live.attach(&slot, None);
        live.attach(&slot, None);
        live.detach(&slot);

        live.apply(&appended(&slot, result_message("success", false)));
        assert!(
            !live.snapshot().unseen.is_unseen(&slot),
            "the tab still open on it has shown the turn",
        );

        live.detach(&slot);
        live.apply(&appended(&slot, result_message("success", false)));
        assert!(
            live.snapshot().unseen.is_unseen(&slot),
            "and with both gone the seat is unlooked again",
        );
    }

    /// The window between the page being served and its stream attaching: a
    /// completion landing in it would otherwise sit on a page that is open.
    #[test]
    fn attaching_clears_a_mark_that_armed_before_it() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        live.apply(&appended(&slot, result_message("success", false)));
        assert!(live.snapshot().unseen.is_unseen(&slot), "precondition: the turn armed it");

        live.attach(&slot, None);
        assert!(
            !live.snapshot().unseen.is_unseen(&slot),
            "the page opening is the reader being shown the seat",
        );
    }

    /// A seat nobody is showing leaves the map rather than sitting in it at
    /// zero, which on a process up for a week is a seat-shaped leak.
    #[test]
    fn a_closed_page_leaves_no_entry_behind() {
        let slot = SessionSlot::lead("Org", "forge");
        let mut live = Live::new();
        live.attach(&slot, None);
        assert_eq!(live.attached.len(), 1, "precondition: the seat is held");

        live.detach(&slot);
        assert!(
            live.attached.is_empty(),
            "the last page closing takes the seat out of the map: {:?}",
            live.attached,
        );
    }

    /// **A turn's start is the home's news, and without the arm the row kept
    /// the idle it last read** (#1887, measured at ~40s while an agent ran
    /// underneath). The read is what turns the row running: the roster folds
    /// `turn_pending`, which is stamped when the prompt is routed rather than
    /// when a frame lands, so the state is already there for the asking.
    #[test]
    fn a_prompts_lifecycle_redraws_the_page() {
        let slot = SessionSlot::lead("Org", "forge");

        for state in ["queued", "started", "completed", "cancelled", "discarded", "refused"] {
            let update = SessionUpdate::PromptLifecycle {
                key: slot.clone(),
                uuid: "p1".to_owned(),
                state: state.to_owned(),
            };
            assert!(fleet_news(&update).any(), "a prompt's {state} is what a row is re-read for");
        }

        // The free-form state the protocol promises to carry rather than
        // drop: a build that has not seen it still moves the row.
        let unseen = SessionUpdate::PromptLifecycle {
            key: slot,
            uuid: "p1".to_owned(),
            state: "a-state-from-a-newer-cli".to_owned(),
        };
        assert!(fleet_news(&unseen).any(), "an unseen state is news too");
    }

    /// The registry that moves `has_background_work`, which is the promotion
    /// that draws an idle-but-busy seat as running. The CLI announces it as a
    /// conversation frame, and that frame is what the home re-reads for; the
    /// `BackgroundTasksChanged` mirror of the same instant carries the
    /// inspector's set and is not the fleet's.
    #[test]
    fn the_clis_background_announcement_redraws_the_page() {
        let slot = SessionSlot::lead("Org", "forge");

        // The wire's own shape: the typed decode needs the frame's `uuid`,
        // and a fixture without one falls back to the generic system variant,
        // which no arm here answers.
        let announced = serde_json::from_value(serde_json::json!({
            "type": "system",
            "subtype": "background_tasks_changed",
            "tasks": [{ "task_id": "t1", "task_type": "local_bash", "description": "a wait" }],
            "uuid": "u1",
            "session_id": "s",
        }))
        .expect("parse a background announcement");

        assert!(
            fleet_news(&appended(&slot, announced)).any(),
            "the registry moving is a row's re-read",
        );
    }

    /// A release that has begun moves a row, wherever the close came from.
    /// Its own frame is the only one a close made from another view sends
    /// before the removal lands (#1930), so a home subscriber that did not
    /// re-read here kept the row reading as working through the teardown.
    #[test]
    fn a_release_that_has_begun_redraws_the_page() {
        let slot = SessionSlot::lead("Org", "forge");

        assert!(
            fleet_news(&SessionUpdate::Releasing { key: slot }).any(),
            "the seat going to sleep is a row's re-read",
        );
    }

    /// The bulk of the stream is the conversation, which the home does not
    /// draw; re-sending it for one would rebuild the fleet per token.
    #[test]
    fn a_chat_message_does_not_redraw_the_page() {
        let slot = SessionSlot::lead("Org", "forge");

        assert!(
            !fleet_news(&SessionUpdate::ChatAppended {
                key: slot,
                source: None,
                msg: serde_json::from_value(serde_json::json!({
                    "type": "user",
                    "message": { "role": "user", "content": "hello" },
                    "session_id": "s",
                }))
                .expect("parse a user message"),
            })
            .any(),
            "a chat message is not something the home draws",
        );
    }

    /// A take's readings are the composer's news, not the fleet's. The
    /// region this stream re-sends draws rows, so redrawing it for every
    /// level would rebuild the whole page twenty times a second while a
    /// take runs.
    #[test]
    fn a_takes_readings_are_not_the_fleets_news() {
        let slot = SessionSlot::lead("Org", "forge");

        for update in [
            SessionUpdate::DictateStarted {
                key: slot.clone(),
                floor_db: -50.0,
                generation: 1,
                initiator: None,
            },
            SessionUpdate::DictateLevel {
                key: slot.clone(),
                peak_db: -20.0,
                generation: 1,
                initiator: None,
            },
        ] {
            assert!(!fleet_news(&update).any(), "{update:?} is not a row changing");
        }
    }

    /// The updates that land after the listener binds, and that a page
    /// opened in that window has already painted an answer for: the
    /// catalog scan, the dictation snapshot, the claude version probe and
    /// the account pool settling. Catches a page that keeps the empty
    /// answer until the next tick.
    ///
    /// The pool is the one that bites hardest, because nothing else in the
    /// stream mentions it: a quiet forge emits no other update, so a card
    /// that read `0 ready, probing` at subscribe reads it for the life of
    /// the connection.
    #[test]
    fn the_late_boot_updates_redraw_the_page() {
        for update in [
            SessionUpdate::CatalogLoaded,
            SessionUpdate::DictateAvailability,
            SessionUpdate::CliVersionChanged,
            accounts_changed(),
        ] {
            assert!(fleet_news(&update).any(), "{update:?} is exactly a render wake-up");
        }
    }
}