forge_server/transport/

envelope.rs

//! What a client and this server exchange.
//!
//! Both enums are tagged on `kind`, which is added beside a message's own
//! fields rather than replacing anything, so a reader can tell what a
//! message is without knowing its shape. **The tag is the contract**:
//! adding a variant is compatible and renaming one is not.

use forge_primitives::document::DocumentKind;
use forge_primitives::git_diff::LayerState;
use forge_primitives::{ClientConfig, SessionSlot};
use forge_workspace::DictateAxes;
use forge_workspace::env::git_diff::content::FileContent;
use serde::{Deserialize, Serialize};

use super::wire::ReviewedThreadWire;
use crate::live::fleet_news;
use crate::{Command, SessionUpdate};

/// What a client can watch. A subscription names one of these and is
/// answered with that subject's snapshot.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Subject {
    Home,
    /// Every project's board detail: its rows, its schedules and its
    /// connector subscriptions.
    ///
    /// A subject of its own rather than fields of [`Subject::Home`] because
    /// the board is its own page's data: the home is a glance at the fleet,
    /// and carrying every project's rows made a home read about 708 KB, most
    /// of it board data a home page draws none of (#2517).
    Board,
    Session(SessionSlot),
    /// The token/cost pool, scanned on the ask.
    ///
    /// It belongs to no seat - the report is the whole session-JSONL pool,
    /// bucketed by project and model - and no update carries it: the pool
    /// changes as transcripts are written, and nothing in the stream says a
    /// transcript's tokens moved. So a subscription is answered with one scan
    /// and hears nothing after it, and a client asks again by subscribing
    /// again.
    Usage,
    /// The models page's read: the pinned models, the catalogue check,
    /// the feed's rows and the updates it proposes.
    ///
    /// Carried as a subject of its own rather than inside `Home` because
    /// the rows are the whole catalogue and only the models page draws
    /// them. Unlike `Usage` it does hear updates: a check landing, boot's
    /// or a page's own, arrives as `SessionUpdate::DictateModelsChanged`.
    DictateModels,
    /// The Extensions page's read: the plugin inventory, the marketplace
    /// health and the last update run.
    ///
    /// A subject of its own for `DictateModels`' reason - only the
    /// Extensions page draws any of it - and it hears the whole `Plugins*`
    /// family: an action's outcome, a run's progress, and the refreshed
    /// inventory are all news about what this subscription is showing.
    Extensions,
}

impl Subject {
    /// Whether an update belongs to what this subscription asked for.
    ///
    /// A session subscription is a seat, so the workspace is ASKED which seat
    /// a variant routes to rather than this matching the variants itself.
    ///
    /// A home subscription is a row per seat plus the App-level facts, and it
    /// takes the classification that already exists for the row half:
    /// [`fleet_news`], which is what the view folding this stream draws the
    /// fleet from. It is narrow on purpose - a row states a seat's lifecycle,
    /// its status and its pending state, and no row shows a word of its
    /// conversation - so a second table of variants here would both drift and
    /// hand a home subscriber every token of every seat in the fleet.
    ///
    /// **The fleet region is not the whole home.** Two home fields carry no
    /// slot and are not fleet news: the service status and the fatal error.
    /// They come with the slot-less arm below, and without it a client would
    /// draw what it read once at subscribe for the life of the connection.
    /// The account pool is a home field too, but its announcement is also a
    /// fleet redraw, so it arrives through the first arm.
    ///
    /// **That arm is wider than the client's read rule, by the plugin
    /// family.** The plugin records cross to a home subscriber only because
    /// they name no seat; no field of the home snapshot carries them, their
    /// reader is the `extensions` subject, and `coversHome`
    /// (`client/src/wire/fleet.ts`) declines the home read for them on
    /// purpose. Delivery here is the superset.
    pub fn covers(&self, update: &SessionUpdate) -> bool {
        match self {
            // **The board's own variants are not the home's.** A cron or a
            // connector set left the home's payload with the rows, so a home
            // subscriber that heard one would be asked to re-read for news its
            // snapshot cannot carry (rule 25). `TasksChanged` stays: the
            // fleet's counts are derived from the same tasks, so the home's
            // answer does carry its effect.
            Self::Home => {
                (fleet_news(update).any() || update.slot().is_none()) && !board_detail(update)
            }
            // The board snapshot carries every project's rows, crons and
            // connector sets, so the variants that move them - plus the seat
            // events a row's `stale` mark is read from - belong here. A row's
            // time-derived marks are the page's own beat, not a frame.
            Self::Board => {
                board_detail(update)
                    || matches!(update, SessionUpdate::TasksChanged { .. })
                    || seat_liveness(update)
            }
            Self::Session(seat) => update.slot() == Some(seat),
            // Nothing carries the pool: its snapshot is a scan, and a
            // transcript's tokens moving is not an update any variant
            // announces.
            Self::Usage => false,
            // The check's own landing, and the load's signal: a page open
            // during a first 3 GB download hears models finished loading
            // through `DictateAvailability` and reads again, so its state
            // chips are not stuck on `pending` until the next check. The
            // feed belongs to no seat, so nothing else reaches here.
            Self::DictateModels => matches!(
                update,
                SessionUpdate::DictateModelsChanged { .. } | SessionUpdate::DictateAvailability
            ),
            // The whole `Plugins*` family: every one is news about what
            // this subscription draws - an action's outcome, a run's rows
            // moving, the refreshed inventory, a failed refresh, a
            // rollback's result. The family belongs to no seat, so nothing
            // else routes here.
            Self::Extensions => matches!(
                update,
                SessionUpdate::PluginsInventoryUpdated { .. }
                    | SessionUpdate::PluginsInventoryRefreshFailed { .. }
                    | SessionUpdate::PluginsCliActionSucceeded { .. }
                    | SessionUpdate::PluginsCliActionFailed { .. }
                    | SessionUpdate::PluginsUpdateRunProgress { .. }
                    | SessionUpdate::PluginsUpdateRunFinished { .. }
                    | SessionUpdate::PluginsRollbackSucceeded { .. }
                    | SessionUpdate::PluginsRollbackFailed { .. }
            ),
        }
    }
}

/// The updates whose whole effect is the board's own snapshot.
///
/// Crons and connector sets moved out of the home's payload with the rows
/// (#2517), so the home must not hear them; the board carries both.
fn board_detail(update: &SessionUpdate) -> bool {
    matches!(
        update,
        SessionUpdate::CronSchedulesChanged { .. }
            | SessionUpdate::ConnectorSubscriptionsChanged { .. }
    )
}

/// The seat events a board row's `stale` mark is read from.
///
/// `seat_is_live` reads the domain/agent registry and the worker registry,
/// and these are the frames that change either: a seat starting, connecting,
/// being replaced, releasing, failing to connect, or a worker's status
/// moving. Turn frames are not here - a task write is what moves a row's work,
/// and a page's own beat advances its clocks.
///
/// **The coverage is not exact.** A session task exiting and a worktree
/// rollback release the handle with no frame of their own, and `Spawning`
/// fires before the handle exists - so in those windows a row's mark can lag
/// until the seat's next covered frame, or until the page's beat re-reads.
fn seat_liveness(update: &SessionUpdate) -> bool {
    matches!(
        update,
        SessionUpdate::Spawning { .. }
            | SessionUpdate::Connected { .. }
            | SessionUpdate::SessionReplaced { .. }
            | SessionUpdate::Releasing { .. }
            | SessionUpdate::ConnectionFailed { .. }
            | SessionUpdate::WorkerStatusChanged { .. }
    )
}

/// What a client sends.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ClientMessage {
    /// A liveness probe, sent by a transport after a quiet spell and by the
    /// page on a return from the background.
    ///
    /// **Application-level rather than a WebSocket control frame**: a browser
    /// page cannot send a Ping, so control frames would leave the web build
    /// with no probe at all. Any inbound frame answers one; this exists so a
    /// quiet peer is always asked something.
    Ping,
    Subscribe {
        what: Subject,
        /// Whether this client can answer the prompts it is shown.
        ///
        /// Off unless the client says otherwise, because the core parks a
        /// turn on the reply of whoever registered as answering: a client
        /// counted as able to answer a prompt it has no surface to draw hangs
        /// the turn rather than failing it. A client with a dock to answer
        /// from says so here; the connection registers with the core
        /// accordingly before it forwards anything.
        ///
        /// **It is the CONNECTION's role, not the subject's.** The core keeps
        /// one role per connection and it only ever rises, so a client that
        /// can answer states it on the first subscribe it makes - usually the
        /// home - and not only while a seat it could draw is showing.
        ///
        /// **What it guards is an install with no answerer anywhere.** Every
        /// update is delivered to every subscriber regardless of role; the
        /// role decides only what the core may PARK on. A forge with an
        /// answering client attached already has an answerer, so a
        /// second client that says nothing is a guest beside one that can.
        /// Where nobody can answer, an ask is cancelled at birth rather than
        /// held for a reply that could not come. Declaring this means that
        /// case never arises while this client is attached.
        #[serde(default)]
        answering: bool,
        /// The review page's own read: refresh the branch diff's base before
        /// the record is encoded, and pin the stats and content reads to the
        /// sha that refresh resolved.
        ///
        /// **Off for every other read.** The ordinary record read is the one
        /// behind every seat page and must not wait on the network; a page
        /// that draws a diff asks again with this set when it opens, so the
        /// state beside that diff belongs to the moment it was read.
        #[serde(default)]
        review: bool,
    },
    Unsubscribe {
        what: Subject,
    },
    /// Which seat this connection has on screen, `None` when it shows none.
    ///
    /// **A state assertion, not an event**: the last one wins and repeating it
    /// is a no-op. A showing spends a seat's diamond, its failure mark and its
    /// death trace - once the seat is also subscribed, so a client that keeps
    /// every visited seat subscribed spends nothing until the reader opens
    /// one. Nothing answers it.
    Showing {
        seat: Option<SessionSlot>,
    },
    /// `reply_to` is set only when the client wants an answer - which is the
    /// eight commands whose reply rides a oneshot, and nothing else.
    ///
    /// Boxed because a command is 360 bytes and every other variant here is
    /// a fraction of that (clippy::large_enum_variant). Serde writes a box
    /// as the value inside it, so the wire form is unchanged.
    Command {
        command: Box<Command>,
        reply_to: Option<u64>,
    },
    More {
        conversation: SessionSlot,
        before: Option<String>,
        turns: u32,
    },
    /// The frames a seat's stream emitted between two numbers, for a client
    /// that read a gap in them.
    ///
    /// **The numbers are the seat's own**, minted as each frame crossed, so a
    /// repair folds into the client's stream exactly where the frames fell
    /// rather than reading as a second conversation. `epoch` is the occupant
    /// those numbers belong to: a seat whose occupant has changed has started
    /// counting again, and a repair across the two is refused rather than
    /// answered.
    SeatRange {
        slot: SessionSlot,
        epoch: Option<forge_primitives::SessionId>,
        from: u64,
        to: u64,
    },
    /// One commit's own diff, for the review canvas' commit scopes.
    ///
    /// Asked on demand rather than carried on the record: a branch can
    /// hold dozens of commits, and every reader of a record would pay
    /// their content on every read for a scope only one page draws.
    CommitDiff {
        session: SessionSlot,
        sha: String,
    },
    /// Attach this connection to one of a seat's browser tabs, as a viewer.
    ///
    /// `tab` is the tab to drive; `None` is the asking seat's current tab,
    /// else its profile's first. The answer is a `browser_view` naming the
    /// profile's wire id and the tab, then a `browser_nav_state`, then frames
    /// as they paint. Attaching again replaces what this connection was
    /// viewing.
    BrowserView {
        seat: SessionSlot,
        tab: Option<u16>,
        /// Answers the attach: `{"ok": true, "profile": <the one attached>,
        /// "tab": <the tab>}`, or `{"ok": false, "why": ...}` with the
        /// server's own words. **A raise that cannot be answered is a dark
        /// surface**, so a client that can wait sends one; the profile and tab
        /// are what the page's Done lowers.
        reply_to: Option<u64>,
    },
    /// Open the page a parked hand-off named: the same attach, addressed by
    /// the page's CDP target instead of a tab number.
    ///
    /// **A number is not an identity across a host restart.** The successor
    /// host's first tab takes number 1 again, so an ask parked through the
    /// replacement must name its page; a page that is gone is refused with
    /// the tabs that are open now.
    BrowserViewPage {
        seat: SessionSlot,
        target: String,
        /// The same answer `BrowserView` gives: the profile and tab attached,
        /// or the server's refusal.
        reply_to: Option<u64>,
    },
    /// The files a person chose for a page's dialog, or none for a dismiss.
    ///
    /// **The bytes cross base64** because the wire is JSON; the server
    /// writes them to its own temp directory and hands the host the paths,
    /// because the host reads a file's content at the moment it is answered.
    BrowserFileAnswer {
        id: u64,
        files: Vec<BrowserFileWire>,
    },
    /// Stop viewing the attached profile. **Lowering the takeover is a
    /// detach**: the idle clock counts an attached viewer as a use, and a
    /// connection that switched profiles must not leave a phantom viewer on
    /// the old one.
    BrowserDetach,
    /// Attach this connection to one of a seat's terminals, as a viewer. The
    /// answer is a `terminal_view` carrying the screen as a VT stream, then
    /// output frames as the shell writes them.
    TerminalView {
        seat: SessionSlot,
        /// The terminal id, or `None` to open the seat's terminal - creating
        /// it when the seat has none - and attach to it. **The person's door
        /// on a seat every agent has left alone.**
        #[serde(default)]
        id: Option<u16>,
        /// **The grid the client will draw on**, sent with the attach so the
        /// repaint is laid out for it: a client that attached first and
        /// resized after had to draw nothing until a second repaint arrived,
        /// which read as the screen flushing.
        #[serde(default)]
        cols: Option<u16>,
        #[serde(default)]
        rows: Option<u16>,
        /// Answers the attach: `{"ok": true, "id": <the one attached>}` or
        /// `{"ok": false, "why": ...}`. A raise that cannot be answered is a
        /// dark surface, so a client that can wait sends one.
        reply_to: Option<u64>,
    },
    /// Stop viewing the attached terminal: the idle clock and the resize
    /// arbitration both read the truth from it.
    TerminalDetach,
    /// The attached surface changed size: the PTY and the model resize.
    TerminalResize {
        seat: SessionSlot,
        id: u16,
        cols: u16,
        rows: u16,
    },
    /// Client-driven flow control, so a phone that backgrounded stops being
    /// written to rather than being dropped. A resume is answered with a
    /// fresh `terminal_view`, because the frames it missed are not replayed.
    TerminalPause {
        paused: bool,
    },
    /// Take the person's control of an attached terminal, or give it back.
    /// **One controller at a time, like the browser's role**: a take always
    /// lands, and the previous holder hears it lost the role.
    TerminalControl {
        seat: SessionSlot,
        id: u16,
        take: bool,
    },
    /// The person's close door: end the terminal's process group.
    TerminalClose {
        seat: SessionSlot,
        id: u16,
    },
    /// One input event from the controlling viewer.
    ///
    /// `seq` is the viewer's own ordering: the server drops an event at or
    /// below the last one the role carried, so a superseded gesture never
    /// lands late. The event is the gesture, not CDP - the server maps it.
    BrowserInput {
        seq: u64,
        event: forge_primitives::browser::BrowserInputEvent,
    },
    /// Take the controller role, or give it back.
    BrowserControl {
        take: bool,
    },
    /// Move through the page's history: the takeover bar's arrows.
    BrowserNav {
        back: bool,
    },
    /// The controller's surface changed size: the page is laid out to it.
    ///
    /// `width` and `height` are the surface's DEVICE pixels - the client
    /// multiplies its CSS box by the ratio it draws at - and `scale` is that
    /// ratio, which the host gives CEF.
    BrowserResize {
        width: u32,
        height: u32,
        scale: f64,
    },
    /// The tabs of this seat's profile, for a client's strip. Answered
    /// through `reply_to` with `{"ok": true, "tabs": {
    /// "profile", "id", "up", "connected", "closing_at_ms", "rows" }}`, and
    /// then pushed as a `browser_tabs` whenever they change.
    BrowserTabs {
        seat: SessionSlot,
        reply_to: Option<u64>,
    },
    /// One document, read out of a seat's own tree for a viewer's route.
    ///
    /// `path` resolves against the seat's working directory, and `reviewable`
    /// is what the hand-off asked for - the hand-off's own fact, refused on a
    /// PDF because that kind can never carry a review. Whether the DOCUMENT
    /// is reviewable is the read's own answer, not this flag's echo. The
    /// answer is a `document_read` message or an `error` naming it; those are
    /// its only two.
    DocumentRead {
        session: SessionSlot,
        path: String,
        reviewable: bool,
    },
    /// One file's whole content at a revision, for the review page's
    /// between-hunk expansion.
    ///
    /// The record's diff carries bounded hunks, so the lines BETWEEN them
    /// are exactly what it does not carry - and only a page whose reader
    /// opened a gap needs them, which is why this is asked for on demand
    /// rather than carried on every record. `sha` names the scope's own
    /// revision: a commit's tree for a commit scope, and `None` for the
    /// seat's branch tip, which is the new side of the all-changes diff.
    /// The answer is a `file_content` message; a read that could not run
    /// answers as itself rather than as an error.
    FileContent {
        session: SessionSlot,
        path: String,
        sha: Option<String>,
    },
    /// One page's own perf records, for a client with no shell to write them.
    ///
    /// **The records stay JSON, unparsed here**: they are the page's instrument
    /// (`client/src/perf.ts`), and the server's part is the file - written
    /// beside `forge.log` so a web client produces diagnostics like the shells
    /// do. Nothing answers it; a batch that could not be written is a debug
    /// record, not the client's failure.
    ClientPerf {
        records: Vec<serde_json::Value>,
    },
    /// A batched save of the effective config. A save that writes is
    /// answered with [`ServerMessage::SettingsSaved`]; one that does not is
    /// answered with [`ServerMessage::SettingsRefused`]. A refused save
    /// writes nothing, so no push follows it.
    SettingsEdit {
        /// The fingerprint of the raw file text the page loaded.
        fingerprint: String,
        edits: Vec<forge_primitives::settings::SettingsEditRequest>,
        /// The page's own number for this save, echoed by
        /// [`ServerMessage::SettingsSaved`] and
        /// [`ServerMessage::SettingsRefused`] so a reply is matched to the
        /// request that made it. Required: a client that forgets it is
        /// refused rather than silently answered as id 0.
        id: u64,
    },
}

/// Why a settings save was refused.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SettingsRefusal {
    /// The file moved since the page loaded it.
    Stale,
    /// The candidate failed the boot loader.
    Invalid,
    /// A guard refused the save.
    Refused,
    /// The file could not be read or written. Nothing was applied: a write
    /// that lands and then fails its directory fsync is reported as written.
    Io,
}

/// One chosen file on the wire: its name and its bytes, base64.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BrowserFileWire {
    pub name: String,
    pub data: String,
}

/// What the server sends.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ServerMessage {
    /// The answer to [`ClientMessage::Ping`], and proof the peer is reading.
    Pong,
    /// The effective config, pushed once after the greeting and again after
    /// every applied reload. Server-wide rather than seat-scoped: every
    /// connection draws the same document, and a connection that attaches
    /// while a reload is in flight reads the current load here - which is
    /// why no settings read verb exists.
    Settings {
        document: forge_primitives::SettingsDocument,
    },
    /// A settings save that wrote nothing. `message` is the loader's own
    /// words for an invalid candidate or the guard's own for a refusal;
    /// `key_path` anchors it at one of the save's keys where one is known.
    SettingsRefused {
        reason: SettingsRefusal,
        message: String,
        #[serde(default)]
        id: u64,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        key_path: Option<String>,
    },
    /// A settings save that wrote. The reload push carries the new document;
    /// this is the saver's own acknowledgement, matched by `id`, so a page
    /// clears its draft only for its own save.
    SettingsSaved {
        id: u64,
    },
    Greeting {
        version: u32,
        /// The build that sent this, in the two forms a view's own
        /// header draws. The greeting is the only thing a client is
        /// guaranteed to have before it reads or refuses anything, so a skew
        /// can name both halves from here and nowhere else.
        forge_version: String,
        forge_version_short: String,
        settings: ClientSettings,
    },
    Snapshot {
        subject: Subject,
        data: serde_json::Value,
    },
    /// Boxed for the same reason as a command: the update is the largest
    /// thing on this wire and every other variant is small beside it.
    Update {
        update: Box<SessionUpdate>,
        /// The number the frame's oldest update was minted under, `None` for
        /// a frame belonging to no seat.
        ///
        /// **A client reads the numbers to see what it missed.** Every frame
        /// the workspace emits for a seat is numbered on that seat's counter,
        /// so a client that reads 5 and then 7 knows exactly one frame went
        /// missing and asks the seat for it (`seat_range`); the snapshot's
        /// `header.seq` says where a fresh reader joins.
        seq: Option<u64>,
        /// How many of the seat's numbers this frame stands for: 1 for a
        /// frame that is one update, and n for a run the batch folded into
        /// one - a run of token appends crosses as the single frame they
        /// draw as, and its span is what keeps the client from reading the
        /// numbers it already drew as a gap.
        ///
        /// **0 only on the turn the batch forges from a delivery**, which
        /// has no number; every other frame stands for at least one, the
        /// out-of-band `read_call_output` answer (`seq: null`, `span: 1`)
        /// included.
        span: u64,
    },
    /// A page of one conversation, in answer to `more`: whole turns, each with
    /// the key the server named it by and the messages it ran as. Grouping a
    /// run of calls inside a turn is the client's, so the messages cross and no
    /// drawing of them does.
    Page {
        conversation: SessionSlot,
        turns: Vec<crate::transport::wire::TurnWire>,
        cursor: Option<String>,
        /// The schedules the fired prompts in `turns` came from, keyed by the
        /// prompt's own id: what titles a cron row a page carries long after
        /// its delivery frame is gone.
        #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
        cron_names: std::collections::BTreeMap<String, forge_primitives::cron::CronName>,
    },
    /// The answer to `seat_range`: the frames numbered in `from..=to`, each
    /// with the number it was minted under, oldest first.
    ///
    /// **Answered from the seat's own window while it holds them, and from the
    /// transcript below that.** A range the window has shed comes back as the
    /// conversation's own rows numbered to END at the range's end, so a client
    /// resumes where it was reading instead of asking again for a stretch that
    /// cannot arrive. A range from another epoch, or one forge holds no seat
    /// for, is an `error` naming `seat_range`.
    SeatFrames {
        slot: SessionSlot,
        epoch: Option<forge_primitives::SessionId>,
        frames: Vec<SeatFrameWire>,
    },
    /// The answer to `commit_diff`: one commit's own bounded diff, in the
    /// same layer shape the record's content read carries, so a client
    /// draws both scopes through one parse.
    ///
    /// **A `sha` git cannot read is the scan's own `scan_failed` layer, not
    /// an `error`** - the diff-viewer contract keeps a read that answered
    /// something apart from a read that could not run, and the client draws
    /// the state as itself. Only a seat forge holds no session for is an
    /// `error` naming `commit_diff`.
    CommitDiff {
        sha: String,
        files: LayerState<Vec<FileContent>>,
        /// Something was withheld for the read's budget, the flag the
        /// record's own content read carries.
        truncated: bool,
    },
    /// The answer to `file_content`: the file's lines at the revision asked
    /// for, or why the read could not run.
    ///
    /// `session`, `path` and `sha` are named back so a page holding two
    /// seats, two scopes or several files routes each answer to its own ask.
    FileContent {
        session: SessionSlot,
        path: String,
        sha: Option<String>,
        content: crate::transport::wire::FileContentWire,
    },
    /// The answer to a `Command` that carried a `reply_to`, including a
    /// refusal, so a client awaiting one is never left watching a channel
    /// that stays empty.
    Reply {
        reply_to: u64,
        body: serde_json::Value,
    },
    /// One tab's viewer state: which viewer controls it, and how many are
    /// attached.
    ///
    /// Sent on attach and whenever either changes, so every viewer draws the
    /// same truth about who is driving. `id` is the profile's wire id, the
    /// same one the binary frames carry; `tab` is the tab this state is for;
    /// and `viewer` is THIS connection's own - the number `controller` names
    /// when this viewer holds the role, so a bar can say "Take control"
    /// rather than guessing.
    BrowserView {
        id: u16,
        profile: String,
        tab: u16,
        controller: Option<u64>,
        viewers: u32,
        viewer: u64,
    },
    /// One terminal's screen, as a VT stream a client can draw, and where the
    /// byte stream continues from.
    TerminalView {
        seat: SessionSlot,
        id: u16,
        cols: u16,
        rows: u16,
        /// The last sequence the repaint reflects; the next frame carries
        /// one above it.
        seq: u32,
        /// The visible screen as a VT stream, valid UTF-8.
        screen: String,
        /// How many clients are attached.
        viewers: u32,
        /// This connection's own viewer number, so a control change can say
        /// whether the holder is this viewer.
        viewer: u64,
        /// The connection holding the person's control, `None` when the agent
        /// does.
        controller: Option<u64>,
    },
    /// Who holds a terminal's input, pushed to every viewer of it on a take
    /// or a release.
    TerminalControl {
        seat: SessionSlot,
        id: u16,
        controller: Option<u64>,
    },
    /// Where the viewed tab's history sits: what the bar's arrows may do.
    BrowserNavState {
        id: u16,
        tab: u16,
        can_back: bool,
        can_forward: bool,
    },
    /// The viewed tab's browser stopped.
    ///
    /// A host that died under a viewer is said out loud: a frozen last frame
    /// reads as a slow page, and the person has no way to tell the two apart.
    BrowserSurfaceFailed {
        id: u16,
        tab: u16,
        reason: String,
    },
    /// The tabs of the viewer's profile, as a client's strip draws them.
    ///
    /// Pushed to a connection that asked with `browser_tabs`, once per change:
    /// `up` is the host process, `connected` is a driver attached to it, and
    /// `closing_at_ms` is the idle clock's deadline.
    BrowserTabs {
        id: u16,
        profile: String,
        up: bool,
        connected: bool,
        closing_at_ms: Option<u64>,
        rows: Vec<forge_primitives::browser::BrowserTabRow>,
    },
    /// A page asked for a file: the shell opens its picker and answers
    /// [`ClientMessage::BrowserFileAnswer`] under the same id. **`tab` is the
    /// tab that asked**, so a viewer is asked only for the one it is on.
    BrowserFileDialog {
        id: u64,
        tab: u16,
        multiple: bool,
    },
    /// A page's download finished on the server: the shell opens its save
    /// door and writes the bytes, or says `error` and writes nothing. `tab`
    /// is the tab it started from.
    BrowserDownload {
        /// The tab the download started from. **Absent on a profile-level
        /// notice**: the backlog carries every tab's offers, so the ones it
        /// dropped may belong to any tab of the profile.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        tab: Option<u16>,
        name: String,
        /// The file's bytes, base64; absent when `error` says why not.
        data: Option<String>,
        error: Option<String>,
        /// Whether the offer was dropped before the viewer read it, rather
        /// than a save that failed: the shell titles the two differently.
        /// Additive and defaulted - an older client reads the loss as the
        /// refusal it already draws.
        #[serde(default)]
        dropped: bool,
        /// How many offers the backlog dropped. **The title says one loss or
        /// many from it**, so it agrees with the count in the body. Additive
        /// and absent on a save.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        lost: Option<u64>,
    },
    Error {
        what: String,
        why: String,
        /// The conversation the refusal belongs to, where it belongs to one: a
        /// `more` for a seat the server could not answer names its seat, so a
        /// client holding several seats' asks drains only its own. Optional and
        /// additive - a client of either age reads it, and a refusal that names
        /// no seat is read the way it always was.
        seat: Option<SessionSlot>,
        /// The dispatch a refused command answered by, echoed from what the
        /// press minted: a client that holds a draft until its own dispatch
        /// answers tells this refusal from any other by it.
        dispatch_id: Option<u64>,
    },
    /// One document, as the read answered: the seat it was read for, the
    /// path it resolved to, the bytes, the detected kind, the flag the
    /// hand-off passed, the document's own hash, and the review the
    /// document already carries.
    ///
    /// `session` is named back so a client holding two seats routes each
    /// answer to its own viewer - the same reason a refusal names its seat.
    /// `bytes` are base64: the ceiling makes a document up to a third
    /// larger than its file, well inside the socket's own frame bound.
    /// `document_kind` and not `kind`: a field named `kind` collides with
    /// the envelope's own tag.
    ///
    /// `threads` and `reviews` are the document scope's own (`doc:<path>`)
    /// store read, taken in the same handler: the viewer draws its comments,
    /// and each thread carries where its anchor sits in THIS document - the
    /// code review's re-anchoring, resolved here because only the server
    /// holds the whole file's lines. A thread the revision left nowhere is
    /// carried with an `outdated` landing rather than dropped.
    /// `content_hash` crosses as DECIMAL DIGITS for the same reason a
    /// thread's anchor hash does - JS numbers are exact only to 2^53.
    DocumentRead {
        session: SessionSlot,
        path: String,
        document_kind: DocumentKind,
        bytes: String,
        reviewable: bool,
        content_hash: String,
        threads: Vec<ReviewedThreadWire>,
        reviews: Vec<forge_primitives::ReviewSet>,
    },
}

/// One repaired frame: the number the seat minted for it, and the update
/// itself in the same shape a live `update` carries.
///
/// **The same shape is the point.** A repaired frame folds through the same
/// pipeline a live one does, so a client has one path for a conversation's
/// frames rather than a second one for the frames it had to ask for.
#[derive(Debug, Serialize, Deserialize)]
pub struct SeatFrameWire {
    pub seq: u64,
    pub update: SessionUpdate,
}

/// What the client draws with, from `forge.toml` by way of the server.
///
/// The client never reads the config file: `[client] mark`, `[client] theme`
/// and `[client] font` arrive here, and `None` on any of them means the
/// built-in rather than a pinned value - a commented line in a hand-authored
/// file has to mean "unset".
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ClientSettings {
    pub mark: Option<String>,
    pub theme: Option<String>,
    pub font: Option<String>,
    /// The axes a client that captures starts on and resets to: the
    /// `[dictate]` keys, over the crate's own defaults. A client that
    /// captures sends the values in force with each take, so this is the
    /// default it departs from rather than anything the server remembers.
    pub dictate: DictateAxes,
}

impl ClientSettings {
    /// The client's settings: the `[client]` keys off the config, and the
    /// axes the workspace resolved.
    pub fn new(client: &ClientConfig, dictate: DictateAxes) -> Self {
        Self {
            mark: client.mark.clone(),
            theme: client.theme.clone(),
            font: client.font.clone(),
            dictate,
        }
    }
}

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

    /// The pool's frame with an empty pool, which is all these routing tests
    /// 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(),
            },
        }
    }

    /// **The pool's fields are inline, not nested, and every field the client
    /// narrows crosses.** The client reads the frame's payload with the same
    /// function it reads a home snapshot's `accounts` field with, so
    /// `accounts_changed.loading` has to be the row list itself, and the
    /// gateway, the usage rows and their names have to be the same shapes.
    #[test]
    fn the_pools_frame_carries_the_pool_inline() {
        let mut update = accounts_changed();
        let SessionUpdate::AccountsChanged { accounts } = &mut update else {
            unreachable!("the helper builds this variant")
        };
        accounts.loading.push(forge_workspace::AccountLoadingRow {
            display_name: "A".to_owned(),
            state: forge_workspace::LoadingState::Ready,
            last_error: None,
            retry_after: Some(std::time::Duration::from_secs(30)),
            auth: forge_primitives::account::AccountAuth::Token,
            at_cap: true,
            cooling: false,
        });
        accounts.all_loaded = true;
        accounts.gateway.ready = true;
        accounts.gateway.port = 8787;
        accounts.gateway.bind_error = Some("port in use".to_owned());
        accounts.usage.push(forge_workspace::AccountUsageView {
            display_name: "A".to_owned(),
            snapshot: None,
        });

        let json = serde_json::to_value(&update).expect("encode");
        let body = json.get("accounts_changed").expect("the variant tag is the frame's key");
        assert!(body.get("accounts").is_none(), "the payload is not nested twice: {body}");
        assert_eq!(body["all_loaded"], true);
        assert_eq!(body["gateway"]["port"].as_u64(), Some(8787), "the port crosses: {body}");
        assert_eq!(body["gateway"]["bind_error"], "port in use");
        assert_eq!(body["loading"][0]["display_name"], "A");
        assert_eq!(body["loading"][0]["state"], "ready");
        assert_eq!(body["loading"][0]["at_cap"], true, "the selection signal crosses");
        assert_eq!(body["loading"][0]["cooling"], false);
        assert_eq!(body["usage"][0]["display_name"], "A");
        assert!(body["usage"][0]["snapshot"].is_null(), "an absent snapshot crosses as null");
        let back: SessionUpdate = serde_json::from_value(json).expect("decode");
        let SessionUpdate::AccountsChanged { accounts } = back else {
            panic!("the frame decoded into another variant")
        };
        assert_eq!(accounts.loading.len(), 1, "the row survives the round trip");
        assert_eq!(accounts.gateway.bind_error.as_deref(), Some("port in use"));
        assert_eq!(accounts.usage.len(), 1);
    }

    /// **A file chooser and its answer name the same dialog.** The answer's
    /// bytes cross base64 because the wire is JSON, and a client that lost
    /// the id would hand the server files for a dialog it cannot match.
    #[test]
    fn a_browser_file_dialog_and_its_answer_round_trip() {
        let ask = ServerMessage::BrowserFileDialog { id: 7, tab: 2, multiple: true };
        let json = serde_json::to_string(&ask).expect("encode");
        assert!(json.contains("\"kind\":\"browser_file_dialog\""), "{json}");
        let ServerMessage::BrowserFileDialog { id, tab, multiple } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!((id, multiple), (7, true));

        let answer = ClientMessage::BrowserFileAnswer {
            id: 7,
            files: vec![BrowserFileWire { name: "cv.pdf".to_owned(), data: "AAEC".to_owned() }],
        };
        let json = serde_json::to_string(&answer).expect("encode");
        assert!(json.contains("\"kind\":\"browser_file_answer\""), "{json}");
        let ClientMessage::BrowserFileAnswer { id, files } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(id, 7);
        assert_eq!(tab, 2, "the ask names the tab it belongs to");
        assert_eq!(files[0].name, "cv.pdf");
        assert_eq!(files[0].data, "AAEC", "the bytes cross base64");

        // **A download carries either bytes or its refusal, never both.**
        let saved = ServerMessage::BrowserDownload {
            tab: Some(2),
            name: "report.pdf".to_owned(),
            data: Some("AAEC".to_owned()),
            error: None,
            dropped: false,
            lost: None,
        };
        let json = serde_json::to_string(&saved).expect("encode");
        assert!(json.contains("\"kind\":\"browser_download\""), "{json}");
        let ServerMessage::BrowserDownload { tab, name, data, error, dropped, lost } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(tab, Some(2), "the save names the tab it started from");
        assert_eq!(name, "report.pdf");
        assert_eq!(data.as_deref(), Some("AAEC"));
        assert!(error.is_none());
        assert!(!dropped, "a save the person can take is not a dropped offer");
        assert!(lost.is_none(), "and a save carries no loss count");

        let refused = ServerMessage::BrowserDownload {
            tab: Some(2),
            name: "huge.bin".to_owned(),
            data: None,
            error: Some("this file is larger than 32 MiB".to_owned()),
            dropped: false,
            lost: None,
        };
        let json = serde_json::to_string(&refused).expect("encode");
        let ServerMessage::BrowserDownload { data, error, dropped, .. } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert!(data.is_none(), "a refusal carries no bytes");
        assert!(error.is_some_and(|why| why.contains("32 MiB")));
        assert!(!dropped, "a write that failed is not a dropped offer");

        // **A drop carries the flag the shell reads**, under that key: the
        // two samples above both say false, and a rename or a skip would
        // leave the shell's title silently wrong. **It names no tab and
        // carries its count**: the loss is the profile's, and the title is
        // pluralized from `lost`.
        let dropped = ServerMessage::BrowserDownload {
            tab: None,
            name: "download".to_owned(),
            data: None,
            error: Some("2 finished downloads were dropped".to_owned()),
            dropped: true,
            lost: Some(2),
        };
        let json = serde_json::to_string(&dropped).expect("encode");
        assert!(json.contains("\"dropped\":true"), "the wire key the shell reads: {json}");
        assert!(json.contains("\"lost\":2"), "and the count the title is read from: {json}");
        assert!(!json.contains("\"tab\""), "a profile-level loss names no tab: {json}");
        let ServerMessage::BrowserDownload { dropped, lost, tab, .. } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert!(dropped, "and it round-trips as a drop");
        assert_eq!(lost, Some(2), "with its count");
        assert_eq!(tab, None, "and no tab");
    }

    /// **A viewer's attach, its state, and the input it sends.**
    ///
    /// The profile id is what pairs the state with the binary frames, and the
    /// tab is which of them the viewer is on; the input carries the person's
    /// gesture and its own sequence, never a CDP method - the server is what
    /// maps one to the other.
    #[test]
    fn a_browser_view_and_its_input_round_trip() {
        let attached = ClientMessage::BrowserView {
            seat: SessionSlot::lead("TestOrg", "proj"),
            tab: Some(3),
            reply_to: Some(41),
        };
        let json = serde_json::to_string(&attached).expect("encode");
        assert!(json.contains("\"kind\":\"browser_view\""), "{json}");
        let ClientMessage::BrowserView { seat, tab, reply_to } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(seat, SessionSlot::lead("TestOrg", "proj"), "the seat crosses");
        assert_eq!(tab, Some(3), "and the tab it named");
        assert_eq!(reply_to, Some(41), "and the answer the asker waits on");

        let paged = ClientMessage::BrowserViewPage {
            seat: SessionSlot::lead("TestOrg", "proj"),
            target: "page-3".to_owned(),
            reply_to: Some(42),
        };
        let json = serde_json::to_string(&paged).expect("encode");
        assert!(json.contains("\"kind\":\"browser_view_page\""), "{json}");
        let ClientMessage::BrowserViewPage { seat, target, reply_to } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(seat, SessionSlot::lead("TestOrg", "proj"), "the seat crosses");
        assert_eq!(target, "page-3", "and the page the hand-off named");
        assert_eq!(reply_to, Some(42), "and the answer the asker waits on");

        let view = ServerMessage::BrowserView {
            id: 3,
            profile: "shared".to_owned(),
            tab: 2,
            controller: Some(9),
            viewers: 2,
            viewer: 9,
        };
        let json = serde_json::to_string(&view).expect("encode");
        assert!(json.contains("\"kind\":\"browser_view\""), "{json}");
        let ServerMessage::BrowserView { id, tab, controller, viewers, viewer, .. } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(id, 3, "the id the frames carry");
        assert_eq!(tab, 2, "the tab the state is for");
        assert_eq!(controller, Some(9), "who holds the role");
        assert_eq!(viewers, 2, "and how many are attached");
        assert_eq!(viewer, 9, "and this connection's own number");

        let input = ClientMessage::BrowserInput {
            seq: 4,
            event: forge_primitives::browser::BrowserInputEvent::Mouse {
                action: forge_primitives::browser::MouseAction::Press,
                x: 1.0,
                y: 2.0,
                button: 0,
                clicks: 1,
                modifiers: forge_primitives::browser::InputModifiers::default(),
            },
        };
        let json = serde_json::to_string(&input).expect("encode");
        assert!(json.contains("\"kind\":\"browser_input\""), "{json}");
        assert!(json.contains("\"action\":\"press\""), "the gesture crosses as itself: {json}");
        let ClientMessage::BrowserInput { seq, event } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(seq, 4);
        assert!(matches!(event, forge_primitives::browser::BrowserInputEvent::Mouse { .. }));

        let failed = ServerMessage::BrowserSurfaceFailed {
            id: 3,
            tab: 2,
            reason: "the browser host exited".to_owned(),
        };
        let json = serde_json::to_string(&failed).expect("encode");
        assert!(json.contains("\"kind\":\"browser_surface_failed\""), "{json}");
        assert!(json.contains("\"tab\":2"), "the failure names the tab that stopped: {json}");
    }

    /// **The strip's whole read crosses as one list**, so a client draws the
    /// tabs, the host's health and the idle deadline from one frame rather
    /// than a read per field. The rows carry the owning session, which is what
    /// the strip groups by.
    #[test]
    fn a_browser_tabs_list_carries_its_rows_and_the_profile_state() {
        let asked = ClientMessage::BrowserTabs {
            seat: SessionSlot::lead("TestOrg", "proj"),
            reply_to: Some(7),
        };
        let json = serde_json::to_string(&asked).expect("encode");
        assert!(json.contains("\"kind\":\"browser_tabs\""), "{json}");

        let list = ServerMessage::BrowserTabs {
            id: 4,
            profile: "TestOrg/default".to_owned(),
            up: true,
            connected: false,
            closing_at_ms: Some(1_760_000_000_000),
            rows: vec![forge_primitives::browser::BrowserTabRow {
                number: 2,
                title: "portal".to_owned(),
                url: "https://x/".to_owned(),
                owner: SessionSlot::worker("TestOrg", "proj", "w"),
                opener: None,
                state: forge_primitives::browser::BrowserTabState::InUse,
                dialog: Some(forge_primitives::browser::BrowserTabDialog {
                    kind: "confirm".to_owned(),
                    message: "Leave this page?".to_owned(),
                }),
            }],
        };
        let json = serde_json::to_string(&list).expect("encode");
        assert!(json.contains("\"kind\":\"browser_tabs\""), "{json}");
        assert!(
            json.contains("\"dialog\":{\"kind\":\"confirm\""),
            "the dialog crosses with its kind: {json}"
        );
        let ServerMessage::BrowserTabs { id, up, connected, closing_at_ms, rows, .. } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(id, 4, "the id the binary frames carry");
        assert!(up && !connected, "the host and the driver are two facts");
        assert_eq!(closing_at_ms, Some(1_760_000_000_000));
        assert_eq!(rows.len(), 1, "and the rows the strip draws");
        assert_eq!(rows[0].owner, SessionSlot::worker("TestOrg", "proj", "w"));
    }

    /// A subscribe that says nothing about answering reads as one that cannot
    /// answer: the core parks a turn on an answering client, so the default
    /// has to be the safe one.
    #[test]
    fn a_subscribe_defaults_to_not_answering() {
        let older: ClientMessage = serde_json::from_str(r#"{"kind":"subscribe","what":"home"}"#)
            .expect("a bare subscribe decodes");
        let ClientMessage::Subscribe { answering, .. } = older else {
            panic!("a bare subscribe is still a subscribe")
        };
        assert!(!answering, "a client that says nothing cannot answer");
    }

    /// **A page carries the schedules its fired prompts came from.** The
    /// delivery frame names a schedule once; a row replayed from a transcript
    /// afterwards has only the read, so the pairing crosses here - typed, and
    /// keyed by the prompt's own id (#1815).
    #[test]
    fn a_page_carries_the_cron_pairing_it_serves() {
        let seat = SessionSlot::lead("TestOrg", "proj");
        let page = ServerMessage::Page {
            conversation: seat,
            turns: vec![crate::transport::wire::TurnWire {
                key: Some("t1".to_owned()),
                messages: vec![serde_json::json!({ "type": "user", "uuid": "p-1" })],
            }],
            cursor: None,
            cron_names: std::collections::BTreeMap::from([(
                "p-1".to_owned(),
                forge_primitives::cron::CronName {
                    cron_id: "c1".to_owned(),
                    description: Some("Morning summary".to_owned()),
                },
            )]),
        };
        let json = serde_json::to_string(&page).expect("encode");
        assert!(json.contains("\"cron_names\":{\"p-1\":{\"cron_id\":\"c1\""), "{json}");
        let ServerMessage::Page { cron_names, .. } = serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(
            cron_names.get("p-1").and_then(|name| name.description.as_deref()),
            Some("Morning summary"),
            "the schedule a reloaded row titles itself by",
        );
    }

    /// The document read a viewer's route makes, and the answer it gets.
    ///
    /// The request names the seat whose tree the path resolves against; the
    /// answer names that seat BACK, so a client holding two seats routes
    /// each answer to its own viewer rather than to whichever asked last.
    #[test]
    fn a_document_read_request_and_its_answer_round_trip_through_json() {
        let seat = SessionSlot::lead("TestOrg", "proj");
        let asked = ClientMessage::DocumentRead {
            session: seat.clone(),
            path: "docs/readme.md".to_owned(),
            reviewable: true,
        };
        let json = serde_json::to_string(&asked).expect("encode");
        assert!(json.contains("\"kind\":\"document_read\""), "{json}");
        let ClientMessage::DocumentRead { session, path, reviewable } =
            serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(session, seat, "the seat whose tree the path resolves against");
        assert_eq!(path, "docs/readme.md");
        assert!(reviewable, "the reviewable flag the request carried");

        let answer = ServerMessage::DocumentRead {
            session: seat.clone(),
            path: "/Users/me/proj/docs/readme.md".to_owned(),
            document_kind: DocumentKind::Markdown,
            bytes: "aGk=".to_owned(),
            reviewable: true,
            content_hash: "18446744073709551615".to_owned(),
            threads: Vec::new(),
            reviews: Vec::new(),
        };
        let json = serde_json::to_string(&answer).expect("encode");
        assert!(json.contains("\"kind\":\"document_read\""), "{json}");
        assert!(
            json.contains("\"content_hash\":\"18446744073709551615\""),
            "the document's hash crosses as decimal digits, as a thread's anchor does: {json}",
        );
        let ServerMessage::DocumentRead {
            session,
            path,
            document_kind,
            bytes,
            reviewable,
            content_hash,
            threads,
            reviews,
        } = serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(session, seat, "the answer names its seat back");
        assert_eq!(path, "/Users/me/proj/docs/readme.md", "the resolved absolute path");
        assert_eq!(document_kind, DocumentKind::Markdown, "the kind the read detected");
        assert_eq!(bytes, "aGk=", "the document's bytes, base64, as they cross");
        assert!(reviewable);
        assert_eq!(content_hash, "18446744073709551615", "the hash survives the round trip");
        assert!(threads.is_empty(), "an uncommented document carries no threads");
        assert!(reviews.is_empty(), "and no review sets");
    }

    /// **A page's perf batch crosses as its own message**, with the records
    /// exactly as the page serialised them: the server writes them, it does
    /// not re-shape them, so a field the page learned reaches the file.
    #[test]
    fn a_client_perf_batch_round_trips_with_its_records() {
        let batch = ClientMessage::ClientPerf {
            records: vec![serde_json::json!({
                "schema": "forge-perf/v1",
                "kind": "frame",
                "run_id": "r",
                "ts_ms": 3,
                "ms": 120.5,
            })],
        };
        let json = serde_json::to_string(&batch).expect("encode");
        assert!(json.contains("\"kind\":\"client_perf\""), "{json}");
        let ClientMessage::ClientPerf { records } = serde_json::from_str(&json).expect("decode")
        else {
            panic!("{json} decoded into another message")
        };
        assert_eq!(records.len(), 1);
        assert_eq!(records[0]["kind"], "frame", "the page's own shape survives");
        assert_eq!(records[0]["ms"], 120.5);
    }

    #[test]
    fn a_subscribe_round_trips_through_json() {
        let sent = ClientMessage::Subscribe { what: Subject::Home, answering: true, review: false };
        let json = serde_json::to_string(&sent).expect("encode");
        assert!(json.contains("\"kind\":\"subscribe\""), "{json}");

        let back: ClientMessage = serde_json::from_str(&json).expect("decode");
        let ClientMessage::Subscribe { what, answering, .. } = back else {
            panic!("{json} decoded into another message");
        };
        assert_eq!(what, Subject::Home, "the subject survives the round trip");
        assert!(answering, "and so does the capability the client declared");
    }

    /// **The exact frames Dock.svelte writes for a vault answer decode.** The
    /// fixture is the same file the client test asserts the dock writes, so
    /// neither side can drift into a different nesting on its own.
    #[test]
    fn the_docks_vault_answer_frames_decode() {
        use forge_primitives::vault::VaultAskDecision;

        let frames: serde_json::Value = serde_json::from_str(include_str!(
            "../../../../client/src/composer/vault-answer-frames.json"
        ))
        .expect("the shared frames are json");

        for name in ["approve", "send_back"] {
            let frame = frames.get(name).expect("the shared frame exists");
            let json = serde_json::json!({
                "kind": "command",
                "command": frame,
                "reply_to": null,
            })
            .to_string();
            let ClientMessage::Command { command, reply_to } =
                serde_json::from_str(&json).expect("the dock's answer decodes")
            else {
                panic!("the dock's answer decoded into another message");
            };
            assert!(reply_to.is_none());
            let Command::RespondVaultAsk { answer, .. } = *command else {
                panic!("the dock's answer carries another command");
            };
            match name {
                "approve" => {
                    assert_eq!(answer.decision, VaultAskDecision::Approve);
                    assert_eq!(answer.master_password.as_deref(), Some("open sesame"));
                    assert_eq!(answer.two_factor.as_deref(), Some("123456"));
                }
                _ => assert!(
                    matches!(
                        &answer.decision,
                        VaultAskDecision::SendBack { note }
                            if note == "ask for the password, not the TOTP"
                    ),
                    "the send-back variant survives: {:?}",
                    answer.decision,
                ),
            }
        }
    }

    /// The models subject hears its own landing and the load's signal -
    /// the only push that says the models finished loading - and nothing
    /// else. It is the whole live half of that page: a subscriber that
    /// never hears a landing reads as quiet, not as broken.
    #[test]
    fn the_models_subject_hears_the_landing_and_the_load() {
        let landing = SessionUpdate::DictateModelsChanged {
            models: forge_workspace::catalogue::DictateModelsSnapshot {
                enabled: true,
                models_dir: None,
                speaking: forge_workspace::TtsView {
                    enabled: false,
                    snapshot: forge_workspace::TtsSnapshot::default(),
                    models_dir: None,
                    speaking_model: forge_workspace::tts::SPEAKING_MODEL.to_owned(),
                    active: forge_workspace::tts::SpeakingVariantView {
                        variant: forge_workspace::tts::PINNED_VARIANT.to_owned(),
                        file: "model_quantized.onnx".to_owned(),
                        quant: "int8".to_owned(),
                        size: 0,
                        installed: false,
                        active: true,
                    },
                    from: forge_workspace::tts::SpeakingFrom::Pin,
                    variants: Vec::new(),
                    voices: Vec::new(),
                    voice: None,
                    preview_error: None,
                },
                in_use: Vec::new(),
                check: forge_workspace::catalogue::CatalogueCheck::Never,
                updates: Vec::new(),
                rows: Vec::new(),
                install: forge_workspace::install::InstallState::Idle,
                activate: forge_workspace::install::ActivateState::Idle,
                installed: Vec::new(),
                bench: forge_workspace::bench::BenchState::Idle,
                results: Vec::new(),
                speaking_bench: forge_workspace::bench::SpeakingBenchState::Idle,
                speaking_results: Vec::new(),
                read_aloud: forge_workspace::bench::ReadAloudState {
                    recordings: Vec::new(),
                    recording: false,
                    error: None,
                    passage: String::new(),
                    terms: Vec::new(),
                },
            },
        };

        assert!(
            Subject::DictateModels.covers(&landing),
            "the check's landing is the subject's whole live half",
        );
        assert!(
            Subject::DictateModels.covers(&SessionUpdate::DictateAvailability),
            "and the preflight's signal, so a page open during a first download re-reads its \
             state chips instead of drawing `pending` until the next check",
        );
        assert!(
            !Subject::DictateModels.covers(&accounts_changed()),
            "another machine-level update is not the models page's to redraw",
        );
        assert!(
            Subject::Home.covers(&landing),
            "and the slot-less arm carries it home, which is what the client's own census \
             classifies it against",
        );
    }

    /// The pool belongs to no seat, so a home subscription hears it and a
    /// seat's does not.
    ///
    /// **Both home arms carry it, and the slot-less one is the load-bearing
    /// assertion here.** `fleet_news` classifies it a redraw because the
    /// band's own card is part of the region the web home re-sends; taking it
    /// out of that arm would leave this test green, because the slot-less arm
    /// below delivers it anyway. What the test can fail on is the variant
    /// gaining a seat: a key would send the pool to one session's page and
    /// leave every other subscriber drawing a card that never moves.
    #[test]
    fn the_pools_announcement_reaches_home_and_no_seat() {
        let update = accounts_changed();
        let seat = SessionSlot::lead("TestOrg", "proj");

        assert!(update.slot().is_none(), "the pool addresses no seat, which is what routes it");
        assert!(Subject::Home.covers(&update), "home is the only subscription that could carry it");
        assert!(
            !Subject::Session(seat).covers(&update),
            "a seat's subscription is a seat, and the pool is not one",
        );
    }

    /// The output read reaches the seat that asked and not the home: it is a
    /// row's own content, and no home row draws a word of conversation.
    ///
    /// **The keyed pair is what this fails on.** A `CallOutput` that lost its
    /// slot would be taken by the home's slot-less arm - sending one seat's
    /// tail to every fleet subscriber - so `Session(covers)` and
    /// `Home(does not)` are asserted together.
    #[test]
    fn the_output_read_reaches_its_seat_and_not_the_home() {
        let seat = SessionSlot::lead("TestOrg", "proj");
        let update = SessionUpdate::CallOutput {
            key: seat.clone(),
            call_id: "tu-1".to_owned(),
            output: forge_primitives::CallOutput::NoPath,
        };

        assert!(
            Subject::Session(seat.clone()).covers(&update),
            "the seat that asked is sent its own answer",
        );
        assert!(
            !Subject::Home.covers(&update),
            "and the home is not: no row draws a conversation's words, and the slot-less arm \
             would take it if the key went",
        );
    }

    /// **The Extensions page hears the whole `Plugins` family and nothing
    /// else.** A variant dropped from the arm leaves the page drawing a run
    /// that ended, or an inventory an action already moved, with no way to
    /// tell it has fallen behind; a variant added to the arm would hand the
    /// page a seat's own news.
    #[test]
    fn extensions_covers_the_plugins_family_and_no_seat_news() {
        for update in [
            SessionUpdate::PluginsInventoryUpdated {
                cwd_raw: "/tmp/proj".to_owned(),
                snapshot: forge_primitives::plugins::PluginsInventorySnapshot::default(),
                claude_path: std::path::PathBuf::from("claude"),
            },
            SessionUpdate::PluginsInventoryRefreshFailed {
                cwd_raw: "/tmp/proj".to_owned(),
                message: "no".to_owned(),
                trigger: forge_primitives::plugins::PluginUpdateTrigger::Manual,
            },
            SessionUpdate::PluginsCliActionSucceeded {
                cwd_raw: "/tmp/proj".to_owned(),
                result: forge_primitives::plugins::PluginsCliActionSuccess {
                    snapshot: forge_primitives::plugins::PluginsInventorySnapshot::default(),
                    message: "done".to_owned(),
                    claude_path: std::path::PathBuf::from("claude"),
                },
                dispatch_id: Some(1),
            },
            SessionUpdate::PluginsCliActionFailed {
                cwd_raw: "/tmp/proj".to_owned(),
                message: "no".to_owned(),
                dispatch_id: None,
            },
            SessionUpdate::PluginsUpdateRunProgress {
                cwd_raw: "/tmp/proj".to_owned(),
                run: forge_primitives::plugins::PluginUpdateRun {
                    trigger: forge_primitives::plugins::PluginUpdateTrigger::Manual,
                    finished: false,
                    rows: Vec::new(),
                },
            },
            SessionUpdate::PluginsUpdateRunFinished {
                cwd_raw: "/tmp/proj".to_owned(),
                run: forge_primitives::plugins::PluginUpdateRun {
                    trigger: forge_primitives::plugins::PluginUpdateTrigger::Manual,
                    finished: true,
                    rows: Vec::new(),
                },
                snapshot: None,
                claude_path: None,
            },
            SessionUpdate::PluginsRollbackSucceeded {
                cwd_raw: "/tmp/proj".to_owned(),
                plugin_id: "a@probe".to_owned(),
                scope: "user".to_owned(),
                message: "done".to_owned(),
                snapshot: forge_primitives::plugins::PluginsInventorySnapshot::default(),
                claude_path: std::path::PathBuf::from("claude"),
            },
            SessionUpdate::PluginsRollbackFailed {
                cwd_raw: "/tmp/proj".to_owned(),
                plugin_id: "a@probe".to_owned(),
                message: "no".to_owned(),
                snapshot: None,
            },
        ] {
            assert!(
                Subject::Extensions.covers(&update),
                "the Extensions page must hear {update:?}; a dropped arm leaves it stale",
            );
        }

        let seat = SessionSlot::lead("TestOrg", "proj");
        let session_news = SessionUpdate::TurnComplete { key: seat.clone(), terminal_reason: None };
        assert!(
            !Subject::Extensions.covers(&session_news),
            "a seat's own news is not the Extensions page's",
        );
        assert!(
            !Subject::Extensions.covers(&SessionUpdate::CatalogLoaded),
            "and neither is the catalog's",
        );
    }

    /// A parked draft is home news on its own: the seat's row moves into the
    /// needs-you group while the draft is held and back out when it resolves,
    /// so a home-ONLY subscriber that was never sent the pair reads the fleet
    /// as it stood before the draft - exactly the symptom #1758 was filed
    /// for, with no session page open to forward it.
    #[test]
    fn the_slack_draft_pair_reaches_home_and_its_seat() {
        let seat = SessionSlot::lead("TestOrg", "proj");
        for update in [
            SessionUpdate::SlackPostPending {
                key: seat.clone(),
                draft: forge_primitives::slack::SlackDraft {
                    id: uuid::Uuid::new_v4(),
                    workspace: "acme".to_owned(),
                    conversation: "C1".to_owned(),
                    conversation_label: "acme".to_owned(),
                    thread_ts: None,
                    text: "hello".to_owned(),
                    operation: forge_primitives::slack::SlackDraftOperation::Post,
                    thread: None,
                    tagged: Vec::new(),
                },
            },
            SessionUpdate::SlackDraftResolved {
                key: seat.clone(),
                id: uuid::Uuid::new_v4(),
                ending: forge_primitives::slack::SlackDraftEnding::Abandoned,
            },
        ] {
            assert!(
                fleet_news(&update).any(),
                "admitted by the redraw arm, not by the wildcard: {update:?}",
            );
            assert!(Subject::Home.covers(&update), "so the home is sent it: {update:?}");
            assert!(
                Subject::Session(seat.clone()).covers(&update),
                "and the seat it routes on is sent it too: {update:?}",
            );
        }
    }

    /// A document hand-off has to reach the client wherever the reader is:
    /// the ask is drawn on the raise, and the home subscription is the one
    /// every client holds. A reader sitting on the home never heard the
    /// seat-keyed raise until the redraw arm carried it (measured in the
    /// e2e, 2026-10-12), so this pins the arm by name - dropping it leaves
    /// the home silent and the ask undrawn.
    #[test]
    fn a_document_hand_off_reaches_home_and_its_seat() {
        let seat = SessionSlot::lead("TestOrg", "proj");
        for update in [
            SessionUpdate::DocumentHandOffPending {
                key: seat.clone(),
                id: uuid::Uuid::new_v4(),
                path: "/proj/docs/plan.md".to_owned(),
                reviewable: true,
            },
            SessionUpdate::DocumentHandOffResolved {
                key: seat.clone(),
                id: uuid::Uuid::new_v4(),
                ending: forge_primitives::document::DocumentHandOffEnding::NotNow,
            },
        ] {
            assert!(
                fleet_news(&update).any(),
                "admitted by the redraw arm, not by the wildcard: {update:?}",
            );
            assert!(Subject::Home.covers(&update), "so the home is sent it: {update:?}");
            assert!(
                Subject::Session(seat.clone()).covers(&update),
                "and the seat it routes on is sent it too: {update:?}",
            );
        }
    }

    /// The section pushes reach the BOARD, and the task push reaches the home
    /// too (#2517).
    ///
    /// **The home keeps `TasksChanged` because its snapshot still carries it**:
    /// the fleet row's queue and waiting counts are derived from the same
    /// tasks. Crons and connector sets left the home's payload with the rows,
    /// so the home must not hear them (rule 25); the board snapshot carries
    /// all three.
    #[test]
    fn the_section_pushes_reach_the_board_and_tasks_also_home() {
        let tasks = SessionUpdate::TasksChanged {
            key: SessionSlot::lead("TestOrg", "proj"),
            tasks: Vec::new(),
        };
        assert!(
            fleet_news(&tasks).any(),
            "admitted by the redraw arm, not by the wildcard: {tasks:?}",
        );
        assert!(Subject::Board.covers(&tasks), "the board draws every row: {tasks:?}");
        assert!(
            Subject::Home.covers(&tasks),
            "and the home's fleet counts are derived from the same tasks: {tasks:?}",
        );

        for update in [
            SessionUpdate::CronSchedulesChanged {
                key: SessionSlot::lead("TestOrg", "proj"),
                crons: Vec::new(),
            },
            SessionUpdate::ConnectorSubscriptionsChanged {
                key: SessionSlot::lead("TestOrg", "proj"),
                gotify: Vec::new(),
                slack: Vec::new(),
            },
        ] {
            assert!(Subject::Board.covers(&update), "the board draws both: {update:?}");
            assert!(
                !Subject::Home.covers(&update),
                "the home's snapshot carries neither, so it must not be asked to \
                 re-read for it: {update:?}",
            );
        }

        assert!(
            !Subject::Board.covers(&accounts_changed()),
            "and nothing outside the board's own three reaches it",
        );
    }

    /// **A board row's `stale` mark is read from seat liveness**, so the
    /// frames that move the registries behind `seat_is_live` reach the board -
    /// and a turn frame does not (#2517).
    #[test]
    fn the_seat_liveness_frames_reach_the_board_and_a_turn_frame_does_not() {
        let seat = SessionSlot::lead("TestOrg", "proj");
        let 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,
        };
        for update in [
            SessionUpdate::Spawning {
                key: seat.clone(),
                project_name: "proj".to_owned(),
                cwd: "/tmp".to_owned(),
                display_name: "lead".to_owned(),
            },
            SessionUpdate::Connected {
                key: seat.clone(),
                session_id: forge_primitives::SessionId::new("s"),
                cwd: "/tmp".to_owned(),
                current_model: model(),
                available_models: Vec::new(),
                mode: None,
                history: Vec::new(),
                dropped: 0,
                compaction_count: 0,
            },
            SessionUpdate::SessionReplaced {
                key: seat.clone(),
                session_id: forge_primitives::SessionId::new("s"),
                cwd: "/tmp".to_owned(),
                current_model: model(),
                available_models: Vec::new(),
                mode: None,
                history: Vec::new(),
                dropped: 0,
                compaction_count: 0,
            },
            SessionUpdate::Releasing { key: seat.clone() },
            SessionUpdate::ConnectionFailed {
                key: seat.clone(),
                message: "gone".to_owned(),
                fatal: false,
            },
            SessionUpdate::WorkerStatusChanged {
                project_key: forge_workspace::ProjectKey::new("TestOrg-proj"),
                action: forge_workspace::protocol::WorkerStatusAction::Removed,
                status: forge_primitives::WorkerStatus {
                    label: "w1".to_owned(),
                    charter: "a worker".to_owned(),
                    status: forge_primitives::WorkerLiveness::Running,
                    session_id: "s".to_owned(),
                    slot: seat.clone(),
                    spawned_at: std::time::SystemTime::UNIX_EPOCH,
                    spawned_by: seat.clone(),
                    diagnostic: None,
                    heal_attempt: None,
                    heal_next_at: None,
                    activity: None,
                },
                worktree: forge_workspace::protocol::WorktreeDisposition::Absent,
            },
        ] {
            assert!(
                Subject::Board.covers(&update),
                "a row's owner-gone chip is read from these: {update:?}",
            );
        }

        assert!(
            !Subject::Board
                .covers(&SessionUpdate::TurnComplete { key: seat, terminal_reason: None }),
            "a turn frame moves no board row's marks",
        );
    }

    /// The client's settings come off the server's own config, so the
    /// client never reads `forge.toml` and the two cannot drift - the
    /// `[client]` keys off it directly, and the `[dictate]` axes as the
    /// workspace resolved them.
    #[test]
    fn the_client_settings_are_the_configs_view_settings() {
        let config = ClientConfig {
            mark: Some("strike".to_owned()),
            theme: Some("dark".to_owned()),
            font: Some("system".to_owned()),
        };
        let axes = DictateAxes {
            styling: forge_dictate::normalize::Styling::Formal,
            structure: forge_dictate::normalize::Structure::Lists,
            context: forge_dictate::normalize::Context::Email,
        };

        let settings = ClientSettings::new(&config, axes);

        assert_eq!(settings.mark.as_deref(), Some("strike"));
        assert_eq!(settings.theme.as_deref(), Some("dark"));
        assert_eq!(settings.font.as_deref(), Some("system"));
        assert_eq!(settings.dictate, axes, "and the axes a capturing client starts on");
    }

    /// **The page's own JSON answers a hand-off.** The client sends its answer
    /// as `{ ending: { type }, source }`, and a shape serde accepts from Rust
    /// but not from the page would answer nothing on a live socket - which is
    /// a decline the person never gave, or worse, a park that never ends.
    #[test]
    fn a_hand_off_answer_decodes_from_the_clients_own_json() {
        let sent = serde_json::json!({
            "kind": "command",
            "command": {
                "respond_browser_hand_off": {
                    "key": { "org": "TestOrg", "project": "proj", "label": "lead" },
                    "id": "0192e1c0-0000-7000-8000-0000000000aa",
                    "answer": { "ending": { "type": "feedback" }, "source": "feedback" },
                },
            },
            "reply_to": null,
        });

        let ClientMessage::Command { command, .. } =
            serde_json::from_value::<ClientMessage>(sent).expect("the client's own shape decodes")
        else {
            panic!("a command frame is a command");
        };
        let Command::RespondBrowserHandOff { answer, .. } = *command else {
            panic!("the answer is the hand-off's");
        };
        assert_eq!(
            answer.ending,
            forge_primitives::browser::HandOffEnding::Feedback,
            "the person spoke",
        );
        assert_eq!(
            answer.source,
            forge_primitives::browser::HandOffSource::Feedback,
            "and the door that sent it is named",
        );
        assert_eq!(answer.client, None, "a client cannot know its own connection");
    }
}