forge_server/transport/

wire.rs

//! A subject's wire form, and the one place it is decided.
//!
//! The server hands a view the surface's own values, and a client reads a
//! shape built for it here rather than the core's structs. A twin per
//! subject keeps a wire change in one file instead of in a handler and a
//! test, and the fixtures beside this module are what make a changed shape
//! fail a test rather than a client.
//!
//! **The twins are complete.** A field is carried unless it is a Rust
//! mechanism rather than data - a `oneshot`, an `Arc<Workspace>` - because
//! the client is not today's pages: it is built against this wire, and a
//! field missing here cannot be added later without a server change.

use std::collections::{BTreeMap, HashMap};
use std::path::Path;
use std::sync::Arc;

use anyhow::Result;
use forge_primitives::git_diff::LayerState;
use forge_primitives::plugins::{PluginUpdateRun, PluginsInventorySnapshot};
use forge_primitives::review::{ReviewAnchor, ReviewSet, ReviewSide, ReviewThread};
use forge_primitives::runtime::{AvailableAgent, AvailableCommand, MonitorRecord};
use forge_primitives::slack::SlackSubscription;
use forge_primitives::{ContentBlock, GotifySubscription, Message, SessionSlot, UserEnvelope};
use forge_workspace::env::git_diff::content::{DiffContent, FileContent, scan as scan_content};
use forge_workspace::env::git_diff::file::{merge_base, read_file_lines};
use forge_workspace::env::git_diff::resolver::{
    AnchorPlacement, OutdatedReason, place_anchor_in_content, place_anchor_in_lines_with,
};
use forge_workspace::env::processes::ProcessSnapshot;
use forge_workspace::{McpServers, ProjectView};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::file_index::FileIndex;
use crate::surface::{AgentRow, PendingAsk, ViewSurface};
use crate::transcript::{Rendered, TaskEnding, TurnSpan};
use crate::transport::TransportState;
use crate::transport::conversation::{Conversation, Held};
use crate::transport::envelope::Subject;
use crate::work::{WorkState, git_work_view, work_from_scan};

/// The server run's own identity, minted once.
///
/// **A client's ask marks outlive a forge restart while the counter does
/// not**, so the roster carries this and every mark is scoped by it.
fn boot_id() -> &'static str {
    static BOOT: std::sync::OnceLock<String> = std::sync::OnceLock::new();
    BOOT.get_or_init(|| {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |held| held.as_nanos());
        format!("{}-{nanos}", std::process::id())
    })
}

/// The seat the fixture surface is populated for.
///
/// One seat is enough: a fixture pins the SHAPE of a snapshot, and a second
/// seat would pin the same shape twice.
pub fn fixture_seat() -> SessionSlot {
    SessionSlot::lead("TestOrg", "proj")
}

/// One of the surface's two independent review reads: the records, or why
/// they could not be read. They fail apart, so neither is collapsed into
/// the other, and a client can tell an empty branch from an unreadable one.
#[derive(Serialize, Deserialize)]
#[serde(tag = "outcome", content = "value", rename_all = "snake_case")]
pub enum ReadWire<T> {
    Read(T),
    Failed(String),
}

impl<T> From<std::result::Result<T, String>> for ReadWire<T> {
    fn from(read: std::result::Result<T, String>) -> Self {
        match read {
            Ok(records) => Self::Read(records),
            Err(why) => Self::Failed(why),
        }
    }
}

/// The home, as a client sees it: every read a home-scoped view makes.
///
/// The last two are the App-level state that reaches a home subscriber as
/// an update and nothing else - a subject's snapshot has to cover every
/// update that subject receives, or a late subscriber sees a smaller world
/// than one that attached at boot.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct HomeWire {
    pub projects: Vec<ProjectWire>,
    pub agents: Vec<AgentWire>,
    /// One row per project for the fleet page: the counts and the named
    /// misses (a stalled queue, a worker holding no row). A project's own
    /// rows ride [`ProjectBoardWire::rows`]; this is the glance.
    pub fleet: Vec<forge_workspace::board::FleetRow>,
    /// The seats whose last turn finished while no client was showing them.
    ///
    /// A view draws a mark per row from this, and it is the fact a
    /// viewer cannot reconstruct: a turn that ended before a client attached
    /// leaves nothing in the transcript to say it went unwatched, so a client
    /// reading only the records would draw every row as settled.
    pub unseen: Vec<SessionSlot>,
    /// The pool as both the snapshot and `AccountsChanged` state it; see
    /// [`forge_workspace::AccountPoolView`].
    pub accounts: forge_workspace::AccountPoolView,
    pub connectors: ConnectorsWire,
    pub dictate: DictateWire,
    /// The speaking role's switch and model directory: the machine-wide
    /// facts a voice control reads before it names a seat.
    pub voice: VoiceWire,
    /// The claude CLI versions the core holds, as the core's own snapshot
    /// serialises. Its crate is one this one may not name.
    pub cli_version: Option<Value>,
    /// The forge build serving this socket: the full stamp and the short one.
    ///
    /// **A client draws the short one as its `server` build, beside its own
    /// labelled `client` build.** The header states which forge commit is
    /// RUNNING, and the client is a different program with a version of its
    /// own, so the two are named apart rather than one standing for both.
    pub forge_version: String,
    pub forge_version_short: String,
    /// The server run's own identity, minted once per process.
    ///
    /// **An ask's number restarts with the process**, so a client that kept
    /// its marks across a forge restart would read a fresh ask as one it had
    /// already announced. The run rides the roster and every mark is scoped
    /// by it, so no two runs can be confused.
    pub boot_id: String,
    pub service_status: Option<Value>,
    /// The last fatal error, held by the core: an App-level event with no
    /// state behind it reaches only whoever was subscribed when it fired.
    pub fatal_error: Option<Value>,
    /// The device enrolments the settings page lists and the unlock card
    /// reads. **No wrapped blob is here**: the list is read by every client,
    /// and the bytes stay behind a release.
    pub biometric: ReadWire<Vec<forge_primitives::biometric::BiometricEnrolment>>,
    /// The shared vault session's lock state and idle bound, for the
    /// settings vault card. **No token or key is here**: the card draws
    /// whether the vault is open and when it locks itself.
    pub vault: Option<VaultWire>,
}

/// The shared vault session as the settings card draws it.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct VaultWire {
    /// Whether the session holds its keys open right now.
    pub locked: bool,
    /// The idle bound the config set, or `null` for no idle lock.
    pub idle_minutes: Option<u64>,
}

/// The Extensions page's read: the plugin inventory, the PATH checks the
/// scan cannot carry, and the last update run.
///
/// **`snapshot` and `update_run` are the same shapes the live frames
/// carry.** `PluginsInventoryUpdated`, `PluginsUpdateRunProgress` and
/// `PluginsUpdateRunFinished` each embed these objects, so a client that
/// takes one off the stream and a client that read the snapshot are
/// drawing the same bytes (rule 25: the read path and the live path are
/// one obligation). `lsp` is read-only - no frame carries it, and a page
/// re-reads to refresh it.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ExtensionsWire {
    /// `None` when no plugins root resolves (`CLAUDE_CONFIG_DIR` or
    /// `HOME`): the inventory is unknowable, which is its own state rather
    /// than an empty one.
    pub snapshot: Option<PluginsInventorySnapshot>,
    pub update_run: Option<PluginUpdateRun>,
    pub lsp: Vec<LspCheckWire>,
}

/// One declared LSP server and whether its binary resolves on PATH.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct LspCheckWire {
    pub plugin: String,
    pub server: String,
    pub command: String,
    pub on_path: bool,
}

/// One project row: the project, and the per-row reads the home draws it from.
///
/// The cells beside the project are what the row states, and each is reached
/// from the ROW rather than from a seat: a home row is a project, and a
/// project is a row whether or not anything has started it. The board's own
/// detail - rows, schedules and connector sets - is [`ProjectBoardWire`]'s
/// (#2517).
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ProjectWire {
    pub project: ProjectView,
    /// The branch and the count for the project's own tree.
    ///
    /// **Keyed by the project's LEAD slot, which exists for every declared
    /// project** whether or not a session has opened it, and read at the
    /// project's own path - so a row's cell is filled for a project nobody
    /// has started, which is the whole point of the cell. That is also why it
    /// is not the read a SEAT draws: a worker's own tree is `AgentWire::work`,
    /// and this one is the project's own path whichever seat is asking.
    ///
    /// This is the shared cache the transport holds, so several clients
    /// subscribing do not multiply git invocations: the cache is what makes
    /// the per-row cost one scan's rather than N times it.
    pub work: WorkState,
    /// Whether a spawn in this project would find an account. Read beside
    /// `project.has_model`, which is what tells the two reasons a spawn cannot
    /// run apart.
    pub would_bind: bool,
    /// The account this project's row chips, and its state.
    ///
    /// A view draws the chip from `Roster::chip_for`, which derives the
    /// state from the project's model and the account pool. `has_model` is not
    /// enough to derive it from, so a client could draw the row and not the
    /// chip it carries.
    pub chip: Option<forge_workspace::SessionChipInfo>,
}

/// The board's own read: every project's live detail, for the page and the
/// rail that draw it (#2517).
///
/// **Its own subject and its own snapshot.** The rows, schedules and
/// connector sets are most of what a ~708 KB home read used to carry, and a
/// home read happens on every update that covers the fleet; a page that draws
/// none of them should not pay for them.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct BoardWire {
    pub projects: Vec<ProjectBoardWire>,
}

/// One project's identity, which is all a board row needs to place it.
///
/// The rest of [`ProjectView`] - its sessions, its accounts, its paths - is
/// the home's: the board neither hears nor draws it (#2517).
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ProjectBoardIdentity {
    pub org: String,
    pub name: String,
}

/// One project's board detail.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ProjectBoardWire {
    pub project: ProjectBoardIdentity,
    /// The project's live rows, with what the board derives on each:
    /// worked time, ages, and the marks. Computed once, in the workspace's
    /// board module, and passed through - the client never re-derives a
    /// mark differently.
    pub rows: Vec<forge_workspace::board::BoardRow>,
    /// The schedules this project holds.
    ///
    /// A view draws its SCHEDULES section from these, and nothing on
    /// this wire carried them: a client could create a cron and never see it
    /// again.
    pub crons: Vec<forge_primitives::CronEntry>,
    /// What this project's sessions are subscribed to, one list per connector.
    ///
    /// The subscriptions are per PROJECT and the section that draws them sits
    /// on the row, so they ride here rather than on the home's
    /// [`ConnectorsWire`] - which carries the liveness facts, which belong to
    /// the server and the workspace rather than to a project.
    pub connectors: ProjectConnectorsWire,
}

/// One seat's row: the surface's own [`AgentRow`], plus the read that is per
/// SEAT rather than per project.
///
/// A home row is a seat, and its `where` cell is that seat's own working tree.
/// `ProjectWire::work` cannot answer it - that read is the project's own path
/// from the lead's seat - so a worker row drew the project's branch, and on a
/// fleet with workers in worktrees that is eight rows reading `main`.
///
/// The fields are spelled out rather than the row being carried whole, so a
/// field added to [`AgentRow`] has to be carried here deliberately rather than
/// reaching a client by accident.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct AgentWire {
    pub slot: SessionSlot,
    pub label: String,
    pub lifecycle: forge_primitives::SessionLifecycleState,
    /// Which heal attempt is next while the seat is `Restarting`; `None`
    /// for every other lifecycle.
    pub heal_attempt: Option<u32>,
    /// How many restarts a heal may make, so the row can say "1 of 3".
    pub heal_attempts: Option<u32>,
    pub has_background_work: bool,
    pub pending: Option<forge_workspace::PendingInteractionKind>,
    pub pending_depth: usize,
    /// The ask's identity: how many asks have ever arrived for the seat.
    /// A client announces a new ask by this number, so a re-read is not news
    /// and an ask replacing another is.
    pub pending_seq: u64,
    pub last_activity: Option<std::time::SystemTime>,
    pub reason: Option<String>,
    /// When the seat's newest turn ended in failure, filtered by what has
    /// been shown: `None` once the seat has been shown since the failure,
    /// or while it is being shown now.
    pub failed_turn: Option<std::time::SystemTime>,
    /// Whether a session on the seat ended and no reader has been shown it
    /// since: the row's failure is the durable trace of a crash nobody
    /// watched.
    pub died: bool,
    /// The seat's own tree as a NAME, home-shortened, when the worker runs
    /// somewhere other than its project's directory: the fact the row's
    /// where cell says. `None` for a seat in its project's own tree.
    pub tree: Option<String>,
    /// The same tree's STATE - branch, changed count, gate - `None` for a
    /// seat forge holds no directory for.
    ///
    /// Read through the same shared cache the project rows use, so a lead's
    /// tree is one read drawn twice rather than two reads.
    pub work: Option<WorkState>,
}

impl From<&AgentRow> for AgentWire {
    fn from(row: &AgentRow) -> Self {
        Self {
            slot: row.slot.clone(),
            label: row.label.clone(),
            lifecycle: row.lifecycle,
            heal_attempt: row.heal_attempt,
            heal_attempts: row.heal_attempts,
            has_background_work: row.has_background_work,
            pending: row.pending,
            pending_depth: row.pending_depth,
            pending_seq: row.pending_seq,
            last_activity: row.last_activity,
            reason: row.reason.clone(),
            failed_turn: row.failed_turn,
            died: row.died,
            tree: row.tree.clone(),
            work: None,
        }
    }
}

/// What the inbound connectors are watching, server-wide.
///
/// The liveness facts and nothing else: what each connector is subscribed to
/// belongs to a project and rides [`ProjectBoardWire::connectors`], so a
/// subscriptions field here would be one no read can fill - and a page drawing
/// it shows the section empty however long it waits.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ConnectorsWire {
    pub gotify: GotifyWire,
    pub slack: SlackWire,
}

#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct GotifyWire {
    pub connected: bool,
}

#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct SlackWire {
    /// Workspace label to whether that workspace's pump is live, as a list
    /// of pairs: JSON objects with dynamic keys are awkward for a client to
    /// iterate, and the order is the store's.
    pub connected_workspaces: Vec<(String, bool)>,
    pub load_failed: bool,
}

/// One project's connector subscriptions, as the row's section draws them.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ProjectConnectorsWire {
    pub gotify: Vec<GotifySubscription>,
    pub slack: Vec<SlackSubscription>,
}

/// Dictation's preflight state.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct DictateWire {
    /// Whether `[dictate] enabled` is set. **Carried rather than inferred
    /// from an empty `models` list**: the list is empty both for a
    /// switched-off section and for a default snapshot, so a reader asserting
    /// the cause from the value tells a healthy configuration it is off, and
    /// nothing in the payload says which.
    pub enabled: bool,
    pub snapshot: Value,
    pub models_dir: Option<std::path::PathBuf>,
}

/// The voice switch and where its model lands, as the home carries them.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct VoiceWire {
    pub enabled: bool,
    pub models_dir: Option<std::path::PathBuf>,
}

/// One session, as a client sees it: every read a session-scoped view makes.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct SessionWire {
    pub slot: SessionSlot,
    pub state: SessionStateWire,
    /// What this seat's composer is doing, which no record answered before.
    pub composer: ComposerWire,
    pub header: SessionHeaderWire,
    pub mcp: Option<McpServers>,
    pub processes: Option<ProcessSnapshot>,
    /// The seat's live tasks, which the processes feed leads its rows with:
    /// the CLI's background registry plus the long foreground calls it is
    /// tracking while they run, each saying which it is (`is_backgrounded`).
    ///
    /// The tasks the OS walk cannot see for itself are the ones this exists
    /// for: a backgrounded bash is `setsid`-detached, so it sits outside
    /// claude's tree, and a view draws it from the registry - with the
    /// command telling it whether the walk has already adopted the process
    /// behind one. A foreground call is in the tree already; the row is what
    /// names it while it runs.
    pub background_tasks: Vec<forge_workspace::BackgroundTask>,
    pub monitors: Vec<MonitorRecord>,
    /// The seat's terminal, when it has one: the strip's row and the state
    /// it draws. Absent until the terminal is opened; the strip draws its
    /// row for every seat and reads this for the state, so a seat with none
    /// draws **off** and **start**.
    pub terminal: Option<forge_primitives::terminal::TerminalSummary>,
    /// The prompt a one-ask dock draws: the front of `pending_asks`.
    ///
    /// **Derived from the same list at serialisation**, so the two cannot
    /// drift. What a client reading only this draws is the front, which on a
    /// mixed set is NOT what the single read answered before the list: that
    /// preferred a question over a permission prompt and took same-kind asks
    /// in map order, where the front is the oldest ask with a draft leading.
    pub pending_ask: Option<PendingAskWire>,
    /// Every prompt this seat is holding: **a draft leads, then arrival
    /// order**. A parallel batch parks several at once, so a client that
    /// attached mid-batch reads the ones behind the front rather than losing
    /// all but one.
    pub pending_asks: Vec<PendingAskWire>,
    /// The newest turns, not the whole conversation. See [`SUBSCRIBE_TURNS`].
    pub conversation: ConversationWire,
    pub slash_commands: Vec<AvailableCommand>,
    pub subagents: Vec<AvailableAgent>,
    /// The session's sub-agent instances, as the core joined them: one card
    /// per dispatch with its calls under it. The catalogue above names the
    /// TYPES the CLI offers; this is what actually ran.
    pub subagent_instances: Vec<forge_primitives::runtime::SubagentCard>,
    /// The seat's own walk, from the store its loop keeps fresh, or a walk
    /// taken for the read when the loop has not run. Shared rather than walked
    /// per subscriber: the walk is a whole tree. Serialises as the index
    /// itself.
    pub file_index: Arc<FileIndex>,
    /// The sessions this seat can jump back to with `/resume`, newest first.
    ///
    /// Carried on the record because the composer's other sources are: a
    /// `/resume` list is only asked for while a draft's argument is being
    /// typed, and a read on demand would need its own ask, its own answer and
    /// a shell command to carry it - the record already reaches every reader
    /// the composer draws on. Both a lead's and a worker's read is one scan of
    /// the directory that seat's own resume runs in.
    pub resume_sessions: Vec<ResumableSessionWire>,
    pub reviews: ReviewsWire,
    /// The seat-tree review half, beside the branch's: the files page's
    /// comments, each thread placed against the seat's working files. Same
    /// wire shape as `reviews` so a page parses one thing; the scope is
    /// the seat, not a branch, which is what the files page keys to.
    pub tree_reviews: ReviewsWire,
    /// The working tree behind the git section: the branch, how much has
    /// changed, and the gate. The changed files themselves ride `diff`
    /// below.
    pub work: WorkState,
    /// The tree behind the git row's depth: the uncommitted files with
    /// their marks and counts, the branch's chain ahead of its default
    /// with the commits that produced it, and which branch that is.
    ///
    /// What the row above cannot say: a branch and a count tell a reader
    /// nothing about WHAT is working, and this is what a hover over the
    /// row draws. The content read below carries the same layers with
    /// hunks, for the surface that reviews them.
    pub git: forge_primitives::git_diff::GitWorkView,
    /// The open pull request this seat's branch is on, and the issues it
    /// closes.
    ///
    /// Beside `work` rather than inside it because the two come from
    /// different reads - the branch and the count are the row's own, and this
    /// is the heavier scan - and a client draws the inspector's GIT section
    /// from both. Nothing carried them, so a client would draw the section
    /// without the `PR #N -> closes #M` row a view leads it with.
    pub pr: Option<forge_primitives::git::GitPrInfo>,
    pub closes: Vec<forge_primitives::git::GitIssueRef>,
    /// The changed files with their raw hunks, bounded and flagged: what a
    /// review surface draws a branch from, as data and never as a
    /// rendering.
    ///
    /// Read beside `work` rather than pushed with it: the row above moves
    /// with the tree, while this is taken once per read that encodes a
    /// record - a cold load, a reconnect, a seat swap - the way a view's
    /// `/diff` scans once on open. Its caps and flags live in
    /// forge-agent's `git_diff::content`, which is where the read's shape
    /// is decided.
    pub diff: DiffContent,
}

/// One session a seat can resume, as `/resume`'s list draws it: the id a pick
/// writes into the draft, and the transcript's own title beside it.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ResumableSessionWire {
    pub session_id: String,
    pub summary: String,
}

/// Where a session's reads find their own working tree.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct SessionStateWire {
    pub slot: SessionSlot,
    pub scan_cwd: std::path::PathBuf,
    /// The prompts still waiting in the CLI's queue, oldest first.
    ///
    /// The read path for the pile: `prompt_queued` and `prompt_lifecycle`
    /// speak only on a change, so a client attaching mid-queue - a fresh
    /// load, a refresh, a seat switch - has nothing else to draw the waiting
    /// prompts from, and two attached clients have to agree about them.
    pub queue: Vec<forge_workspace::protocol::QueuedPrompt>,
    /// Clients with voice on for this seat, this one included.
    pub voice_listeners: u32,
}

/// What one seat's composer is doing.
///
/// A client attaching to a running session cannot reconstruct any of it: the
/// stream announces a compaction and a sign-in once each and retains nothing.
///
/// **No take, and the notice only when it is unsent**: a take belongs to the
/// connection that started it, whose own updates carry its meter, its phases
/// and its words, so the shared record says nothing about one and no other
/// subscriber draws it. A LANDED take's words, though, are the seat's until a
/// prompt carries them away - a page that reloads between the landing and the
/// send reads them back here (#1360) - and they are marked `carried` so a
/// page reads a re-hand as the seat's words rather than a landing of its own.
///
/// The ASKS the composer is answering ride `pending_asks` on this same record
/// rather than appearing here, because that is the same thing the session's own
/// read answers and two copies of it would drift.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ComposerWire {
    /// The machine turn running on the seat, which takes the composer away:
    /// absent for a seat taking input.
    pub machine: Option<forge_primitives::runtime::MachineTurn>,
    /// The push-to-talk key and how a press maps onto a take, in the
    /// vocabulary `forge.toml` accepts.
    ///
    /// **Carried rather than left to each client's default.** What a
    /// keyboard press means is the user's own configuration, and a client
    /// that drew the affordance from a hardcoded key would honour a
    /// different chord on every install that moved it.
    pub bind: String,
    pub mode: String,
    /// The landed dictation nobody has sent yet, in the same shape the
    /// `dictate_ended` update carries, plus `carried: true`.
    pub notice: Option<serde_json::Value>,
}

/// The session's header facts, including the two a client cannot otherwise
/// reach: the catalogue a picker draws its rows from, and whether a turn is
/// in flight.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct SessionHeaderWire {
    /// The occupant's id, or `None` when there is no occupant to name:
    /// nothing started, nothing connected yet, or an id dropped since.
    ///
    /// **This is the epoch [`Self::seq`] belongs to.** Every frame the seat's
    /// stream carries is numbered, and a change of occupant starts the count
    /// over at 1 - so a client that reads this id with the number below knows
    /// which frames it missed and which it never had.
    pub session_id: Option<forge_primitives::SessionId>,
    /// The number the seat's next frame will carry.
    ///
    /// A client that attaches mid-stream reads where the stream stands, so
    /// the frames it hears from here continue the count it holds rather than
    /// looking like a gap. It is the successor - never 0, never the last
    /// number minted.
    pub seq: u64,
    pub model: Option<Value>,
    pub effort: Value,
    pub permission_mode: Option<Value>,
    pub context: Value,
    pub available_models: Vec<Value>,
    pub turn_in_flight: bool,
}

/// What a seat is held on: every kind of parked interaction, not the two the
/// dock happened to draw first. Each request crosses as the core's own shape,
/// because a client draws it rather than re-deriving it.
#[derive(Serialize, Deserialize)]
#[serde(tag = "kind", content = "request", rename_all = "snake_case")]
pub enum PendingAskWire {
    Permission(Value),
    Question(Value),
    SlackDraft(Value),
    BrowserHandOff(Value),
    TerminalHandOff(Value),
    DocumentHandOff(Value),
    SettingsWrite(Value),
    Vault(Value),
}

impl From<&PendingAsk> for PendingAskWire {
    fn from(ask: &PendingAsk) -> Self {
        let encode = |body: Result<Value, serde_json::Error>| body.unwrap_or(Value::Null);
        match ask {
            PendingAsk::Permission(request) => {
                Self::Permission(encode(serde_json::to_value(request.as_ref())))
            }
            PendingAsk::Question(request) => {
                Self::Question(encode(serde_json::to_value(request.as_ref())))
            }
            PendingAsk::SlackDraft(draft) => {
                Self::SlackDraft(encode(serde_json::to_value(draft.as_ref())))
            }
            PendingAsk::BrowserHandOff(handoff) => {
                Self::BrowserHandOff(encode(serde_json::to_value(handoff.as_ref())))
            }
            PendingAsk::TerminalHandOff(handoff) => {
                Self::TerminalHandOff(encode(serde_json::to_value(handoff.as_ref())))
            }
            PendingAsk::DocumentHandOff(ask) => {
                Self::DocumentHandOff(encode(serde_json::to_value(ask.as_ref())))
            }
            PendingAsk::SettingsWrite(ask) => {
                Self::SettingsWrite(encode(serde_json::to_value(ask.as_ref())))
            }
            PendingAsk::Vault(draft) => Self::Vault(encode(serde_json::to_value(draft.as_ref()))),
        }
    }
}

/// The conversation a session's transcript replayed, and how many times it
/// has compacted.
///
/// **Its turns, not a bare frame list.** A client folds turns, so a
/// conversation handed over as frames with no boundaries is one it cannot fold
/// at all; carrying the same turns a page carries is what makes the whole
/// conversation and a window of it read the same way.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ConversationWire {
    pub turns: Vec<TurnWire>,
    /// The handle above the newest page, the same one a `more` answers with.
    ///
    /// A snapshot carries the page a `more` would fetch, cursor included, so
    /// a client seeds its conversation from this answer and pages above it
    /// from here rather than asking for the same page twice (#2420). `None`
    /// is the conversation's first page: there is nothing above it.
    pub cursor: Option<String>,
    pub compaction_count: u32,
    /// The schedules the fired prompts in these turns came from, keyed by the
    /// prompt's own id.
    ///
    /// **The read half of the pairing `CronPromptAppended` carries once.**
    /// The frame titles the row while it lands, and the row replayed from the
    /// transcript afterwards carries the prompt alone - so the pairing rides
    /// the read too, for the rows this page holds and no others.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub cron_names: BTreeMap<String, forge_primitives::cron::CronName>,
}

/// The schedules the fired prompts in `turns` came from, keyed by the
/// prompt's own id.
///
/// Only a delivery row is looked up: its prompt id is the pairing's key, and
/// the map holds nothing for a row no schedule produced.
pub fn cron_names_of(
    surface: &ViewSurface,
    slot: &SessionSlot,
    turns: &[TurnWire],
) -> BTreeMap<String, forge_primitives::cron::CronName> {
    let mut uuids: Vec<&str> = Vec::new();
    for turn in turns {
        for frame in &turn.messages {
            if frame.get("type").and_then(Value::as_str) != Some("user") {
                continue;
            }
            let Some(uuid) = frame.get("uuid").and_then(Value::as_str) else {
                continue;
            };
            if !uuid.is_empty() && !uuids.contains(&uuid) {
                uuids.push(uuid);
            }
        }
    }
    if uuids.is_empty() {
        return BTreeMap::new();
    }
    surface.cron_names(slot, &uuids)
}

/// The review reads for one branch, each carrying its own answer.
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ReviewsWire {
    pub threads: ReadWire<Vec<ReviewedThreadWire>>,
    pub reviews: ReadWire<Vec<ReviewSet>>,
}

/// One file's content at a revision, as the `file_content` answer carries
/// it: the lines a review page's between-hunk expansion draws.
///
/// A read that answered something is never folded into one that failed -
/// the same distinction the diff reader keeps - so a refusal crosses as its
/// own arm with its own words rather than as an empty read.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum FileContentWire {
    /// The lines the read carried, and whether its ceiling cut them short.
    Read { lines: Vec<String>, truncated: bool },
    /// The read could not run, in its own sentence.
    Failed { why: String },
}

/// One review thread as a record carries it: the stored thread, and where
/// its anchor sits in the content the record ships beside it.
///
/// **The landing is resolved here because only the server can resolve
/// it.** A client draws the bounded content it is sent and has no working
/// tree to re-scan; the server's own resolver is the one answer to
/// whether an anchor still lands, so the record carries that answer
/// rather than a second implementation deriving one a language away.
///
/// **A thread written inside a commit lands too, when the content
/// carries it.** Its own commit's view draws it at its recorded line
/// without any of this; the landing is what lets the all-changes view
/// draw it beside the same code in the branch's numbering - earned by
/// content and neighbours, never by the commit-numbered line, which
/// addresses unrelated code here.
///
/// `resolution` is `None` when this read has no landing to report: a
/// thread whose branch layer the scan could not read, or one
/// numbered against another base.
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct ReviewedThreadWire {
    #[serde(flatten)]
    pub thread: ReviewThread,
    /// Where the anchor sits in the content this answer carries, or `None`
    /// when the read has no lines to place against - a document whose bytes
    /// are not text.
    pub resolution: Option<ThreadResolutionWire>,
}

/// Where an anchor sits, as a view draws it: the line in the anchor's own
/// side's numbering, or why it could not be placed.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum ThreadResolutionWire {
    /// Placed, at `line`. `moved_from` is the recorded line when the
    /// anchor no longer sits where it was written.
    Placed { line: u32, moved_from: Option<u32> },
    /// Not placed, for `reason`.
    Outdated { reason: OutdatedReasonWire },
}

/// The resolver's own reason, crossing as data rather than as prose.
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OutdatedReasonWire {
    Gone,
    Ambiguous { matches: usize },
}

/// Every thread with its landing against `content` - the same content the
/// record ships, so a page draws a thread where this answer says.
///
/// **Only a thread numbered against this diff's own base is placed.** A
/// whole-diff anchor records the base it was written against, and a
/// view's overlay opened against another target writes line numbers that
/// address unrelated code here (the placement refuses exactly that:
/// "line numbers against another base would anchor onto unrelated code"). A
/// thread from another base answers `None` - no landing - which the page
/// draws as itself rather than parking it on a guess.
fn reviewed_threads(
    threads: Vec<ReviewThread>,
    content: &DiffContent,
    default_branch: Option<&str>,
    whole: &WholeFiles,
) -> Vec<ReviewedThreadWire> {
    let files = match &content.branch_ahead {
        LayerState::Populated(files) => Some(files.as_slice()),
        LayerState::Clean | LayerState::ScanFailed => None,
    };
    threads
        .into_iter()
        .map(|thread| {
            let home_base = default_branch == Some(thread.anchor.base_ref.as_str());
            let resolution = match (&thread.commit, files) {
                (_, Some(files)) if home_base => {
                    // **A commit-scoped anchor's line counts in its own
                    // commit's numbering**, so the in-place shortcut is off
                    // for it: the landing has to be earned by content and
                    // neighbours (the resolver's own `trust_line_number`),
                    // which is what keeps the line from anchoring onto
                    // unrelated code here.
                    let trusted = thread.commit.is_none();
                    resolution_wire(&thread.anchor, files, trusted, trusted.then_some(whole))
                }
                _ => None,
            };
            ReviewedThreadWire { thread, resolution }
        })
        .collect()
}

/// One whole file an anchor needed, and whether the read carried all of it.
struct WholeFile {
    lines: Vec<String>,
    truncated: bool,
}

/// The whole files an anchor needed, one per `(path, side)`.
type WholeFiles = std::collections::HashMap<(String, ReviewSide), WholeFile>;

/// At most this many whole-file reads per record encode. An ordinary record
/// reads none - every anchor sits in its hunks - and a review with more
/// comments outside its hunks than this answers the rest from the bounded
/// content rather than running an unbounded pile of git subprocesses on
/// every seat encode.
const WHOLE_FILE_READS_PER_RECORD: usize = 8;

/// Whether any hunk of `files` carries `anchor.line` on the anchor's side.
///
/// **This is the rule that decides which authority answers**: a line the
/// hunks carry is placed there, and a line they do not carry is answered by
/// the whole file alone - never by a content match to some other hunk line
/// that happens to read the same.
fn line_carried(anchor: &ReviewAnchor, files: &[FileContent]) -> bool {
    files.iter().filter(|file| file.path == anchor.path).any(|file| {
        file.hunks.iter().any(|hunk| {
            hunk.lines.iter().any(|line| match anchor.side {
                ReviewSide::New => line.new_line == Some(anchor.line),
                ReviewSide::Old => line.old_line == Some(anchor.line),
            })
        })
    })
}

/// The whole files for the anchors no hunk carries a line for.
///
/// **A comment written on a revealed context line is outside the diff's
/// hunks by construction**, and the file it was written on is the whole
/// answer. The new side is numbered at the branch tip and the old side at
/// the merge-base, because the branch-ahead diff is a three-dot diff; an old
/// side read at the base's current tip would shift every line when the base
/// has advanced. A renamed file's old side is read at its old path. Only a
/// thread this record's own diff can draw is read for: another base's
/// thread, or a commit-scoped one whose line counts in its own commit, keeps
/// the resolver's answer. A read that fails leaves the anchor to the bounded
/// content, as before.
async fn whole_files_for(
    cwd: &std::path::Path,
    ceiling: u64,
    threads: &[ReviewThread],
    content: &DiffContent,
    default_branch: Option<&str>,
    // The revision the old side resolves against: the sha the diff was pinned
    // to when the read refreshed its base, else the ref's name. A pinned diff
    // read at the name's merge-base would place anchors against a base the
    // diff does not describe.
    base_rev: Option<&str>,
) -> WholeFiles {
    let files = match &content.branch_ahead {
        LayerState::Populated(files) => files.as_slice(),
        LayerState::Clean | LayerState::ScanFailed => return WholeFiles::new(),
    };
    let mut whole = WholeFiles::new();
    // Resolved once per encode, and only if an old-side anchor asks for it.
    // **The ATTEMPT is remembered, not only its success**: a base whose
    // merge-base cannot be resolved must not cost one resolve per anchor,
    // and the read bound below never sees those attempts.
    let mut old_rev: Option<Option<String>> = None;
    let mut reads = 0_usize;
    for thread in threads {
        let anchor = &thread.anchor;
        if thread.commit.is_some() || default_branch != Some(anchor.base_ref.as_str()) {
            continue;
        }
        let Some(file) = files.iter().find(|file| file.path == anchor.path) else {
            continue;
        };
        if line_carried(anchor, files) {
            continue;
        }
        let key = (anchor.path.clone(), anchor.side);
        if whole.contains_key(&key) || reads >= WHOLE_FILE_READS_PER_RECORD {
            continue;
        }
        let (rev, path) = match anchor.side {
            ReviewSide::New => ("HEAD".to_owned(), file.path.clone()),
            ReviewSide::Old => {
                let Some(base) = base_rev.or(default_branch) else {
                    continue;
                };
                if old_rev.is_none() {
                    old_rev = Some(merge_base(cwd, base).await);
                }
                let Some(found) = old_rev.as_ref().and_then(|held| held.as_deref()) else {
                    continue;
                };
                (found.to_owned(), file.old_path.clone().unwrap_or_else(|| file.path.clone()))
            }
        };
        reads += 1;
        if let Ok(read) = read_file_lines(cwd, &rev, &path, ceiling).await {
            whole.insert(key, WholeFile { lines: read.lines, truncated: read.truncated });
        }
    }
    whole
}

/// One anchor placed against the bounded content the record carries. With
/// `trusted` the line number is evidence too, because the branch-ahead
/// layer is the numbering the anchor was written in - the same trust a
/// view's overlay extends to its own home scope. Untrusted, the placement
/// is earned by content and neighbours alone.
///
/// `whole` is the file held for a line the hunks do not carry, and `None`
/// for a commit-scoped thread, whose numbering belongs to its own commit and
/// must not borrow the branch's read. A read cut before the anchored line
/// cannot answer, and answers no landing rather than `Gone`.
fn resolution_wire(
    anchor: &ReviewAnchor,
    files: &[FileContent],
    trusted: bool,
    whole: Option<&WholeFiles>,
) -> Option<ThreadResolutionWire> {
    let held = whole.and_then(|map| map.get(&(anchor.path.clone(), anchor.side)));
    let placed = if line_carried(anchor, files) {
        place_anchor_in_content(anchor, files, trusted)
    } else {
        match held {
            Some(file) => {
                let past_the_cut =
                    usize::try_from(anchor.line).is_ok_and(|line| line > file.lines.len());
                if file.truncated && past_the_cut {
                    return None;
                }
                place_anchor_in_lines_with(anchor, &file.lines, trusted)
            }
            // The whole file is the authority for a line the hunks do not
            // carry; without the read, the bounded content is the fallback.
            None => place_anchor_in_content(anchor, files, trusted),
        }
    };
    Some(match placed {
        AnchorPlacement::Placed { line, moved_from } => ThreadResolutionWire::Placed {
            line,
            // **A cross-scope landing names no `moved_from`**: the recorded
            // number counts in another scope's numbering, and naming it
            // beside a line of this one would be a number the reader cannot
            // place.
            moved_from: if trusted { moved_from } else { None },
        },
        AnchorPlacement::Outdated(reason) => ThreadResolutionWire::Outdated {
            reason: match reason {
                OutdatedReason::Gone => OutdatedReasonWire::Gone,
                OutdatedReason::Ambiguous { matches } => OutdatedReasonWire::Ambiguous { matches },
            },
        },
    })
}

/// [`resolution_wire`]'s twin for the seat tree: the placement the surface
/// already resolved against the seat's own working file crosses as-is -
/// the numbering is the anchor's home, so `moved_from` is always named
/// when there is one.
pub(crate) fn placement_wire(placement: AnchorPlacement) -> ThreadResolutionWire {
    match placement {
        AnchorPlacement::Placed { line, moved_from } => {
            ThreadResolutionWire::Placed { line, moved_from }
        }
        AnchorPlacement::Outdated(reason) => ThreadResolutionWire::Outdated {
            reason: match reason {
                OutdatedReason::Gone => OutdatedReasonWire::Gone,
                OutdatedReason::Ambiguous { matches } => OutdatedReasonWire::Ambiguous { matches },
            },
        },
    }
}

/// One turn, as a page carries it: the key the server named it by, and the
/// messages it ran as.
///
/// **The messages, not a drawing of them.** How a run of tool calls groups
/// inside a turn is a drawing decision, so it belongs to whoever draws; what
/// crosses is the turn's own boundary, and that stays here because the paging
/// contract is built on it - `more` asks for turns, and a page above the
/// window's floor
/// cuts on them rather than leaving a client to stitch half a turn to half.
///
/// **Below that floor a page can carry a stretch inside a turn**, which is
/// what a running turn's older rows are: a session whose life is one unended
/// turn has no boundary to cut on, and its rows reach a late-attaching view
/// only in runs. What a client is owed there is the pair: a page never opens
/// on a tool result without the call that produced it.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct TurnWire {
    /// `None` when nothing named the turn: the key comes from a `Result`
    /// frame, and a conversation read from a transcript carries none.
    pub key: Option<String>,
    /// The frames this turn ran as, which is the same shape the session
    /// record's `conversation` carries.
    pub messages: Vec<Value>,
}

/// One page of history: the newest turns, in conversation order, and the
/// handle that asks for the ones above them.
pub struct Page {
    pub turns: Vec<TurnWire>,
    /// `None` means there is nothing above this page, which is the one case
    /// a client stops asking.
    pub cursor: Option<String>,
}

/// Where each turn opens, one entry per turn, and what names it.
///
/// **One message opens one turn**, so a user row carrying two blocks does not
/// put two boundaries at one index. A cursor names a message and resolves to
/// the first match, so two entries sharing an index would leave a walk a page
/// of its own to hand back before it moved - an empty one, since both
/// boundaries are that message.
fn opens_of(spans: &[TurnSpan]) -> Vec<(usize, Option<&str>)> {
    let mut opens: Vec<(usize, Option<&str>)> = Vec::with_capacity(spans.len());
    for span in spans {
        match opens.last_mut() {
            Some((at, key)) if *at == span.opens_at => {
                if key.is_none() {
                    *key = span.key.as_deref();
                }
            }
            _ => opens.push((span.opens_at, span.key.as_deref())),
        }
    }
    opens
}

/// The messages `from..to` as one turn, which is the shape both the session
/// record and a page carry.
///
/// **A task ending a call in this turn is owed rides along.** A client folds
/// one turn at a time, so an ending the CLI persisted in another turn is one
/// it cannot reach - and the notice opens a turn of its own, which puts it in
/// another turn whenever anything that does open one sits between. The row
/// the CLI wrote stays where it is and draws nothing; what is added is a frame
/// of its own, which is content of the wire's own type rather than a second
/// copy of a drawing.
fn turn_wire(
    messages: &[Message],
    from: usize,
    to: usize,
    key: Option<&str>,
    endings: &HashMap<String, TaskEnding>,
) -> TurnWire {
    let mut frames: Vec<Value> =
        messages[from..to].iter().map(|message| frame_of(message, message.session_id())).collect();
    for ending in owed_endings(messages, from, to, endings) {
        let forged = forged_ending(ending);
        frames.push(frame_of(&forged, forged.session_id()));
    }
    TurnWire { key: key.map(str::to_owned), messages: frames }
}

/// One message as the page carries it, or the notice that says it will not.
///
/// **A failed encode used to cross as `Value::Null`**, and the fold reads
/// `type` off every frame - so the page threw on exactly the frame it could
/// not describe. The notice names the failure and draws in its place: nothing
/// is dropped, and nothing arrives as a value no reader can read. The warning
/// names the session the frame belongs to, which is the identity a reader of
/// the log can act on.
fn frame_of<T: Serialize>(value: &T, session: Option<&str>) -> Value {
    match serde_json::to_value(value) {
        Ok(frame) => frame,
        Err(error) => {
            tracing::warn!(
                event_name = "conversation_frame_unsent",
                session = session.unwrap_or("unknown"),
                error = %error,
                "a conversation frame could not be serialised; drawn as a notice instead",
            );
            serde_json::json!({
                "type": "system",
                "subtype": "forge_notice",
                "severity": "warning",
                "text": format!("a conversation frame could not be serialised: {error}"),
            })
        }
    }
}

/// Fold each run of consecutive thinking-token counters in a turn into the one
/// frame it draws as.
///
/// The fold reads a counter for two things only, its growth and its stamp,
/// and draws no row for it - so a turn that thinks for an hour is thousands of
/// frames that all say the same running figure. One frame per run carries it
/// exactly: the summed `estimated_tokens_delta`, the newest frame's own id and
/// stamp, and the stamps the run covers (`folded_from` through its own
/// `timestamp`) with `folded_frames` saying how many the one stands for. **The
/// live stream already serves a flush's run this way** (`batch::merge`); this
/// is that shape at the page's own cut.
///
/// **The range is what a landing reads, not a single instant.** A page can
/// open on a turn's SUFFIX - the held window's front cuts inside a running
/// turn when no turn opener sits above the line - so a client holding the
/// counters above the page's first one must keep them, and a run that recorded
/// only where it ends would drop them.
///
/// **A run that may start mid-turn is never folded.** When a turn's first
/// carried frame is itself a counter, the window's front cut inside that
/// running turn, and the run's true start is not in the page - so its leading
/// counters are served as themselves. Every fold a page carries therefore
/// starts at the first counter after a frame that is no counter, which is a
/// boundary any earlier page of the same run shares; a later page's fold then
/// nests inside or around an earlier one instead of crossing it.
///
/// **A run is one dispatch's, and the dispatch lives in the frame's extras
/// rather than in its typed fields.** No captured baseline carries
/// `parent_tool_use_id` on a counter, so the grouping is a guard rather than a
/// live path: if the CLI ever stamps one, a sub-agent's counters are the
/// subagents surface's and the chat fold skips them whole, and folding them
/// into the parent's run would move its figure and its clock. **A folded run
/// is stamped throughout**: a frame with no stamp breaks the run rather than
/// being folded under a bound it cannot be placed by.
///
/// It runs after the page is cut, so the budget still counts the bytes a page
/// used to carry and keeps the same turns it always did.
fn compact_counters(frames: &mut Vec<Value>) {
    let held = std::mem::take(frames);
    let mut out: Vec<Value> = Vec::with_capacity(held.len());
    // The turn's first frame opens it unless the page cut the turn, and only
    // a proven opening lets the leading run fold.
    let mut leading = held.first().is_some_and(is_counter);
    for frame in held {
        if !is_counter(&frame) {
            leading = false;
        }
        if let Some(last) = out.last_mut()
            && !leading
            && is_counter(last)
            && is_counter(&frame)
            && same_run(last, &frame)
            && counter_stamp(last).is_some()
            && counter_stamp(&frame).is_some()
        {
            // The newest frame survives with its own id, session and extras,
            // the same exchange `batch::merge` makes; only the growth, the
            // count and the range are the run's. The frame being replaced
            // carries the range's start once it is itself a fold, and before
            // that it is the run's opening frame.
            let previous = std::mem::replace(last, frame);
            let delta = grown(last).saturating_add(grown(&previous));
            let count = counted(&previous).saturating_add(1);
            let from = previous.get("folded_from").or_else(|| counter_stamp(&previous)).cloned();
            if let Some(object) = last.as_object_mut() {
                if let Some(from) = from {
                    object.insert("folded_from".to_owned(), from);
                }
                object.insert("estimated_tokens_delta".to_owned(), Value::from(delta));
                object.insert("folded_frames".to_owned(), Value::from(count));
            }
            continue;
        }
        out.push(frame);
    }
    *frames = out;
}

/// Whether the frame is a thinking-token counter.
fn is_counter(frame: &Value) -> bool {
    frame["type"] == "system" && frame["subtype"] == "thinking_tokens"
}

/// The dispatch a counter ran under, empty when it is the turn's own.
fn counter_dispatch(frame: &Value) -> Option<&str> {
    frame.get("parent_tool_use_id").and_then(Value::as_str).filter(|id| !id.trim().is_empty())
}

/// Whether `next` continues the counter run `last` is part of.
fn same_run(last: &Value, next: &Value) -> bool {
    counter_dispatch(last) == counter_dispatch(next)
}

/// The stamp a counter frame carries, when it carries one.
fn counter_stamp(frame: &Value) -> Option<&Value> {
    frame.get("timestamp").filter(|stamp| stamp.is_string())
}

/// The growth one counter frame stands for: one frame's delta, or a run's sum.
fn grown(frame: &Value) -> i64 {
    frame["estimated_tokens_delta"].as_i64().unwrap_or(0)
}

/// How many frames a counter stands for, itself included.
fn counted(frame: &Value) -> u64 {
    frame["folded_frames"].as_u64().unwrap_or(1)
}

/// The keys a client actually draws out of a structured record: a mutation's
/// patch and its outside-edit mark, a question's answers and the notes typed
/// beside them.
const DRAWN_RECORD_KEYS: [&str; 4] = ["structuredPatch", "userModified", "answers", "annotations"];

/// What the cap did to one served record, for the log line beside it.
#[derive(Clone)]
pub(crate) struct RecordCap {
    /// The record's size before the cap saw it.
    pub bytes: usize,
    /// The keys the served record kept, empty when the marker replaced it.
    pub kept: String,
}

/// The record reduced to `keys`, or `None` when it holds none of them.
fn reduced_to(record: &Value, keys: &[&str]) -> Option<Value> {
    let held = record.as_object()?;
    let mut kept = serde_json::Map::new();
    for key in keys {
        if let Some(value) = held.get(*key) {
            kept.insert((*key).to_owned(), value.clone());
        }
    }
    (!kept.is_empty()).then_some(Value::Object(kept))
}

/// Whether a value serialises into the cap.
fn fits(record: &Value, cap: usize) -> bool {
    serde_json::to_string(record).is_ok_and(|text| text.len() <= cap)
}

/// Bound one served `tool_use_result` to `cap` bytes (#2374).
///
/// **The record is REDUCED before it is dropped, and the reduction is
/// staged**: the drawn keys together first, then the one key that carries the
/// row's own drawing - a question's `answers`, an edit's `structuredPatch`.
/// A question whose typed notes are what pushed it over keeps its answer and
/// loses only the notes, so it draws as the card it was, exactly as the live
/// frame does. The bare marker is the last resort; `userModified` rides it
/// because the row's outside-edit mark reads it.
///
/// `Some` says the record was over the cap, and what it kept, so the caller
/// can log what the served page left out.
pub(crate) fn cap_tool_record(record: &mut Value, cap: usize) -> Option<RecordCap> {
    if cap == usize::MAX {
        return None;
    }
    let Ok(text) = serde_json::to_string(&*record) else {
        return None;
    };
    if text.len() <= cap {
        return None;
    }
    let bytes = text.len();
    let stages: [&[&str]; 3] = [&DRAWN_RECORD_KEYS, &["answers"], &["structuredPatch"]];
    for keys in stages {
        let Some(reduced) = reduced_to(record, keys) else {
            continue;
        };
        if !fits(&reduced, cap) {
            continue;
        }
        let kept = reduced
            .as_object()
            .map(|held| held.keys().cloned().collect::<Vec<_>>().join(","))
            .unwrap_or_default();
        *record = reduced;
        return Some(RecordCap { bytes, kept });
    }
    let mut marker = serde_json::json!({ "omitted": true, "bytes": bytes });
    if record.get("userModified").and_then(Value::as_bool) == Some(true) {
        marker["userModified"] = Value::Bool(true);
    }
    *record = marker;
    Some(RecordCap { bytes, kept: String::new() })
}

/// Say what the cap did to an executed call's record (#2374).
///
/// Debug, because a bounded page is forge working as intended rather than a
/// problem of its own - and named by seat and call, so "where did this payload
/// go" is answered by one line rather than by a reproduction.
fn log_record_cap(seat: &SessionSlot, id: Option<&str>, cap: usize, report: &RecordCap) {
    tracing::debug!(
        event_name = "tool_record_capped",
        slot = %seat.display(),
        tool_use_id = id.unwrap_or(""),
        bytes = report.bytes,
        cap,
        kept = %report.kept,
        disposition = if report.kept.is_empty() { "omitted" } else { "reduced" },
        "a served page bounded an executed tool call's record; the call ran and the transcript \
         keeps the whole record, and [server] tool_record_cap_kb = 0 serves it whole",
    );
}

/// Bound every frame's `tool_use_result` in a page about to be SERVED (#2374).
///
/// **Only what a client is served is bounded; the transcript and the visible
/// `tool_result` block are untouched.** The structured record is drawn only
/// for mutations and questions - everything else is parsed and kept for
/// nothing - and one seat's mailbox records made a single page 70 MB, so the
/// page caps the record by `[server] tool_record_cap_kb` and leaves a marker
/// the client draws where it would have read the record. `usize::MAX` (the
/// `0` key) serves records whole.
pub(crate) fn cap_tool_records(turns: &mut [TurnWire], cap: usize, seat: &SessionSlot) {
    for turn in turns {
        for (id, report) in cap_turn(turn, cap) {
            log_record_cap(seat, id.as_deref(), cap, &report);
        }
    }
}

/// [`cap_tool_records`] for one turn, answering what it bounded and the call
/// each belongs to rather than logging it here.
///
/// **A caller that may not serve the turn must be able to drop the report**:
/// a first page cut to a byte budget bounds the records it MEASURES, and only
/// the turns it keeps are served, so only those are logged.
fn cap_turn(turn: &mut TurnWire, cap: usize) -> Vec<(Option<String>, RecordCap)> {
    if cap == usize::MAX {
        return Vec::new();
    }
    let mut reports = Vec::new();
    for frame in &mut turn.messages {
        // The call's own id, read off the result block that answers it, so
        // the line names the exact call a reader can find in the transcript.
        let id = frame
            .get("message")
            .and_then(|message| message.get("content"))
            .and_then(Value::as_array)
            .and_then(|blocks| {
                blocks.iter().find_map(|block| {
                    block.get("tool_use_id").and_then(Value::as_str).map(str::to_owned)
                })
            });
        if let Some(result) = frame.get_mut("tool_use_result")
            && let Some(report) = cap_tool_record(result, cap)
        {
            reports.push((id, report));
        }
    }
    reports
}

/// Hand the window's floor back as a page's cursor when the page opens on the
/// window's front (#2420).
///
/// **The window's own start is not the history's**: a page cut there carries
/// no handle of its own while the history above it begins at the floor. Both
/// a `more` answer and the session record serve that page, and both must hand
/// the same handle back or a client reads the history above as ended.
pub(crate) fn floor_cursor(page: &mut Page, dropped: usize) {
    if page.cursor.is_none() && dropped > 0 {
        page.cursor = Some(dropped.to_string());
    }
}

/// The same bound on a live frame's record, before it goes to a client
/// (#2374): a reload and a live frame draw the same record or the marker.
pub(crate) fn cap_message_record(msg: &mut Message, cap: usize, seat: &SessionSlot) {
    if let Message::User { tool_use_result: Some(record), message, .. } = msg {
        let id = message.content.iter().find_map(|block| match block {
            ContentBlock::ToolResult { tool_use_id, .. }
            | ContentBlock::ServerToolResult { tool_use_id, .. } => Some(tool_use_id.to_owned()),
            _ => None,
        });
        if let Some(report) = cap_tool_record(record, cap) {
            log_record_cap(seat, id.as_deref(), cap, &report);
        }
    }
}

/// The endings that name a call this turn holds and that the turn does not
/// already carry.
fn owed_endings<'a>(
    messages: &[Message],
    from: usize,
    to: usize,
    endings: &'a HashMap<String, TaskEnding>,
) -> Vec<&'a TaskEnding> {
    endings
        .values()
        .filter(|ending| !(from..to).contains(&ending.at))
        .filter(|ending| turn_holds_call(&messages[from..to], &ending.call))
        .collect()
}

/// Whether these frames make the call `id`.
fn turn_holds_call(messages: &[Message], id: &str) -> bool {
    messages.iter().any(|message| {
        let Message::Assistant { message: envelope, .. } = message else {
            return false;
        };
        envelope.content.iter().any(|block| {
            matches!(
                block,
                ContentBlock::ToolUse { id: call, .. }
                    | ContentBlock::ServerToolUse { id: call, .. }
                    if call == id
            )
        })
    })
}

/// The notice as a frame of its own, for the turn that holds the call it ends.
///
/// Forged rather than read off the wire, like the turn a delivery draws as:
/// the CLI wrote this ending into a row of its own, and this is that ending
/// delivered where a per-turn fold can see it. It carries no `uuid`, because
/// the CLI mints the transcript's id for its own row and nothing here may
/// invent one that disagrees with it.
fn forged_ending(ending: &TaskEnding) -> Message {
    Message::User {
        message: UserEnvelope {
            role: "user".to_owned(),
            content: vec![crate::transcript::notice_block(&ending.text)],
            extras: serde_json::Map::new(),
        },
        session_id: String::new(),
        parent_tool_use_id: None,
        uuid: None,
        tool_use_result: None,
        timestamp: None,
        synthetic: false,
        extras: serde_json::Map::new(),
    }
}

/// The message range of every turn `spans` names, in conversation order, each
/// with what names it.
///
/// **The one place a turn's boundaries are computed.** The session record
/// carries every turn and a page carries a window of them, and a second
/// computation is how the two would come to disagree about what a turn is.
/// Ranges rather than turns, so a page renders the window it keeps instead of
/// every turn the conversation holds.
fn turn_ranges<'a>(
    messages: &[Message],
    spans: &'a [TurnSpan],
) -> Vec<(usize, usize, Option<&'a str>)> {
    let opens = opens_of(spans);
    // **A conversation can hold no turn at all**, and a client folds turns, so
    // it rides one instead of none. A session whose only rows are shapes the
    // fold opens nothing at - an interrupt marker, a task notice, a body no
    // call holds - is the case; handing a client no turns would leave the
    // whole of it unreachable, which is exactly what a page's `None` cursor
    // stops it asking for.
    if opens.is_empty() {
        return if messages.is_empty() { Vec::new() } else { vec![(0, messages.len(), None)] };
    }
    // A frame above the first turn - an assistant or system row, a delivery
    // the CLI pressed into a running turn - comes before it, so the first turn
    // carries it. Leaving it above would put it where no walk reaches.
    (0..opens.len())
        .map(|at| {
            let from = if at == 0 { 0 } else { opens[at].0 };
            let to = opens.get(at + 1).map_or(messages.len(), |&(open, _)| open);
            (from, to, opens[at].1)
        })
        .collect()
}

/// `messages` as the whole turns `rendered` names, in conversation order.
///
/// The fold's own answer is what a caller passes, so the endings it read ride
/// along rather than being searched for again here.
pub fn all_turns(messages: &[Message], rendered: &Rendered) -> Vec<TurnWire> {
    turn_ranges(messages, &rendered.turns)
        .into_iter()
        .map(|(from, to, key)| {
            let mut turn = turn_wire(messages, from, to, key, &rendered.endings);
            compact_counters(&mut turn.messages);
            turn
        })
        .collect()
}

/// Slice a conversation into whole turns, newest first.
///
/// **The fold is what says where a turn begins** - the span `render` reports,
/// in message terms. Slicing on a count of messages instead would cut a turn
/// in half, which is the breakage the design exists to avoid.
///
/// **The cursor errs toward OVERLAP, never toward a gap.** A client asking
/// for more may be handed a turn it already has - it keys its turns and drops
/// the repeats - but never a HOLE, which is history it has no way to ask for
/// again.
///
/// **The cursor is a POSITION, not a name, and the reason is a fact about the
/// read rather than a preference.** A turn's name is its key, and the fold
/// builds one only from a `Message::Result` frame - while a conversation read
/// from a transcript carries none: `SessionMessageKind` is `User | Assistant |
/// System` with no Result kind, and the replay synthesizer never emits one. So
/// a keyed cursor is `None` on every page of every transcript-derived
/// conversation, and a client reading `None` as "nothing above" stops after
/// the first page. The message the page's first turn opens at is the one thing
/// that always names it.
///
/// **`dropped` is how many messages the held conversation has lost off the
/// front**, and it is what makes that position survive a drop: the held list
/// is renumbered by one and the conversation is not, so a page carries the
/// conversation's numbering and this is the offset between the two.
pub fn page(
    messages: &[Message],
    rendered: &Rendered,
    dropped: usize,
    dropped_thinking: i64,
    before: Option<&str>,
    turns: u32,
) -> Page {
    page_within(
        messages,
        rendered,
        dropped,
        dropped_thinking,
        before,
        turns,
        usize::MAX,
        "page",
        None,
    )
}

/// [`page`], with the whole turns it carries held to `budget` bytes.
///
/// **A page of twenty turns is unbounded in bytes**, and the first page is
/// what a seat switch waits on: a few large turns made one 5 MB, which is
/// seconds over a network and milliseconds locally. A `more` answer is held
/// the same way (`[server] more_page_cap_kb`): its page contract is whole
/// turns too, so one large turn crossed whole and a single page could be tens
/// of megabytes. So the newest page is cut where the budget runs out, at a
/// turn boundary - a client draws turns, and a turn cut in half is one it
/// cannot place. The handle it carries is the first kept turn's, so `more`
/// walks the history above it as it always did.
///
/// **The bytes counted are the bytes SERVED**: each candidate is bounded by
/// `cap` first, exactly as the page would have been after the cut. Measuring
/// the raw records would make the page's size follow a payload the client
/// never receives, and a seat with a few multi-MB records would keep one
/// turn where the capped page fits many.
///
/// **The newest turn is served even when it alone is over the budget**: a
/// budget that answered with no page would draw the seat empty, and one that
/// split the turn would hand a client half of it. `usize::MAX` is no budget.
///
/// `budget_key` is the `forge.toml` key the budget came from, named on a cut:
/// two caps share the default, and a line that did not say which one moved
/// would mislead.
pub fn page_within(
    messages: &[Message],
    rendered: &Rendered,
    dropped: usize,
    dropped_thinking: i64,
    before: Option<&str>,
    turns: u32,
    budget: usize,
    budget_key: &'static str,
    cap: Option<(usize, &SessionSlot)>,
) -> Page {
    // A page of no turns ends where it began: its cursor would name the turn it
    // already opened at, so a client walking back would ask for the same page
    // forever.
    let turns = turns.max(1) as usize;
    let ranges = turn_ranges(messages, &rendered.turns);
    let opens = opens_of(&rendered.turns);

    // A cursor names the message the previous page BEGAN at, so the page above
    // ends where that one started: the two meet exactly.
    //
    // **In the conversation's numbering rather than the held list's.**
    // `dropped` is what the cap has taken off the front: a cursor written
    // before a drop still names the message it meant. A cursor below the
    // floor never arrives here - the `more` walk answers those from the
    // transcript - and one landing under it is the empty page above rather
    // than a window the client did not ask for. A cursor no turn opens at, or
    // one this server did not write, keeps the fallback it always had.
    let ends_at = match before.and_then(|cursor| cursor.parse::<usize>().ok()) {
        None => ranges.len(),
        Some(named) => match named.checked_sub(dropped) {
            None => 0,
            Some(started) => {
                opens.iter().position(|&(open, _)| open == started).unwrap_or(ranges.len())
            }
        },
    };

    // Only the window is rendered: building every turn to keep a few of them
    // would walk and encode the whole conversation on a path a reader hits
    // while scrolling.
    let window_first = ends_at.saturating_sub(turns);
    let (first, mut page_turns) = if budget == usize::MAX {
        let page_turns: Vec<TurnWire> = ranges[window_first..ends_at]
            .iter()
            .map(|&(from, to, key)| turn_wire(messages, from, to, key, &rendered.endings))
            .collect();
        (window_first, page_turns)
    } else {
        // From the newest turn back, while the page still fits. The newest
        // turn is kept regardless, which is what makes a budget under one
        // turn answer with one rather than with none.
        let mut kept: Vec<TurnWire> = Vec::new();
        let mut total = 0usize;
        let mut first = ends_at;
        while first > window_first {
            let (from, to, key) = ranges[first - 1];
            let mut turn = turn_wire(messages, from, to, key, &rendered.endings);
            // The page is served bounded, so it is measured bounded: the cap
            // is what the wire would carry, and the log below names only the
            // turns the cut keeps.
            let reports = cap.map_or_else(Vec::new, |(cap, _)| cap_turn(&mut turn, cap));
            let bytes = serde_json::to_string(&turn).map_or(0, |text| text.len() + 1);
            if !kept.is_empty() && total.saturating_add(bytes) > budget {
                break;
            }
            if let Some((cap, seat)) = cap {
                for (id, report) in reports {
                    log_record_cap(seat, id.as_deref(), cap, &report);
                }
            }
            total = total.saturating_add(bytes);
            kept.push(turn);
            first -= 1;
        }
        // **The cut is named where it happens.** A page that stopped at its
        // budget is not a page that ran out of history: its cursor is what
        // says the turns above it are still askable, and a reader of the log
        // should not have to tell a short page from a finished walk.
        if let Some((_, seat)) = cap.filter(|_| first > window_first) {
            tracing::debug!(
                event_name = "page_cut_by_budget",
                slot = %seat.display(),
                budget_key,
                budget,
                turns_kept = kept.len(),
                turns_available = ends_at - window_first,
                "the page stopped at its byte budget; its cursor names the turns above it",
            );
        }
        kept.reverse();
        (first, kept)
    };
    // After the cut, so the budget still counts the bytes a page used to carry
    // and keeps the same whole turns it always did.
    for turn in &mut page_turns {
        compact_counters(&mut turn.messages);
    }
    // The debt belongs to the turn open at the window's end, which only a page
    // that serves that turn carries. Fold it into the first counter the turn
    // still has, so a client's own fold sums the whole turn.
    if dropped_thinking != 0
        && ends_at == ranges.len()
        && let Some(turn) = page_turns.last_mut()
        && let Some(counter) = turn.messages.iter_mut().find(|frame| is_counter(frame))
    {
        let total = grown(counter).saturating_add(dropped_thinking);
        if let Some(object) = counter.as_object_mut() {
            object.insert("estimated_tokens_delta".to_owned(), Value::from(total));
        }
    }

    // `None` is the real "nothing above this page": a page already opening on
    // the conversation's first turn has nothing to walk back to, and that is
    // the one case a client stops asking.
    let cursor = if first == 0 {
        None
    } else {
        // Written back into the conversation's numbering, which is the
        // numbering the client hands it back in.
        opens.get(first).map(|&(open, _)| (open + dropped).to_string())
    };

    Page { turns: page_turns, cursor }
}

/// How long a request waits for a live seat's first `Connected` before
/// answering without a conversation.
///
/// The fold applies the seal within a frame of it arriving, so this is a
/// failure case rather than a pacing mechanism. **What it decides is the
/// same for both readers and they answer differently**: a snapshot carries an
/// empty conversation, which a client draws as a seat with nothing to show,
/// while a page is REFUSED - an empty page carries `cursor: null`, and a
/// client reads that as the end of the history rather than as a delay.
const SEED_WAIT: std::time::Duration = std::time::Duration::from_millis(500);

/// The seat's conversation, answering from its own transcript when the
/// stream holds none.
///
/// **A seat has two sources of history and each answers what it alone can.**
/// The transport's held copy is the live one: the fold applies every
/// `ChatAppended` the session emits, so a seat that has run in this process
/// already has its conversation here. The transcript on disk is what a seat
/// LEFT: a sleeping seat has no copy, and reading its file costs no wait
/// (measured live 2026-10-12: ~2.6s for a history on disk).
///
/// So a copy is returned as it stands, and a seat with none is seeded from
/// the file. **A copy of another occupant is waited for, never replaced and
/// never answered over**: the seal the header moved on from is the fold's to
/// apply, and a transcript seed here would be a row subset - the `Result`
/// frames a transcript never carries would go with the copy.
///
/// **The two paths do not cost the same on failure.** A held copy that never
/// turns current spends the wait and is then answered as no conversation -
/// refused rather than filled from a subset. A seat with no copy pays the
/// file read and answers at once with what the read inserted, or with
/// nothing when the read's own occupant is stale; a file that holds nothing
/// yet falls through to the same bounded wait for its first `Connected`.
///
/// No lock is held across the wait, and two callers on two cold seats do not
/// serialise: each waits on its own seat's arrival, and the notice only says
/// that some seat's copy moved.
pub async fn conversation_for(
    state: &TransportState,
    slot: &SessionSlot,
    timings: &mut OpenTimings,
) -> Option<Arc<Held>> {
    // **A copy of another occupant is no copy.** Answering with it would pair
    // the new epoch's count with rows it never wrote.
    let occupant = state.surface.header(slot).session_id;
    let current = |held: &Arc<Held>| held.holds_occupant(occupant.as_ref());
    if let Some(held) = state.conversations.get(slot) {
        if current(&held) {
            return Some(held);
        }
        // The seal is on its way; the fold's reseed lands on the copy the
        // seat already holds and wakes this wait.
        return await_current(state, slot, &current, SEED_WAIT).await;
    }
    // **The file seeds only a seat with NO copy.** `insert` keeps what is
    // there, so a copy that appeared while the file was read is the stream's
    // and must not be overwritten.
    if let Some(held) = seed_from_transcript(state, slot, timings).await {
        // **The seed must be the seat's copy too.** Its occupant is the
        // read's own resolution, which can lag a header that already moved
        // on: that copy would answer for the run it did not read.
        return current(&held).then_some(held);
    }
    // **The file holds nothing yet.** A live session's first `Connected` may
    // still be on its way; wait for the fold to apply it, bounded, so a page
    // opening on a just-spawned seat draws it rather than an empty
    // conversation.
    await_current(state, slot, &current, SEED_WAIT).await
}

/// Wait, up to `within`, for `slot`'s held copy to satisfy `current`, and
/// answer it when it does.
///
/// **The notice is enabled before the first look**, so a copy that moves
/// between the look and the wait is still seen on this pass rather than
/// costing the caller its whole budget. `None` when no current copy arrives,
/// which the caller answers as a seat with no conversation.
async fn await_current(
    state: &TransportState,
    slot: &SessionSlot,
    current: &impl Fn(&Arc<Held>) -> bool,
    within: std::time::Duration,
) -> Option<Arc<Held>> {
    let deadline = std::time::Instant::now() + within;
    loop {
        // **A fresh notice every pass.** `Notified` is fused once it completes,
        // and this one fires for any seat: a reused future would return Ready
        // on every later poll (the timeout polls it first) and spin this loop
        // to its deadline instead of waiting. Enabled before the look, so a
        // copy that moves between the look and the wait is still seen.
        let seeded = state.conversations.seeded_notice();
        tokio::pin!(seeded);
        seeded.as_mut().enable();
        if let Some(held) = state.conversations.get(slot).filter(current) {
            return Some(held);
        }
        let left = deadline.saturating_duration_since(std::time::Instant::now());
        if left.is_zero() {
            break;
        }
        if tokio::time::timeout(left, seeded.as_mut()).await.is_err() {
            break;
        }
    }
    let held = state.conversations.get(slot).filter(current);
    if held.is_none() {
        tracing::debug!(
            event_name = "conversation_not_current",
            slot = %slot.display(),
            waited_ms = within.as_millis(),
            "no current conversation for this seat within the wait; the request is answered \
             without one, which a client draws the same way it draws a seat that is not \
             running",
        );
    }
    held
}

/// The conversation the seat's transcript holds, for a seat this transport
/// holds nothing for.
///
/// **This is the read a resume performs** (`session_history`), so what it
/// seeds is what a `Connected` would bring - only sooner, because the file
/// is already written. `None` when the seat records no session at all, or
/// its file holds nothing yet: a seat that has never run has no history, and
/// an empty conversation would claim exactly that.
async fn seed_from_transcript(
    state: &TransportState,
    slot: &SessionSlot,
    timings: &mut OpenTimings,
) -> Option<Arc<Held>> {
    // A test can hold the seed here, between the cache check that found no
    // copy and the read: a `Connected` that lands while this is held is the
    // race the insert below is written to lose.
    #[cfg(any(test, feature = "testing"))]
    state.conversations.pass_the_seed_hold().await;
    let at = std::time::Instant::now();
    let surface = Arc::clone(&state.surface);
    let seat = slot.clone();
    let read =
        tokio::task::spawn_blocking(move || surface.conversation(&seat, Path::new(""))).await;
    let read = read.ok()?;
    if read.messages.is_empty() {
        return None;
    }
    let mut conversation =
        // A file read, whole: the cap's own drop is what the floor counts.
        Conversation::new(read.messages, read.compaction_count, 0);
    // The read is the cold open's cost and nothing else's, so it is timed
    // where it runs rather than inferred from the conversation's total.
    timings.seed_ms = at.elapsed().as_secs_f64() * 1000.0;
    // **The copy says which run it holds** (#2256). A transcript seeded while
    // the seat ran one occupant must not answer under another: the seal the
    // fold has not applied yet leaves this copy standing while the header
    // already names the new run.
    let occupant = read.session_id.map(forge_primitives::SessionId::new);
    conversation.set_occupant(occupant.clone());
    // **A copy the stream inserted while the file was read is the live one,
    // and it stays.** The seat held none when this read started, so a copy
    // that appeared since came from the stream, which carries frames a
    // transcript does not.
    Some(state.conversations.insert(slot, conversation))
}

/// How many of a conversation's newest turns a subscribe carries.
///
/// **A subscribe is not a request for a whole transcript.** A client draws a
/// window and asks `more` for what is above it, so handing it the whole
/// conversation is a transcript's worth of bytes written per client per
/// connect and per refresh - measured at 111.7 MB of socket for four
/// refreshes of a 68.9 MB transcript, which is 99% of that connection's
/// traffic - for turns it will not draw.
///
/// The window matches the client's own `SUBSCRIBE_TURNS` (20), which its seat
/// record trims to, and the `MORE_TURNS` page a client asks for: the newest
/// page a subscribe carries and the newest page a client asks for are the same
/// page, so a client that asks anyway is handed what it already holds, which it
/// drops by turn key.
///
/// **It is a ceiling, not the page.** `[server] first_frame_cap_kb` holds the
/// same page to its bytes (see [`page_within`]), so a conversation whose
/// turns are large carries fewer than this.
pub const SUBSCRIBE_TURNS: u32 = 20;

/// The slices of one session open, in milliseconds, gathered as its read
/// runs (#2374).
///
/// **What splits a slow switch between this server and the client.** A
/// client's own marks say when its click happened and when the first row
/// drew; these are what the server was doing in between, so a gap the two
/// artefacts have to explain is placed by them rather than guessed at.
///
/// **The slices are disjoint.** Each is taken around one read on the path,
/// in the order the path runs them, so their sum against `open_ms` is the
/// work that did not name itself; the seed is the transcript read only a
/// cold seat pays, and the conversation total excludes it. The frame's own
/// record, `frame_write`, carries `prep_ms` (the record serialised and
/// compressed into the bytes) and `write_ms` (the socket's share) - both
/// inside `open_ms`, and split out of what the record's own reads would
/// otherwise have to explain.
#[derive(Debug, Default)]
pub struct OpenTimings {
    pub seed_ms: f64,
    pub conversation_ms: f64,
    pub page_ms: f64,
    pub encode_ms: f64,
    pub prologue_ms: f64,
    pub header_ms: f64,
    pub catch_up_ms: f64,
    pub cards_ms: f64,
    pub work_ms: f64,
    /// The review read's base refresh: the fetch and the tip it is checked
    /// against, waited on with no cap and served off the connection's loop.
    pub base_ms: f64,
    /// The scan pinned to the sha that refresh resolved.
    pub rescan_ms: f64,
    pub content_ms: f64,
    pub reviews_ms: f64,
    pub tree_reviews_ms: f64,
    pub state_ms: f64,
    pub index_ms: f64,
    pub resume_ms: f64,
    pub tail_ms: f64,
    /// How old the PR attempt behind this record is, in seconds; absent when
    /// the scan has never attempted one. The record can carry a PR past its
    /// refetch window while the watch refreshes it, and this is what says how
    /// stale what a reader drew was.
    pub pr_age_s: Option<f64>,
}

/// One read's share of the open, in milliseconds.
fn slice_ms(at: std::time::Instant) -> f64 {
    at.elapsed().as_secs_f64() * 1000.0
}

/// A subject's wire form. The ONE place it is produced.
///
/// Async because a session's git section is a filesystem read, so the
/// working tree's state is awaited rather than guessed at.
///
/// A seat naming no project forge has loaded is an error rather than an empty
/// snapshot, because a client drawing an empty page would read that as a
/// broken one rather than as a seat that does not exist.
///
/// **A project that has simply not started is not that case**: its directory
/// resolves from the declaration, so it is answered - with the session's own
/// fields empty and a working tree read per read, since no held seat has a
/// store to keep one in.
pub async fn encode_subject(state: &TransportState, subject: &Subject) -> Result<Value> {
    encode_subject_timed(state, subject, false).await.map(|(value, _)| value)
}

/// [`encode_subject`], carrying the session's own open timings.
///
/// **Split from the plain encode so the thirty readers that want only the
/// value keep their call**: a session's timeline is the open path's own
/// business, and it is the only arm that fills one.
///
/// `review` is the review page's own read: it refreshes the branch diff's
/// base and pins the stats and content reads to the sha that refresh found.
/// Every other read serves what the stores hold, so a record read never waits
/// on the network.
pub(crate) async fn encode_subject_timed(
    state: &TransportState,
    subject: &Subject,
    review: bool,
) -> Result<(Value, OpenTimings)> {
    let surface = &state.surface;
    let mut timings = OpenTimings::default();
    match subject {
        Subject::Home => Ok((serde_json::to_value(home(state, surface).await)?, timings)),
        Subject::Board => Ok((serde_json::to_value(board(surface))?, timings)),
        Subject::Session(slot) => {
            let at = std::time::Instant::now();
            let roster = surface.roster();
            let Some(cwd) = roster.cwd_for(slot) else {
                anyhow::bail!("forge holds no session for {}", slot.display());
            };
            // The record's reads ARE the seat's first read: the connection
            // holds the seat before this encode, so its snapshot takes the
            // stores under the seat's gate and the watch's first pass starts
            // from what this record carries (#2506).
            request_context_usage_if_unreported(surface, slot);
            request_mcp_snapshot_if_unreported(surface, slot);
            timings.prologue_ms = slice_ms(at);
            let wire = session(state, surface, slot, &cwd, &mut timings, review).await?;
            // The record's own serialisation, and nothing else: what the
            // measurement pass timed as the encode, so the numbers line up.
            let at = std::time::Instant::now();
            let value = serde_json::to_value(&wire)?;
            timings.encode_ms = at.elapsed().as_secs_f64() * 1000.0;
            Ok((value, timings))
        }
        // The pool's own report, scanned here rather than carried in another
        // snapshot: it belongs to no seat, and a home snapshot that scanned
        // the pool would pay for the walk on every subscribe.
        Subject::Usage => Ok((serde_json::to_value(surface.usage().await?)?, timings)),
        // The models page's read, off the workspace's own state: no walk,
        // no lock held across a scan - the fetch that changes it runs
        // elsewhere and lands as its own update.
        Subject::DictateModels => Ok((serde_json::to_value(surface.dictate_models())?, timings)),
        // The Extensions page's read: one disk scan of the plugin
        // inventory per read, plus the PATH checks and the last run.
        Subject::Extensions => Ok((serde_json::to_value(extensions(state).await)?, timings)),
    }
}

/// The Extensions page's read, as the wire carries it.
async fn extensions(state: &TransportState) -> ExtensionsWire {
    let view = state.surface.extensions().await;
    ExtensionsWire {
        snapshot: view.snapshot,
        update_run: view.update_run,
        lsp: view
            .lsp
            .into_iter()
            .map(|check| LspCheckWire {
                plugin: check.plugin,
                server: check.server,
                command: check.command,
                on_path: check.on_path,
            })
            .collect(),
    }
}

/// Ask the core for a context reading on `slot` when the seat reports none.
///
/// The reading exists only once the CLI has computed it, and a view asks
/// for the seat it is addressing. A client subscribing to a seat is that same
/// act, so the ask belongs on the read that encodes the subject: a seat only a
/// client watches would otherwise report nothing for the life of its session,
/// and the header would draw the dash an unasked seat draws rather than a bar.
///
/// Guarded on the reading rather than on the ask having happened, so a seat
/// that already reports one is not probed again by every subscribe, reconnect
/// and second tab. The answer arrives as
/// [`SessionUpdate::ContextUsageSnapshot`](crate::SessionUpdate::ContextUsageSnapshot)
/// on the stream the subscriber is already reading.
fn request_context_usage_if_unreported(surface: &ViewSurface, slot: &SessionSlot) {
    if surface.header(slot).context.percent.is_some() {
        return;
    }
    if let Err(error) = surface.refresh_context_usage(slot) {
        tracing::debug!(
            event_name = "context_usage_request_failed",
            %error,
            slot = %slot.display(),
            "a seat a client reads reports no context usage and its probe was not requested",
        );
    }
}

/// Ask the core for an MCP snapshot on `slot` when the seat reports none.
///
/// The snapshot exists only once the bridge has answered one, and only a
/// view's connect, poll and `/mcp` paths ask for it. A client subscribing
/// to a seat is that same act, so the ask belongs on the read that encodes the
/// subject: a seat only a client watches would otherwise report `mcp: None`
/// for the life of its session, and the strip's MCP row would never draw.
///
/// Guarded on the reading rather than on the ask having happened, so a seat
/// that already reports one - an empty set included, which is a snapshot
/// carrying none rather than no snapshot - is not probed again by every
/// subscribe, reconnect and second tab. The answer arrives as
/// [`SessionUpdate::McpSnapshot`](crate::SessionUpdate::McpSnapshot) on the
/// stream the subscriber is already reading. A seat replacement clears the
/// snapshot, which is the one re-ask this guard makes on its own: the field
/// reads `None` again and the next read asks.
fn request_mcp_snapshot_if_unreported(surface: &ViewSurface, slot: &SessionSlot) {
    if surface.mcp_servers(slot).is_some() {
        return;
    }
    if let Err(error) = surface.refresh_mcp_snapshot(slot) {
        tracing::debug!(
            event_name = "mcp_snapshot_request_failed",
            %error,
            slot = %slot.display(),
            "a seat a client reads reports no MCP snapshot and its ask was not requested",
        );
    }
}

/// The home's record, from the reads a home-scoped view makes.
///
/// Async because a row's work state is a filesystem read, cached by the
/// shared `WorkCache` the transport holds.
async fn home(state: &TransportState, surface: &ViewSurface) -> HomeWire {
    let roster = surface.roster();
    let accounts = surface.accounts();
    let agents = surface.agents();
    let connectors = surface.connectors(None);
    let dictate = surface.dictate();
    let encode = |value: Option<Value>| value;

    let mut projects = Vec::with_capacity(roster.projects.len());
    for project in &roster.projects {
        let seat = SessionSlot::lead(project.org.clone(), project.name.clone());
        projects.push(ProjectWire {
            work: state.work.snapshot(&seat, &project.path).await,
            would_bind: roster.would_bind(&project.key),
            chip: roster.chip_for(&project.key),
            project: project.clone(),
        });
    }

    // Each seat's own tree, read at the directory `cwd_for` resolves FOR IT:
    // the project's path for a lead, the worktree for a git worker. A seat
    // forge holds no directory for keeps `None` rather than borrowing the
    // project's read, which is what the cell's blank has to mean.
    //
    // The live snapshot filters the failure mark by what has been shown:
    // the same facts the diamond rides, read once for both.
    let live = crate::live::Live::lock(&state.live).snapshot();
    let mut agent_rows = Vec::with_capacity(agents.all().len());
    for agent in agents.all() {
        let work = match roster.cwd_for(&agent.slot) {
            Some(cwd) => Some(state.work.snapshot(&agent.slot, cwd.as_path()).await),
            None => None,
        };
        let mut row = AgentWire { work, ..AgentWire::from(agent) };
        row.failed_turn = row.failed_turn.and_then(|at| live.failed_mark(&agent.slot, at));
        agent_rows.push(row);
    }

    HomeWire {
        fleet: surface.fleet(),
        accounts,
        connectors: ConnectorsWire {
            gotify: GotifyWire { connected: connectors.gotify.connected },
            slack: SlackWire {
                connected_workspaces: connectors.slack.connected_workspaces.into_iter().collect(),
                load_failed: connectors.slack.load_failed,
            },
        },
        dictate: DictateWire {
            enabled: dictate.enabled,
            snapshot: serde_json::to_value(dictate.snapshot).unwrap_or(Value::Null),
            models_dir: dictate.models_dir,
        },
        voice: {
            let voice = surface.voices();
            VoiceWire { enabled: voice.enabled, models_dir: voice.models_dir }
        },
        cli_version: surface.cli_version().and_then(|version| serde_json::to_value(version).ok()),
        forge_version: crate::FORGE_VERSION.to_owned(),
        forge_version_short: crate::FORGE_VERSION_SHORT.to_owned(),
        boot_id: boot_id().to_owned(),
        service_status: surface.service_status().and_then(|issue| serde_json::to_value(issue).ok()),
        // The words the server prints on exit, which is what a client draws:
        // the serde tag alone is for matching, and nothing matches on this.
        fatal_error: encode(
            surface.fatal_error().map(|error| Value::String(error.user_message().to_owned())),
        ),
        biometric: surface.biometric_enrolments().into(),
        vault: surface
            .vault_state()
            .map(|state| VaultWire { locked: state.locked, idle_minutes: state.idle_minutes }),
        agents: agent_rows,
        unseen: agents
            .all()
            .iter()
            .map(|row| row.slot.clone())
            .filter(|slot| live.unseen.is_unseen(slot))
            .collect(),
        projects,
    }
}

/// The board's record, from the reads a board-scoped view makes.
///
/// No filesystem read and no seat hold: the rows, schedules and connector
/// sets are the workspace's own state, which is what makes a board read
/// cheap next to the home read it left (#2517).
fn board(surface: &ViewSurface) -> BoardWire {
    let roster = surface.roster();
    let mut projects = Vec::with_capacity(roster.projects.len());
    for project in &roster.projects {
        let subscribed = surface.connectors(Some(&project.name));
        projects.push(ProjectBoardWire {
            project: ProjectBoardIdentity { org: project.org.clone(), name: project.name.clone() },
            rows: surface.board_rows(&project.name),
            crons: roster.crons_for_project(&project.name),
            connectors: ProjectConnectorsWire {
                gotify: subscribed.gotify.subscriptions,
                slack: subscribed.slack.subscriptions,
            },
        });
    }
    BoardWire { projects }
}

/// How long an encode waits for the fold loop to catch up before filling the
/// gap from the seat's own window.
///
/// **A scheduling hop, not a stall.** The common case is already caught up
/// (the wait is not entered) or one hop behind; the fallback below is what
/// keeps the bound from being a way to lose frames.
const CATCH_UP_WAIT: std::time::Duration = std::time::Duration::from_millis(250);

/// How long a read whose copy needs an unapplied seal waits for the fold
/// before refusing.
///
/// **The window cannot stand in for the fold here.** A seal's history is not
/// ringed, so nothing the fill below can reach carries it; only the fold's own
/// apply does. That apply lands in milliseconds when the fold is running, so
/// this bound is reached only by a fold that is not - and a refusal costs the
/// client a whole re-ask (`client/src/session/live.ts`), so the wait is what
/// keeps an ordinary lag from paying it.
const SEAL_WAIT: std::time::Duration = std::time::Duration::from_secs(1);

/// Why a read could not bring its copy current, which the refusal's text says
/// apart: they are different losses and only one is about a number.
#[derive(Debug)]
enum NotReady {
    /// The seat's window no longer reaches back to the copy's front.
    Behind,
    /// The epoch's seal rows are unapplied, so its history cannot be reached.
    AwaitsSeal,
}

/// Bring `slot`'s held conversation up to `want`, or report that it cannot.
///
/// **One requirement: the copy an answer carries must hold every frame
/// numbered at or below `want`.** The fold loop consumes the fanout in a task
/// of its own, so a frame minted before the counter was read can reach the
/// copy after it - waited on here, briefly. When the wait runs out the frames
/// come from the seat's own window instead, which the fanout feeds directly
/// (every frame, subscriber or not) under its two ceilings - a frame count and
/// a byte budget, so a heavy seat's window holds fewer than the cap. Only a
/// gap wider than that window is refused, loudly and by name: a stuck fold is
/// forge's problem, and an answer that pretended otherwise is a row lost in
/// silence.
async fn catch_up(state: &TransportState, slot: &SessionSlot, want: u64) -> Result<(), NotReady> {
    if want == 0 || state.conversations.consumed_for(slot) >= want {
        return Ok(());
    }
    if state.conversations.await_consumed(slot, want, CATCH_UP_WAIT).await {
        return Ok(());
    }
    // **A seal's rows the fold has not applied cannot be stood in for by the
    // ring.** They are what the epoch opened on, and no window holds them
    // (#2446), so the fill below would answer the new epoch's number without
    // them. Waited on here, before the fill lock, for the fold's own mark:
    // applying the seal resets it to zero and the frames after it climb back
    // to `want`, so reaching `want` means the reseed is in place. The wait is
    // what keeps a transient lag from failing the read.
    if awaits_seal_history(state, slot)
        && state.conversations.await_consumed(slot, want, SEAL_WAIT).await
    {
        return Ok(());
    }
    // **The fill runs under the seat's lock, and re-reads the watermark.**
    // The fold may have caught up while the wait ran, and it applies the same
    // frames if it did: the check, the applies and the mark are one step, so
    // a frame crosses exactly once whichever path got there first.
    let fill = state.conversations.fill_lock(slot);
    let _filling = fill.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
    if state.conversations.consumed_for(slot) >= want {
        return Ok(());
    }
    let from = state.conversations.consumed_for(slot).saturating_add(1);
    let frames = state.surface.seat_frames_between(slot, from, want);
    // **A stretch holding only a seal is not a stretch the copy is behind on.**
    // The frames that close an epoch are numbered but never ringed, so no
    // window can serve them however far back it reaches - and counting them as
    // missing would refuse a seat whose newest frame is a release, which is
    // every seat a client reconnects to after it went to sleep.
    let needed = want
        .saturating_sub(from)
        .saturating_add(1)
        .saturating_sub(state.surface.seat_unringed_between(slot, from, want));
    if u64::try_from(frames.len()).unwrap_or(u64::MAX) < needed {
        tracing::error!(
            event_name = "seat_copy_behind",
            slot = %slot.display(),
            want,
            got = frames.len(),
            "the seat's copy is behind the number an answer would carry and its own window no \
             longer reaches back; the read is refused rather than answered with a hole",
        );
        return Err(NotReady::Behind);
    }
    // **A seat with no copy gets one from the frames themselves, and a copy of
    // another occupant is no copy.** The fold holds nothing for a seat nothing
    // has connected on, and a seal the fold has not applied yet leaves the
    // previous run's copy standing: neither may answer for the number this
    // call carries. The stretch materialises a copy - seeded EMPTY, built by
    // the applies below, one place a row may enter so nothing doubles - tagged
    // with the occupant it is for, and the fold skips these frames afterwards
    // by the watermark this marks last.
    //
    // **Except when the epoch opened on a seal's rows, which the ring cannot
    // carry.** A copy built from the frames below would answer the new
    // epoch's number without them, and the occupant check cannot catch it:
    // the synthetic copy carries the header's own occupant (#2446). Refused
    // rather than synthesised; the fold's apply of the seal is what makes the
    // read answer.
    let occupant = state.surface.header(slot).session_id;
    let matching =
        state.conversations.get(slot).is_some_and(|held| held.holds_occupant(occupant.as_ref()));
    if !matching {
        if state.surface.seat_epoch_opened_with_rows(slot) {
            tracing::error!(
                event_name = "seat_copy_awaits_seal",
                slot = %slot.display(),
                want,
                "the seat's copy is behind a seal the fold has not applied, and the ring cannot \
                 carry the history that seal opened on; the read is refused rather than answered \
                 with a hole",
            );
            return Err(NotReady::AwaitsSeal);
        }
        let mut conversation = crate::transport::conversation::Conversation::empty();
        conversation.set_occupant(occupant);
        state.conversations.replace(slot, conversation);
    }
    for frame in &frames {
        state.conversations.apply(&frame.update, frame.seq);
    }
    state.conversations.consumed(slot, want);
    Ok(())
}

/// Whether `slot`'s copy is missing rows its epoch opened on.
///
/// The mark describes the LATEST seal, so between an occupant change and its
/// seal's mint the value is still the previous seal's - which is what notices
/// that gap whenever the previous epoch carried rows. A seat whose first
/// epoch is pending, or whose previous seal was empty, has no such mark, and a
/// read in that mint gap can still synthesise a copy.
fn awaits_seal_history(state: &TransportState, slot: &SessionSlot) -> bool {
    let occupant = state.surface.header(slot).session_id;
    let matching =
        state.conversations.get(slot).is_some_and(|held| held.holds_occupant(occupant.as_ref()));
    !matching && state.surface.seat_epoch_opened_with_rows(slot)
}

/// The refresh a record may state as its own check: none when the run
/// answered nothing (its leader panicked or was cancelled, or a bound cut
/// it), because the words beside that state are forge's own rather than
/// git's.
fn checked_refresh(
    refresh: Option<&forge_workspace::env::git_diff::BaseRefresh>,
) -> Option<&forge_workspace::env::git_diff::BaseRefresh> {
    refresh.filter(|refresh| !refresh.unanswered && !refresh.cut)
}

/// Why a record's `diff` could not check its base: the refresh's words when
/// it ran, the store's when the refresh answered nothing, and none when no
/// refresh ran - the field names git's words, never forge's own.
fn diff_base_error<'a>(
    refresh: Option<&'a forge_workspace::env::git_diff::BaseRefresh>,
    stored_error: Option<&'a str>,
) -> Option<&'a str> {
    match (refresh, checked_refresh(refresh)) {
        (_, Some(refresh)) => refresh.error.as_deref(),
        (Some(_), None) => stored_error,
        (None, None) => None,
    }
}

/// The base a record's `git` row states. **When the read refreshed, the row
/// states the check its `diff` was read against**, never the store's newer
/// measurement (a bypass answered while a fetch landed); a read that did not
/// refresh, or one whose refresh answered nothing, states what the store
/// holds.
fn stated_base<'a>(
    refresh: Option<&'a forge_workspace::env::git_diff::BaseRefresh>,
    stored: Option<forge_primitives::git_diff::BaseState>,
    stored_error: Option<&'a str>,
) -> (Option<forge_primitives::git_diff::BaseState>, Option<&'a str>) {
    match checked_refresh(refresh) {
        Some(refresh) => (Some(refresh.state), refresh.error.as_deref()),
        None => (stored, stored_error),
    }
}

/// One session's record, from the reads a session-scoped view makes.
async fn session(
    state: &TransportState,
    surface: &ViewSurface,
    slot: &SessionSlot,
    cwd: &Path,
    timings: &mut OpenTimings,
    review: bool,
) -> Result<SessionWire> {
    let at = std::time::Instant::now();
    let header = surface.header(slot);
    timings.header_ms = slice_ms(at);
    // **The number is read BEFORE the copy's frames are.** Every frame minted
    // before this read is one the read below must carry; anything minted after
    // it carries a number at or above `seq`, which a client applies live rather
    // than skipping as already drawn. Read after the read, a frame minted in
    // between is in neither - a silently lost row on the reconnect path.
    let seq = surface.seat_next_seq(slot);
    // **And the copy has to have caught up to it.** The copy is filled by the
    // fold loop consuming the same stream in another task, so a frame minted
    // before the read can land after it: waited on briefly, and filled from
    // the seat's own window when the wait runs out. The answer never names a
    // number its copy cannot cover - a refusal is visible, a lost row is not.
    let at = std::time::Instant::now();
    if let Err(reason) = catch_up(state, slot, seq.saturating_sub(1)).await {
        let why = match reason {
            NotReady::Behind => "its copy is behind the number an answer would carry",
            NotReady::AwaitsSeal => {
                "the fold has not applied its epoch's seal, whose history an answer would be \
                 missing"
            }
        };
        anyhow::bail!("the seat's stream has not settled for {}; {why}", slot.display());
    }
    timings.catch_up_ms = slice_ms(at);
    // The conversation's share of the record, taken through one blocking task
    // so the fold never runs on the reactor or under the lock. A seat with no
    // conversation is answered with the empty one.
    let at = std::time::Instant::now();
    let conversation = conversation_for(state, slot, timings).await;
    // **The slices are DISJOINT, and the seed is taken out of this total.**
    // The transcript read a cold seat pays is timed inside `conversation_for`,
    // so a total over the whole call would carry it twice: the sum the line's
    // remainder is read against would go short by the whole read - worst on
    // exactly the slow cold open the line exists to explain.
    timings.conversation_ms = at.elapsed().as_secs_f64() * 1000.0 - timings.seed_ms;
    // The served page's records are bounded here, off the lock like the page
    // itself: a mailbox seat's structured records made one page 70 MB (#2374).
    let cap = surface.tool_record_cap_bytes();
    let first_frame = surface.first_frame_cap_bytes();
    let read = match conversation {
        Some(held) => {
            let seat = slot.clone();
            let capped = seat.clone();
            let surface = Arc::clone(&state.surface);
            tokio::task::spawn_blocking(move || {
                let at = std::time::Instant::now();
                // **The record's conversation IS a page, cursor and all**
                // (#2420): a client seeds its conversation from this answer
                // instead of asking `more` for the same page, and the handle
                // is what names the history above it.
                let (mut page, count, covered) = held.read(|held| {
                    let mut page = page_within(
                        held.messages(),
                        held.rendered(),
                        held.dropped(),
                        held.dropped_thinking(),
                        None,
                        SUBSCRIBE_TURNS,
                        first_frame,
                        "first_frame_cap_kb",
                        Some((cap, &capped)),
                    );
                    floor_cursor(&mut page, held.dropped());
                    (page, held.compaction_count(), held.watermark())
                });
                // **A page cut to a byte budget was capped as it was
                // measured**, turn by turn, so this pass would walk every
                // record of it for nothing. Only the uncut page still needs
                // it.
                if first_frame == usize::MAX {
                    cap_tool_records(&mut page.turns, cap, &capped);
                }
                let names = cron_names_of(&surface, &capped, &page.turns);
                // **The page's cost is the whole subscribe's**, the cron read
                // included: the standing watch reads this number against the
                // record it explains, so a read it excludes reads as cheaper
                // than the record. The row is annotating the page, but it is
                // part of building the answer.
                let page_ms = at.elapsed().as_secs_f64() * 1000.0;
                (page.turns, page.cursor, count, page_ms, covered, names)
            })
            .await
            .unwrap_or_else(|error| {
                tracing::warn!(
                    event_name = "transcript_fold_failed",
                    %error,
                    slot = %seat.display(),
                    "the fold did not finish; the record is answered without it",
                );
                (Vec::new(), None, 0, 0.0, 0, BTreeMap::new())
            })
        }
        None => (Vec::new(), None, 0, 0.0, 0, BTreeMap::new()),
    };
    let (turns, cursor, compaction_count, page_ms, covered, cron_names) = read;
    timings.page_ms = page_ms;
    // **The answer's number is the copy's own**, not catch_up's pass: a seal
    // can reseed the copy after that check, and a pre-seal number would have
    // the client skip the new run's first frames as already drawn (#2443).
    let seq = seq.min(covered.saturating_add(1));
    // The instance list crosses from the fold that ANNOUNCES it rather than
    // from a second walk here: the session task folds a card per frame and
    // pushes `SubagentCardsChanged` off the same walk, so this is what that
    // fold holds. **It was folded here until the cap landed** - over the whole
    // held conversation on every read, which a drop would have shortened: a
    // record answering a window of cards over a window of messages is a
    // section that shrinks with the cap rather than with the conversation.
    let at = std::time::Instant::now();
    let instances = surface.subagent_cards(slot);
    timings.cards_ms = slice_ms(at);
    // The seat's scan, as the loop that owns it last answered it. Nothing is
    // read here: the tree is read for a seat somebody is showing, and this
    // answers what that read found - the tile's branch, count and PR are one
    // instant, so a branch switch cannot render a PR for a branch this record
    // does not name.
    let at = std::time::Instant::now();
    let held = surface.work(slot, cwd, review).await;
    timings.work_ms = slice_ms(at);
    // **Only the review page's own read checks the base.** An ordinary record
    // read serves the stores as they stand - a seat page never waits on the
    // network - and the review read answers the last completed check and kicks
    // a background refresh. A seat with no session is the exception: no watch
    // delivers a kicked check there, so that read waits for its own (#2818).
    let at = std::time::Instant::now();
    let refresh = if review { surface.refresh_diff_base(slot, &held).await } else { None };
    timings.base_ms = slice_ms(at);
    // The refresh and the scan taken after it are one read of one base: the
    // count, the chain and the files cannot describe two refs.
    let at = std::time::Instant::now();
    let pin = match &refresh {
        Some(refresh) => surface
            .diff_scan(slot, &held.cwd, refresh)
            .await
            .map(|scan| (scan, refresh.base_ref.as_deref())),
        None => None,
    };
    // **A refused pin is re-read locally, never served from the pre-refresh
    // store.** The refresh moved the base; the store's count predates it, and
    // that count beside content read at the ref the fetch left is the
    // two-commits answer this read exists to prevent. The re-read is pinned to
    // the sha the refresh resolved and carries its state, so the count, the
    // files and the row's `base` all describe that one base; only a refresh
    // that resolved no sha falls back to the ref's name for both.
    let (held, base_ref) = match pin {
        Some((scan, base_ref)) => (scan, base_ref),
        None => match &refresh {
            Some(refresh) => (
                surface
                    .work_pinned_local(slot, &held.cwd, refresh.base_ref.as_deref(), Some(refresh))
                    .await,
                refresh.base_ref.as_deref(),
            ),
            None => (held, None),
        },
    };
    timings.rescan_ms = slice_ms(at);
    // The content read, beside the stats one: bounded, and taken here
    // rather than stored with the row because a record read is a cold load,
    // a reconnect or a seat swap - not a frame a moving tree pushes.
    let base = match &refresh {
        // The drawn diff reads the ref the refresh left: the gate's pinned
        // scan, or the local re-read at the same sha; only a refresh that
        // resolved no sha reads by name. Either way the state the fetch found
        // is this diff's state.
        Some(refresh) => Some(refresh.state),
        // An ordinary read has not checked the base. Saying so is what keeps a
        // cached record from drawing its diff as current.
        None if held.diff.branch_ahead.is_populated() => {
            Some(forge_primitives::git_diff::BaseState::Unverified)
        }
        None => None,
    };
    // Why the refresh could not run, in git's own words: the page draws it
    // beside the base line rather than only saying the base could not be
    // checked. An ordinary read has no refresh and names no failure, and a
    // placeholder refresh names the store's own check rather than forge's
    // words for a run that never reported.
    let base_error = diff_base_error(refresh.as_ref(), held.base_error.as_deref());
    let at = std::time::Instant::now();
    let diff = scan_content(&held.cwd, &held.diff, base, base_ref, base_error).await;
    timings.content_ms = slice_ms(at);
    timings.pr_age_s = held
        .diff
        .pr_fetched_at
        .and_then(|at| std::time::SystemTime::now().duration_since(at).ok())
        .map(|age| age.as_secs_f64());
    let work = work_from_scan(&held.diff, &held.cwd);
    // The tree behind the row's depth, from the same held scan and through
    // the same constructor the pushed frame goes through - one derivation,
    // so a client that applies the update lands on the record's own answer.
    let (row_base, row_base_error) =
        stated_base(refresh.as_ref(), held.base, held.base_error.as_deref());
    let git = git_work_view(&held.diff, row_base, row_base_error);
    let branch = work.branch.clone().unwrap_or_default();
    let at = std::time::Instant::now();
    let reviews = surface.reviews(slot.project(), &branch);
    // The whole files for anchors the bounded content cannot place: read
    // only for a comment this record's own diff carries the file of, and
    // zero reads when every anchor sits inside its hunks.
    let whole = match &reviews.threads {
        Ok(threads) => {
            whole_files_for(
                &held.cwd,
                surface.document_max_bytes(),
                threads,
                &diff,
                held.diff.default_branch.as_deref(),
                // The sha this diff was pinned to, when the read refreshed and
                // pinned one; the name stands in when it did not.
                base_ref,
            )
            .await
        }
        Err(_) => WholeFiles::new(),
    };
    timings.reviews_ms = slice_ms(at);
    // The seat's tree half, placed against the working files: the reads
    // are files off disk, so they run off the reactor like the walk below.
    // Zero tree threads means zero reads. A panicked read answers Err, so
    // the page draws the failure as itself rather than as an empty review.
    let at = std::time::Instant::now();
    let tree_reviews = {
        let surface = std::sync::Arc::clone(&state.surface);
        let seat = slot.clone();
        tokio::task::spawn_blocking(move || surface.tree_reviews(&seat)).await.unwrap_or_else(
            |error| {
                tracing::warn!(
                    event_name = "tree_reviews_read_failed",
                    %error,
                    slot = %slot.display(),
                    "the seat-tree review read did not finish; the record answers the failure",
                );
                let why = "the seat tree's reviews could not be read".to_owned();
                crate::surface::reviews::TreeReviewsView {
                    scope: forge_primitives::review::tree_scope_ref(slot),
                    threads: Err(why.clone()),
                    reviews: Err(why),
                }
            },
        )
    };
    timings.tree_reviews_ms = slice_ms(at);
    let at = std::time::Instant::now();
    let state_at = surface.session(slot, cwd);
    timings.state_ms = slice_ms(at);
    // The seat's own loop walks the index and pushes the movement, so this
    // answers that store - and a seat nothing has walked yet takes the same
    // walk under the seat's gate, off the reactor: whichever of the record and
    // the first pass gets there first stores, and the other answers it rather
    // than walking the same tree again.
    let at = std::time::Instant::now();
    let file_index = surface.file_index_for_record(slot).await;
    timings.index_ms = slice_ms(at);
    // **The reads the watch seeds from are all taken, so the mark fires here**
    // - after the tree's scan, the index and the process walk, and before the
    // record's remaining reads and its serialisation. The first pass starts
    // from these stores while this frame is still being built; `processes` is
    // read here so the record carries the same walk the pass seeds from.
    let at = std::time::Instant::now();
    let processes = surface.processes(slot);
    surface.note_record_read(slot);

    // One read for both halves of the record: the front is the first of the
    // list, so the two cannot disagree.
    let asks = surface.pending_asks(slot);

    // `/resume`'s list, on the record with the composer's other sources: a
    // lead reads the boot catalog's group, keeping a row only where its
    // transcript is still on disk, and a worker scans its own directory -
    // filesystem work both ways, so it is timed like the other reads and kept
    // out of the tail slice.
    let resume_at = std::time::Instant::now();
    let resume_sessions = surface
        .resumable_sessions(slot)
        .await
        .into_iter()
        .map(|info| ResumableSessionWire { session_id: info.session_id, summary: info.summary })
        .collect();
    timings.resume_ms = slice_ms(resume_at);

    let wire = SessionWire {
        slot: slot.clone(),
        file_index,
        resume_sessions,
        slash_commands: surface.slash_commands(slot),
        subagents: surface.subagents(slot),
        subagent_instances: instances,
        mcp: surface.mcp_servers(slot),
        processes,
        background_tasks: surface.background_tasks(slot),
        monitors: surface.monitors(slot),
        terminal: surface.terminal(slot),
        pending_ask: asks.first().map(PendingAskWire::from),
        pending_asks: asks.iter().map(PendingAskWire::from).collect(),
        conversation: ConversationWire { turns, cursor, compaction_count, cron_names },
        header: SessionHeaderWire {
            session_id: header.session_id.clone(),
            seq,
            model: header.model.as_ref().and_then(|model| serde_json::to_value(model).ok()),
            effort: serde_json::to_value(header.effort).unwrap_or(Value::Null),
            permission_mode: header
                .permission_mode
                .as_ref()
                .and_then(|mode| serde_json::to_value(mode).ok()),
            context: serde_json::to_value(header.context).unwrap_or(Value::Null),
            available_models: header
                .available_models
                .iter()
                .filter_map(|model| serde_json::to_value(model).ok())
                .collect(),
            turn_in_flight: header.turn_in_flight,
        },
        reviews: ReviewsWire {
            threads: ReadWire::from(reviews.threads.map(|threads| {
                reviewed_threads(threads, &diff, held.diff.default_branch.as_deref(), &whole)
            })),
            reviews: ReadWire::from(reviews.reviews),
        },
        tree_reviews: ReviewsWire {
            threads: ReadWire::from(tree_reviews.threads.map(|threads| {
                threads
                    .into_iter()
                    .map(|held| ReviewedThreadWire {
                        thread: held.thread,
                        resolution: held.placement.map(placement_wire),
                    })
                    .collect()
            })),
            reviews: ReadWire::from(tree_reviews.reviews),
        },
        state: SessionStateWire {
            slot: state_at.slot,
            scan_cwd: state_at.scan_cwd,
            queue: state_at.queue,
            voice_listeners: state_at.voice_listeners,
        },
        composer: {
            let live = crate::live::Live::lock(&state.live).snapshot();
            let composer = &live.composer;
            ComposerWire {
                machine: crate::surface::machine_turn(
                    surface.handoff_in_flight(slot),
                    composer.compacting(slot),
                ),
                bind: surface.dictate_bind().label().to_owned(),
                mode: surface.dictate_mode().label().to_owned(),
                notice: composer.landed(slot).map(|landed| {
                    serde_json::json!({
                        "kind": "landed",
                        "text": landed.text,
                        "truncated": landed.truncated,
                        "generation": landed.generation,
                        "carried": true,
                        "surface": landed.surface,
                    })
                }),
            }
        },
        work,
        git,
        pr: held.diff.pr,
        closes: held.diff.closes,
        diff,
    };
    // The tail clock opened before the process read and the resume read falls
    // inside its window, so its slice is the elapsed less that read: the
    // numbers a reader subtracts are disjoint, and the glue that remains is
    // the record's own assembly.
    timings.tail_ms = slice_ms(at) - timings.resume_ms;
    Ok(wire)
}

#[cfg(test)]
mod tests {
    /// **The queue a view reads is three fields.** The attachments the task
    /// keeps for the started row are core-side; a base64 payload on every seat
    /// record read is the thing the row frame itself strips (#2440).
    #[test]
    fn the_served_queue_carries_no_attachment_payload() {
        let state = super::SessionStateWire {
            slot: forge_primitives::SessionSlot::lead("TestOrg", "proj"),
            scan_cwd: std::path::PathBuf::from("/tmp/proj"),
            queue: vec![forge_workspace::protocol::QueuedPrompt {
                uuid: "p1".to_owned(),
                source: forge_workspace::protocol::PromptSource::You,
                text: "hello".to_owned(),
                attachments: vec![forge_primitives::ImageAttachment {
                    data: "iVBORw0KGgo=".to_owned(),
                    mime_type: "image/png".to_owned(),
                }],
            }],
            voice_listeners: 0,
        };
        let encoded = serde_json::to_string(&state).expect("the state encodes");
        assert!(
            !encoded.contains("iVBORw0KGgo=") && !encoded.contains("\"attachments\""),
            "the queue a view reads carries no attachment payload: {encoded}",
        );
        assert!(
            encoded.contains("\"uuid\":\"p1\"") && encoded.contains("\"text\":\"hello\""),
            "and carries the row a view draws: {encoded}",
        );
    }

    /// **The read half of the cron pairing rides the page.** A cron row drawn
    /// while its delivery frame lands is titled by the schedule; the same row
    /// replayed from a transcript after a reload has no frame to read. The
    /// conversation read therefore carries the pair for exactly the rows it
    /// serves, and this pins that a page of frames gets it (#1815).
    #[test]
    fn a_conversation_page_carries_the_schedule_each_fired_prompt_came_from() {
        let dir = tempfile::tempdir().expect("tempdir");
        let support = dir.path().join("app-support");
        std::fs::create_dir_all(&support).expect("app-support");
        {
            // The fleet opens this config dir's own store; seed the same path.
            let db_path = forge_workspace::store::path_for(&support, dir.path());
            std::fs::create_dir_all(db_path.parent().expect("stores/")).expect("stores/");
            let db = forge_workspace::store::Db::open(&db_path).expect("open db");
            forge_workspace::store::cron_names::put(
                &db,
                "p-reload",
                &forge_primitives::cron::CronName {
                    cron_id: "c1".to_owned(),
                    description: Some("Morning summary".to_owned()),
                },
                std::time::SystemTime::now(),
            )
            .expect("write the pairing");
            forge_workspace::store::cron_names::put(
                &db,
                "a-1",
                &forge_primitives::cron::CronName {
                    cron_id: "not-a-fire".to_owned(),
                    description: Some("an assistant row's id".to_owned()),
                },
                std::time::SystemTime::now(),
            )
            .expect("write a pairing under a non-prompt id too");
        }
        let fleet =
            crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
        let page = vec![super::TurnWire {
            key: None,
            messages: vec![
                serde_json::json!({
                    "type": "user",
                    "uuid": "p-reload",
                    "message": {
                        "role": "user",
                        "content": [{ "type": "text", "text": "[Cron]\n\nsummarise overnight CI" }],
                    },
                }),
                serde_json::json!({
                    "type": "assistant",
                    "uuid": "a-1",
                    "message": { "role": "assistant", "content": [] },
                }),
                serde_json::json!({
                    "type": "user",
                    "uuid": "p-other",
                    "message": {
                        "role": "user",
                        "content": [{ "type": "text", "text": "my own words" }],
                    },
                }),
            ],
        }];

        let seat = forge_primitives::SessionSlot::lead("TestOrg", "proj");
        let names = super::cron_names_of(&fleet.surface(), &seat, &page);
        assert_eq!(
            names.get("p-reload").and_then(|name| name.description.as_deref()),
            Some("Morning summary"),
            "the fired row's schedule rides the read, as its delivery frame once did",
        );
        assert!(
            !names.contains_key("p-other"),
            "a page carries the pairings its own fired rows need and no others",
        );
        assert!(
            !names.contains_key("a-1"),
            "a pairing stored under an id an ASSISTANT row carries is not a fired prompt, so the \
             type filter must keep it out of the map",
        );
    }

    /// The transcript rows one turn leaves: what the user wrote, what the
    /// assistant said, and the result that closes it. The result carries a
    /// `uuid`, which is what the fold names the turn by - so a turn's report
    /// has a key, and a key is what a page's cursor is.
    fn a_turns_rows(turn: usize) -> String {
        format!(
            r#"{{"type":"user","uuid":"u{turn}","message":{{"role":"user","content":"turn {turn}"}}}}
{{"type":"assistant","uuid":"a{turn}","message":{{"id":"m{turn}","role":"assistant","model":"claude-opus-5","content":[{{"type":"text","text":"reply {turn}"}}]}}}}
{{"type":"result","uuid":"r{turn}","subtype":"success","duration_ms":1,"duration_api_ms":1,"is_error":false,"num_turns":1,"session_id":"s"}}"#
        )
    }

    /// **The row states the check its diff was read against.** The store can
    /// hold a newer measurement (a bypass answered while a fetch landed), and
    /// taking the row from that side would leave the record disagreeing with
    /// its own diff.
    #[test]
    fn a_record_states_one_base_across_its_diff_and_its_row() {
        use forge_primitives::git_diff::BaseState;
        use forge_workspace::env::git_diff::BaseRefresh;

        let refresh = BaseRefresh {
            state: BaseState::Unverified,
            base_ref: None,
            error: None,
            cut: false,
            unanswered: false,
            at: std::time::Instant::now(),
        };
        assert_eq!(
            super::stated_base(Some(&refresh), Some(BaseState::Current), Some("stale words")),
            (Some(BaseState::Unverified), None),
            "the read's own check is the frame's base, not the store's newer measurement",
        );
        assert_eq!(
            super::stated_base(None, Some(BaseState::Current), Some("last check")),
            (Some(BaseState::Current), Some("last check")),
            "and a read that did not refresh states what the store holds",
        );
        assert!(super::checked_refresh(Some(&refresh)).is_some(), "an answered run is a check");
        let mut failed = refresh.clone();
        failed.error = Some("fatal: could not resolve host".to_owned());
        assert_eq!(
            super::diff_base_error(Some(&failed), Some("last check")),
            Some("fatal: could not resolve host"),
            "a run that ran and failed names its own words",
        );

        for placeholder in [
            BaseRefresh {
                cut: true,
                error: Some("the fetch did not finish inside the 10s it was given".to_owned()),
                ..refresh.clone()
            },
            BaseRefresh {
                unanswered: true,
                error: Some("the base refresh ended without an answer".to_owned()),
                ..refresh.clone()
            },
        ] {
            assert!(
                super::checked_refresh(Some(&placeholder)).is_none(),
                "a placeholder run is not a check",
            );
            assert_eq!(
                super::stated_base(
                    Some(&placeholder),
                    Some(BaseState::Current),
                    Some("last check")
                ),
                (Some(BaseState::Current), Some("last check")),
                "so the row falls back to the store, never to forge's own words",
            );
            assert_eq!(
                super::diff_base_error(Some(&placeholder), Some("git's own failure")),
                Some("git's own failure"),
                "and the diff's reason is git's words, never forge's own",
            );
        }
        assert_eq!(
            super::diff_base_error(None, Some("last check")),
            None,
            "a read with no refresh names no failure of its own",
        );
    }

    /// A value whose serialisation fails, which is the only case a message
    /// can reach the page as anything but itself.
    struct WillNotSerialise;

    impl Serialize for WillNotSerialise {
        fn serialize<S: serde::Serializer>(&self, _serializer: S) -> Result<S::Ok, S::Error> {
            Err(serde::ser::Error::custom("this value cannot be a frame"))
        }
    }

    /// **A null frame throws the fold that receives it.** The page reads
    /// `type` off every frame, so a message that will not serialise must
    /// arrive as a notice naming the failure rather than as `Value::Null`.
    #[test]
    fn a_frame_that_will_not_serialise_draws_a_notice_rather_than_null() {
        let frame = frame_of(&WillNotSerialise, Some("a-session"));
        assert!(
            !frame.is_null(),
            "the null this replaces is the value the fold throws on: {frame}",
        );
        assert_eq!(frame.get("type").and_then(Value::as_str), Some("system"), "{frame}");
        assert_eq!(
            frame.get("subtype").and_then(Value::as_str),
            Some("forge_notice"),
            "the frame is the notice the client already draws: {frame}",
        );
        let Some(text) = frame.get("text").and_then(Value::as_str) else {
            panic!("the notice says what failed: {frame}");
        };
        assert!(text.contains("could not be serialised"), "the failure is named: {frame}");

        let held = frame_of(&serde_json::json!({ "type": "widget_state", "n": 1 }), None);
        assert_eq!(
            held,
            serde_json::json!({ "type": "widget_state", "n": 1 }),
            "a frame that serialises is carried unchanged",
        );
    }

    /// The messages and the fold's own answer for them, over a transcript of
    /// `rows`, read the way the transport reads them.
    fn a_conversation(rows: &[&str]) -> (Vec<Message>, Rendered) {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.seed_transcript("TestOrg", "proj", "lead", rows).expect("the transcript seeds");
        let seat = fixture_seat();
        let surface = fleet.surface();
        let cwd = surface.roster().cwd_for(&seat).expect("the seat has a directory");

        let messages = surface.conversation(&seat, &cwd).messages;
        let rendered = crate::transcript::render(&messages, false);
        (messages, rendered)
    }

    /// `turns` finished turns, as the transcript rows they are.
    fn turns_of(turns: usize) -> Vec<String> {
        (0..turns).map(a_turns_rows).collect()
    }

    /// The words a turn opened on, read off the messages the page carried.
    ///
    /// A turn's identity is its messages now, so this is what a reader checks
    /// one by: the evidence cannot be a position, because `page` hands back
    /// values and no value is ever `ptr::eq` to the original.
    fn opened_on(turn: &TurnWire) -> String {
        turn.messages
            .iter()
            .find_map(|frame| frame["message"]["content"][0]["text"].as_str().map(str::to_owned))
            .unwrap_or_default()
    }

    /// The words every turn of a conversation opened on, oldest first.
    fn every_turn(messages: &[Message], rendered: &Rendered) -> Vec<String> {
        page(messages, rendered, 0, 0, None, u32::MAX).turns.iter().map(opened_on).collect()
    }

    /// The bytes a page's turns occupy on the wire, which is what a budget
    /// counts.
    fn wire_bytes(turns: &[TurnWire]) -> usize {
        turns
            .iter()
            .map(|turn| serde_json::to_string(turn).expect("a turn serialises").len() + 1)
            .sum()
    }

    /// **A first page held to a byte budget is the newest WHOLE turns that
    /// fit, and it is cut nowhere else.** No turn is split, the order is the
    /// conversation's, and the cursor still names the message the first kept
    /// turn opened at - so the page above it is the page above that turn.
    #[test]
    fn a_page_over_the_budget_is_the_newest_whole_turns_that_fit() {
        let rows = turns_of(8);
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);
        let every = page(&messages, &rendered, 0, 0, None, u32::MAX);
        let newest_two = &every.turns[every.turns.len() - 2..];
        let budget = wire_bytes(newest_two);

        let bounded =
            page_within(&messages, &rendered, 0, 0, None, SUBSCRIBE_TURNS, budget, "test", None);

        assert_eq!(
            bounded.turns.iter().map(opened_on).collect::<Vec<_>>(),
            newest_two.iter().map(opened_on).collect::<Vec<_>>(),
            "the budget keeps the newest whole turns and nothing below them",
        );
        assert_eq!(
            bounded.cursor,
            page(&messages, &rendered, 0, 0, None, 2).cursor,
            "the handle above the page names the message the first kept turn opened at",
        );
    }

    /// **A newest turn larger than the whole budget is still served.** A
    /// budget that answered with no page would be a seat drawn empty, and one
    /// that split the turn would hand a client the tail of a turn with no way
    /// to tell that is what it has.
    #[test]
    fn a_page_keeps_the_newest_turn_when_it_alone_is_over_the_budget() {
        let rows = turns_of(3);
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);
        let every = every_turn(&messages, &rendered);

        let bounded =
            page_within(&messages, &rendered, 0, 0, None, SUBSCRIBE_TURNS, 1, "test", None);

        assert_eq!(bounded.turns.len(), 1, "a budget under one turn still answers with one");
        assert_eq!(opened_on(&bounded.turns[0]), every[every.len() - 1], "and it is the newest");
        assert!(bounded.cursor.is_some(), "with history above it to ask for");
    }

    /// **A budget larger than the page changes nothing.** The count is what a
    /// client asked for, and the bytes only ever cut the page it fits in.
    #[test]
    fn a_page_under_the_budget_is_the_page_its_count_asks_for() {
        let rows = turns_of(30);
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);
        let counted = page(&messages, &rendered, 0, 0, None, SUBSCRIBE_TURNS);
        let budget = wire_bytes(&counted.turns);

        let bounded =
            page_within(&messages, &rendered, 0, 0, None, SUBSCRIBE_TURNS, budget, "test", None);

        assert_eq!(
            bounded.turns.len(),
            counted.turns.len(),
            "a page that fits its budget is the page the count asked for",
        );
        assert_eq!(bounded.cursor, counted.cursor, "on the same handle");
    }

    /// A wire frame as a page carries it.
    fn a_frame(text: &str) -> Message {
        serde_json::from_str(text).expect("a wire frame decodes")
    }

    /// A prompt, which is what opens a turn.
    fn opens_turn() -> Message {
        opens_turn_at(1)
    }

    /// The same, under an id and a clock of its own.
    fn opens_turn_at(seq: u32) -> Message {
        a_frame(&format!(
            r#"{{"type":"user","uuid":"u-{seq}","session_id":"s","timestamp":"2026-10-24T09:00:{seq:02}.000000Z","message":{{"role":"user","content":[{{"type":"text","text":"turn {seq}"}}]}}}}"#
        ))
    }

    /// The result that closes a turn.
    fn ends_turn(seq: u32) -> Message {
        a_frame(&format!(
            r#"{{"type":"result","uuid":"r-{seq}","session_id":"s","subtype":"success","is_error":false,"duration_ms":1,"duration_api_ms":1,"num_turns":1}}"#
        ))
    }

    /// One thinking-token counter, as the CLI emits them.
    fn counter(seq: u32, delta: i64) -> Message {
        counter_of(seq, delta, None)
    }

    /// The same, under a sub-agent's dispatch when `dispatch` names one.
    fn counter_of(seq: u32, delta: i64, dispatch: Option<&str>) -> Message {
        let parent =
            dispatch.map_or_else(String::new, |id| format!(r#","parent_tool_use_id":"{id}""#));
        a_frame(&format!(
            r#"{{"type":"system","subtype":"thinking_tokens","estimated_tokens":{delta},"estimated_tokens_delta":{delta},"uuid":"t-{seq}","session_id":"s","timestamp":"2026-10-24T09:00:00.{seq:06}Z"{parent}}}"#
        ))
    }

    /// An assistant message, which draws a row of its own between runs.
    fn says(seq: u32) -> Message {
        a_frame(&format!(
            r#"{{"type":"assistant","uuid":"a-{seq}","session_id":"s","timestamp":"2026-10-24T09:00:00.{seq:06}Z","message":{{"id":"m-{seq}","role":"assistant","model":"claude-opus-5","content":[{{"type":"text","text":"reply"}}]}}}}"#
        ))
    }

    /// A tool result, which draws nothing and is no counter either.
    fn answers(seq: u32) -> Message {
        a_frame(&format!(
            r#"{{"type":"user","uuid":"r-{seq}","session_id":"s","timestamp":"2026-10-24T09:00:00.{seq:06}Z","message":{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"call-1","content":"ok"}}]}}}}"#
        ))
    }

    /// The counters a page's frames carry, as `(delta, folded_frames)`.
    fn counters_on(page: &Page) -> Vec<(i64, Option<u64>)> {
        page.turns
            .iter()
            .flat_map(|turn| turn.messages.iter())
            .filter(|frame| frame["subtype"] == "thinking_tokens")
            .map(|frame| {
                (
                    frame["estimated_tokens_delta"].as_i64().expect("a delta"),
                    frame["folded_frames"].as_u64(),
                )
            })
            .collect()
    }

    /// **A turn's counter storm is one frame per run on a page.**
    ///
    /// The fold reads a counter for its growth and its stamp and draws no row
    /// for it, so the page carries the run's sum and newest stamp once - and
    /// the figure a reader sees is the same number either way. A run of one is
    /// left exactly as the CLI sent it.
    #[test]
    fn a_turns_counter_runs_are_one_frame_each_on_a_page() {
        let messages = vec![
            opens_turn(),
            counter(1, 10),
            counter(2, 20),
            counter(3, 5),
            says(4),
            counter(5, 7),
            counter(6, 3),
            answers(7),
            counter(8, 9),
        ];
        let rendered = crate::transcript::render(&messages, true);

        let page = page(&messages, &rendered, 0, 0, None, u32::MAX);

        assert_eq!(page.turns.len(), 1, "the frames are one live turn");
        let served = &page.turns[0].messages;
        assert_eq!(served.len(), 6, "nine frames, three runs folded: {served:?}");
        assert_eq!(
            counters_on(&page),
            vec![(35, Some(3)), (10, Some(2)), (9, None)],
            "each run is its summed delta and its count, and a run of one is untouched",
        );
        let first = served.iter().find(|frame| frame["folded_frames"] == 3).expect("the first run");
        assert_eq!(first["uuid"], "t-3", "the surviving frame is the newest of its run");
        assert_eq!(
            first["folded_from"], "2026-10-24T09:00:00.000001Z",
            "and it names where its run starts, so a landing knows which counters it covers",
        );
        assert_eq!(
            first["timestamp"], "2026-10-24T09:00:00.000003Z",
            "so the turn's clock reads the newest stamp, as it did over the raw run",
        );
        assert_eq!(
            first["estimated_tokens"], 5,
            "the running value is the newest frame's own, the same exchange the live fold makes",
        );
        assert_eq!(
            served.last().map(|frame| &frame["timestamp"]),
            Some(&serde_json::json!("2026-10-24T09:00:00.000008Z")),
            "the turn still ends on the instant its newest frame was stamped",
        );
    }

    /// **A run is one dispatch's.** A sub-agent's counters are the subagents
    /// surface's, and the chat fold skips them whole - folding them into the
    /// turn's own run, or into another sub-agent's, would move the running
    /// figure and the clock.
    #[test]
    fn a_dispatched_counters_do_not_fold_into_the_turns_own() {
        let messages = vec![
            opens_turn(),
            counter_of(1, 10, Some("tool-1")),
            counter_of(2, 20, Some("tool-1")),
            counter_of(3, 7, Some("tool-2")),
            counter(4, 5),
        ];
        let rendered = crate::transcript::render(&messages, true);

        let page = page(&messages, &rendered, 0, 0, None, u32::MAX);

        assert_eq!(
            counters_on(&page),
            vec![(30, Some(2)), (7, None), (5, None)],
            "each dispatch keeps its own run, and the turn's own counter stays its own",
        );
        let folded = page.turns[0]
            .messages
            .iter()
            .find(|frame| frame["folded_frames"] == 2)
            .expect("the first dispatch's run");
        assert_eq!(folded["parent_tool_use_id"], "tool-1", "under the dispatch it ran as");
        assert_eq!(
            folded["folded_from"], "2026-10-24T09:00:00.000001Z",
            "and its range is its own run's, not the turn's",
        );
    }

    /// **A page above the newest turn carries no cut debt.** The debt belongs
    /// to the turn open at the window's end, so a page that stops below it
    /// leaves every counter as it was.
    #[test]
    fn a_before_page_does_not_carry_the_cut_debt() {
        let messages =
            vec![opens_turn_at(1), counter(2, 10), ends_turn(3), opens_turn_at(4), counter(5, 20)];
        let rendered = crate::transcript::render(&messages, false);
        let newest = page_within(&messages, &rendered, 0, 1_000, None, 1, usize::MAX, "test", None);
        assert_eq!(counters_on(&newest), vec![(1_020, None)], "the newest page carries the debt");
        let cursor = newest.cursor.expect("a page above the newest turn");
        let above =
            page_within(&messages, &rendered, 0, 1_000, Some(&cursor), 2, usize::MAX, "test", None);

        assert_eq!(
            counters_on(&above),
            vec![(10, None)],
            "the page above the newest turn serves its own counters, not the debt",
        );
    }

    /// **The debt lands on the newest turn only.** A page carrying several
    /// turns folds the cut's growth into the first counter of the turn open at
    /// its end, and leaves every older turn's counters alone.
    #[test]
    fn the_cut_debt_lands_on_the_newest_turn_only() {
        let messages =
            vec![opens_turn_at(1), counter(2, 10), ends_turn(3), opens_turn_at(4), counter(5, 20)];
        let rendered = crate::transcript::render(&messages, false);
        let served =
            page_within(&messages, &rendered, 0, 1_000, None, u32::MAX, usize::MAX, "test", None);

        assert_eq!(
            counters_on(&served),
            vec![(10, None), (1_020, None)],
            "the older turn keeps its own counter and the newest carries the debt once",
        );
    }

    /// **A leading run with no proven start is never folded.** A turn whose
    /// first carried frame is a counter is the window's front cutting inside
    /// a running turn, so the run's true start is not in the page - its
    /// counters are served as themselves, and a later run still folds.
    #[test]
    fn a_leading_counter_run_of_a_cut_turn_is_not_folded() {
        let messages = vec![
            counter(1, 10),
            counter(2, 20),
            says(3),
            counter(4, 7),
            counter(5, 3),
            answers(6),
            counter(7, 5),
        ];
        let rendered = crate::transcript::render(&messages, true);

        let page = page(&messages, &rendered, 0, 0, None, u32::MAX);

        assert_eq!(
            counters_on(&page),
            vec![(10, None), (20, None), (10, Some(2)), (5, None)],
            "the leading pair stays raw and the run after the assistant folds",
        );
    }

    /// **A frame with no stamp breaks a run.** A fold's coverage is a stamp
    /// range, so a frame that cannot be placed is served as itself and the
    /// stamped frames either side fold among themselves.
    #[test]
    fn a_counter_frame_with_no_stamp_breaks_the_run() {
        let unstamped = |seq: u32, delta: i64| {
            a_frame(&format!(
                r#"{{"type":"system","subtype":"thinking_tokens","estimated_tokens":{delta},"estimated_tokens_delta":{delta},"uuid":"t-{seq}","session_id":"s"}}"#
            ))
        };
        let messages = vec![opens_turn(), unstamped(1, 10), counter(2, 20), counter(3, 5)];
        let rendered = crate::transcript::render(&messages, true);

        let page = page(&messages, &rendered, 0, 0, None, u32::MAX);

        assert_eq!(
            counters_on(&page),
            vec![(10, None), (25, Some(2))],
            "the unstamped frame is served, and the stamped pair after it folds",
        );
        let folded = page.turns[0]
            .messages
            .iter()
            .find(|frame| frame["folded_frames"] == 2)
            .expect("the stamped pair");
        assert_eq!(
            folded["folded_from"], "2026-10-24T09:00:00.000002Z",
            "its range starts at the first frame it can place",
        );
    }

    /// **The budget still counts the bytes a page carried before the fold.**
    /// Folding after the cut is what keeps a capped page's turns the same as
    /// they were; folding first would fit more of them and change what a page
    /// draws.
    #[test]
    fn the_budget_cuts_on_the_bytes_a_page_carried_before_the_fold() {
        let mut messages = Vec::new();
        let mut seq = 0;
        for turn in 1..=4 {
            messages.push(opens_turn_at(turn));
            for _ in 0..40 {
                seq += 1;
                messages.push(counter(seq, 10));
            }
            messages.push(ends_turn(turn));
        }
        let rendered = crate::transcript::render(&messages, false);
        let ranges = turn_ranges(&messages, &rendered.turns);
        let raw_turn = |&(from, to, key): &(usize, usize, Option<&str>)| {
            serde_json::to_string(&turn_wire(&messages, from, to, key, &rendered.endings))
                .expect("a turn serialises")
                .len()
                + 1
        };
        let newest_two: usize = ranges[ranges.len() - 2..].iter().map(raw_turn).sum();

        let bounded = page_within(
            &messages,
            &rendered,
            0,
            0,
            None,
            SUBSCRIBE_TURNS,
            newest_two,
            "test",
            None,
        );

        assert_eq!(bounded.turns.len(), 2, "the same two turns the raw budget kept");
        assert!(
            wire_bytes(&bounded.turns) < newest_two,
            "and the page is served folded: {} bytes under a {newest_two}-byte cut",
            wire_bytes(&bounded.turns),
        );
    }

    /// A turn whose structured tool result is `pad` bytes of payload.
    fn a_big_result_rows(turn: usize, pad: usize) -> String {
        format!(
            r#"{{"type":"user","uuid":"u{turn}","message":{{"role":"user","content":"turn {turn}"}}}}
{{"type":"assistant","uuid":"a{turn}","message":{{"id":"m{turn}","role":"assistant","model":"claude-opus-5","content":[{{"type":"text","text":"reply {turn}"}}]}}}}
{{"type":"user","uuid":"t{turn}","session_id":"s","message":{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"call-{turn}","content":"ok"}}]}},"toolUseResult":{{"payload":"{}"}}}}
{{"type":"result","uuid":"r{turn}","subtype":"success","duration_ms":1,"duration_api_ms":1,"is_error":false,"num_turns":1,"session_id":"s"}}"#,
            "x".repeat(pad),
        )
    }

    /// **The budget counts what the cap serves, not the raw record.** A
    /// seat with a few multi-MB records would otherwise keep one turn where
    /// the capped page fits several, and the first frame's size would follow
    /// a payload the client never receives.
    #[test]
    fn the_budget_counts_the_capped_record_it_serves() {
        let rows: Vec<String> = (0..4).map(|turn| a_big_result_rows(turn, 20_000)).collect();
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);
        let seat = fixture_seat();

        let raw =
            page_within(&messages, &rendered, 0, 0, None, SUBSCRIBE_TURNS, 2_048, "test", None);
        assert_eq!(raw.turns.len(), 1, "a raw record of 20 KB leaves one turn in 2 KB");

        let capped = page_within(
            &messages,
            &rendered,
            0,
            0,
            None,
            SUBSCRIBE_TURNS,
            2_048,
            "test",
            Some((1_024, &seat)),
        );
        assert!(
            capped.turns.len() > 1,
            "the capped records are small, so the same budget fits more: {} turns",
            capped.turns.len(),
        );
    }

    /// How many task endings the frames name `call` in.
    fn endings_in(turn: &TurnWire, call: &str) -> usize {
        turn.messages
            .iter()
            .filter(|frame| {
                frame["message"]["content"].as_array().is_some_and(|blocks| {
                    blocks.iter().any(|block| {
                        block["type"] == "queued_command"
                            && block["commandMode"] == "task-notification"
                            && block["prompt"].as_str().is_some_and(|text| text.contains(call))
                    })
                })
            })
            .count()
    }

    /// Whether a turn's frames make the call `id`.
    fn holds_call(turn: &TurnWire, id: &str) -> bool {
        turn.messages.iter().any(|frame| {
            frame["message"]["content"].as_array().is_some_and(|blocks| {
                blocks.iter().any(|block| {
                    matches!(block["type"].as_str(), Some("tool_use" | "server_tool_use"))
                        && block["id"] == id
                })
            })
        })
    }

    /// A page's client folds one turn at a time, so an ending that sits in
    /// another turn is an ending it cannot reach - and about three notices in
    /// ten on this machine sit exactly there.
    ///
    /// Both carriers are carried into the call's turn: the copy is a frame of
    /// its own, and the row the CLI persisted stays where it is.
    #[test]
    fn the_ending_is_carried_into_the_call_turn() {
        for (rows, call, said) in [
            (
                crate::fixtures::SAME_TURN_USER_ROW,
                "call_da4c7ee117d14d48ba99e036",
                "Background command \"Watch the account-lifecycle CI run\" failed with exit code 1",
            ),
            (
                crate::fixtures::CROSS_TURN_ATTACHMENT,
                "call_17c05f64497746b0ac450728",
                "Background command \"Restart the harness with a long terminate window\" failed with exit code 100",
            ),
        ] {
            let (messages, rendered) = a_conversation(rows);
            let turns = all_turns(&messages, &rendered);

            // **One turn holds the call, so one turn can end it.** A
            // cross-turn notice is carried as well as kept, so the
            // conversation holds the ending twice - the row the CLI wrote it
            // in and the copy - and only the copy is inside a turn that makes
            // the call.
            let holding: Vec<&TurnWire> =
                turns.iter().filter(|turn| holds_call(turn, call)).collect();
            assert_eq!(holding.len(), 1, "one turn holds {call}, so one can end it");
            let held = holding[0];
            assert!(
                endings_in(held, call) > 0,
                "and the ending rides that turn, where a per-turn fold reads it",
            );
            assert!(
                held.messages.iter().any(|frame| {
                    frame["message"]["content"].as_array().is_some_and(|blocks| {
                        blocks.iter().any(|block| {
                            block["type"] == "queued_command"
                                && block["prompt"].as_str().is_some_and(|text| text.contains(said))
                        })
                    })
                }),
                "reaching the wire as the block a view reads, not as the row's own text",
            );
            assert!(
                turns.iter().flat_map(|turn| turn.messages.iter()).all(|frame| {
                    frame["message"]["content"].as_array().is_none_or(|blocks| {
                        blocks.iter().all(|block| {
                            !(block["type"] == "text"
                                && block["text"]
                                    .as_str()
                                    .is_some_and(|text| text.contains("<task-notification>")))
                        })
                    })
                }),
                "and no turn carries the raw notice as the reader's own words",
            );

            // **And the carry survives a page, which is the path a client
            // walking history actually reads.** `page` cuts its own window
            // from the same fold, and a carry that only `all_turns` made would
            // leave the ending unreachable where scrolling looks for it - a
            // hole no test above this line would show.
            let mut before: Option<String> = None;
            let mut held_on_a_page = 0_usize;
            let mut ended_where_held = 0_usize;
            loop {
                let page = page(&messages, &rendered, 0, 0, before.as_deref(), 1);
                for turn in &page.turns {
                    if holds_call(turn, call) {
                        held_on_a_page += 1;
                        if endings_in(turn, call) > 0 {
                            ended_where_held += 1;
                        }
                    }
                }
                match page.cursor {
                    Some(cursor) if Some(&cursor) != before.as_ref() => before = Some(cursor),
                    _ => break,
                }
            }
            assert!(held_on_a_page > 0, "the walk reaches the turn that holds the call");
            assert_eq!(
                ended_where_held, held_on_a_page,
                "and every copy of that turn a page carries ends it",
            );
        }
    }

    /// **Every row a page serves carries the stamp of the file row it came
    /// from**, for every kind a turn holds: the prompt, a hoisted queued
    /// command, an assistant message, the hook summary and the compaction
    /// boundary.
    ///
    /// The stamp is what a page's clock is read from - an open page anchors a
    /// running turn on the first row carrying one - so a kind that reaches the
    /// wire without its own stamp is a turn that draws no age, or the wrong
    /// one, and nothing downstream can recover a fact the read dropped.
    #[test]
    fn every_kind_a_turn_holds_reaches_a_page_stamped() {
        let (messages, rendered) = a_conversation(&[
            r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"hi"}]},"uuid":"u1","sessionId":"s","timestamp":"2026-10-15T05:18:05.217Z"}"#,
            r#"{"type":"attachment","attachment":{"type":"queued_command","prompt":"queued words","commandMode":"prompt"},"uuid":"q1","sessionId":"s","timestamp":"2026-10-15T05:19:00.100Z"}"#,
            r#"{"type":"queue-operation","operation":"remove","timestamp":"2026-10-15T05:19:03.000Z","sessionId":"s","content":"queued words"}"#,
            r#"{"type":"assistant","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"working"}]},"uuid":"a1","sessionId":"s","timestamp":"2026-10-15T05:19:05.544Z"}"#,
            r#"{"type":"system","subtype":"stop_hook_summary","hookCount":1,"hookInfos":[],"hookErrors":[],"hasOutput":false,"level":"suggestion","preventedContinuation":false,"stopReason":"","toolUseID":"tu1","uuid":"h1","sessionId":"s","timestamp":"2026-10-15T05:19:40.632Z"}"#,
            r#"{"type":"system","subtype":"compact_boundary","uuid":"b1","sessionId":"s","compactMetadata":{"trigger":"auto","preTokens":1018466,"postTokens":19526},"timestamp":"2026-10-15T05:20:11.004Z"}"#,
        ]);
        let served = page(&messages, &rendered, 0, 0, None, u32::MAX);

        let frames: Vec<&Value> =
            served.turns.iter().flat_map(|turn| turn.messages.iter()).collect();
        let kinds: Vec<String> = frames
            .iter()
            .map(|frame| {
                let subtype = frame["subtype"].as_str().unwrap_or("");
                if subtype.is_empty() {
                    frame["type"].as_str().unwrap_or("?").to_owned()
                } else {
                    format!("{}/{subtype}", frame["type"].as_str().unwrap_or("?"))
                }
            })
            .collect();
        assert_eq!(
            kinds,
            vec![
                "user",
                "user",
                "assistant",
                "system/stop_hook_summary",
                "system/compact_boundary",
            ],
            "the page carries every kind the read kept - the denominator this test asserts over",
        );

        let stamps: Vec<Option<&str>> =
            frames.iter().map(|frame| frame["timestamp"].as_str()).collect();
        assert_eq!(
            stamps,
            vec![
                Some("2026-10-15T05:18:05.217Z"),
                Some("2026-10-15T05:19:03.000Z"),
                Some("2026-10-15T05:19:05.544Z"),
                Some("2026-10-15T05:19:40.632Z"),
                Some("2026-10-15T05:20:11.004Z"),
            ],
            "each row carries the row's own stamp, in the row's own order",
        );
    }

    /// A page of no turns ended where it began - an empty page whose cursor
    /// named the turn it had already opened at - so a client walking back
    /// asked for it forever.
    #[test]
    fn a_page_of_no_turns_still_walks_backwards() {
        let rows = turns_of(6);
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);

        // Reached the way a client reaches one, from the cursor below it.
        let lower = page(&messages, &rendered, 0, 0, None, 2);
        let cursor = lower.cursor.expect("there is a page above this one");

        let above = page(&messages, &rendered, 0, 0, Some(&cursor), 0);

        assert!(!above.turns.is_empty(), "a page carries turns rather than none at all");
        assert_ne!(
            above.cursor.as_deref(),
            Some(cursor.as_str()),
            "and it moves rather than naming the turn it already opened at",
        );
    }

    /// The conversation's opening rows come before its first turn, so a page
    /// starting at that turn leaves them above every page, where no walk
    /// reaches them.
    #[test]
    fn the_rows_before_the_first_turn_ride_the_first_page() {
        let (messages, rendered) = a_conversation(&[
            r#"{"type":"assistant","uuid":"a0","message":{"id":"m0","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"a delivery arrived"}]}}"#,
            r#"{"type":"user","uuid":"u1","message":{"role":"user","content":"hello"},"session_id":"s"}"#,
            r#"{"type":"result","uuid":"r1","subtype":"success","duration_ms":1,"duration_api_ms":1,"is_error":false,"num_turns":1,"session_id":"s"}"#,
        ]);

        let first = page(&messages, &rendered, 0, 0, None, 10);

        assert_eq!(first.turns.len(), 1, "precondition: this conversation is one turn");
        assert_eq!(
            first.turns[0].messages.first().map(|frame| frame["uuid"].clone()),
            Some(Value::String("a0".to_owned())),
            "the row before the first turn rides that turn rather than sitting above every page: \
             {:?}",
            first.turns[0].messages,
        );
    }

    /// A conversation with content and no turn boundary still pages.
    ///
    /// The interrupt marker is a user row the fold opens no turn at, so a
    /// session holding only its own interrupted exchange has no boundary at
    /// all. A page that listed no turns AND said `cursor: null` would tell a
    /// client there is nothing above a conversation it has never seen - and
    /// the walk would stop there, with the whole of it unreachable.
    #[test]
    fn a_conversation_with_no_turns_still_pages() {
        // The interrupt marker is a user row the fold opens no turn at, so
        // this conversation has content and no boundary: the shape the page's
        // fallback is for. A delivery's envelope row is a boundary, which is
        // why it cannot stand in for one here.
        let (messages, rendered) = a_conversation(&[
            r#"{"type":"user","uuid":"u1","message":{"role":"user","content":"[Request interrupted by user]"},"session_id":"s"}"#,
            r#"{"type":"assistant","uuid":"a1","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"morning"}]}}"#,
            r#"{"type":"result","uuid":"r1","subtype":"success","duration_ms":1,"duration_api_ms":1,"is_error":false,"num_turns":1,"session_id":"s"}"#,
        ]);

        assert!(
            rendered.turns.is_empty(),
            "precondition: nothing in this conversation opens a turn",
        );
        assert!(!messages.is_empty(), "precondition: and it has content");

        let page = page(&messages, &rendered, 0, 0, None, 10);

        assert_eq!(page.turns.len(), 1, "the conversation rides one turn rather than none");
        assert_eq!(page.turns[0].messages.len(), messages.len(), "and that turn carries all of it");
        assert_eq!(page.cursor, None, "with nothing above it, which is the one honest null");
    }

    /// A user row carrying two blocks opens one turn, not two.
    ///
    /// Two blocks that each open a turn would put two spans at one message. A
    /// cursor names a message and resolves to the first match, so a second
    /// boundary at that index would hand a walker a page of its own before it
    /// moved - an empty one, since the two boundaries are the same message.
    #[test]
    fn a_user_row_with_two_blocks_opens_one_turn() {
        let (messages, rendered) = a_conversation(&[
            r#"{"type":"user","uuid":"u1","message":{"role":"user","content":[{"type":"text","text":"one"},{"type":"text","text":"two"}]},"session_id":"s"}"#,
            r#"{"type":"assistant","uuid":"a1","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"reply"}]}}"#,
            r#"{"type":"result","uuid":"r1","subtype":"success","duration_ms":1,"duration_api_ms":1,"is_error":false,"num_turns":1,"session_id":"s"}"#,
        ]);

        assert_eq!(rendered.turns.len(), 2, "precondition: the fold opens a turn per block");
        assert_eq!(
            rendered.turns[0].opens_at, rendered.turns[1].opens_at,
            "precondition: and both open at the one message that carried them",
        );

        let page = page(&messages, &rendered, 0, 0, None, 10);

        assert_eq!(page.turns.len(), 1, "one message is one turn on the page");
        assert_eq!(
            page.turns[0].messages.len(),
            messages.len(),
            "carrying the whole conversation, since both blocks open at its first row",
        );
        assert_eq!(page.cursor, None, "and the walk is not sent after a page that cannot move");
    }

    /// A page opens on a turn, and consecutive pages meet without a gap.
    #[test]
    fn a_page_opens_on_a_turn_and_the_pages_meet_exactly() {
        let rows = turns_of(50);
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);
        let first = page(&messages, &rendered, 0, 0, None, 10);

        // A page is the turns it was asked for, each opening where a turn
        // does. A page beginning anywhere else hands a client the tail of one
        // turn and no way to tell that is what it has.
        assert_eq!(first.turns.len(), 10, "a page is the turns it was asked for");
        // The page is the NEWEST ten, and it lists them in conversation order: a
        // client walking back gets each page in reading order and asks for the ones
        // above it.
        assert_eq!(
            opened_on(first.turns.first().expect("ten turns")),
            "turn 40",
            "the page reaches back exactly as far as it was asked for, no further",
        );
        assert_eq!(
            opened_on(first.turns.last().expect("ten turns")),
            "turn 49",
            "and it ends on the newest turn of the conversation",
        );

        // And the next page must reach back to where this one began. It may
        // repeat turns - the client keys them and drops the repeats - but it
        // may never skip one, because a skipped turn is history the reader has
        // no way to ask for again.
        let all = every_turn(&messages, &rendered);
        let second = page(&messages, &rendered, 0, 0, first.cursor.as_deref(), 10);
        let second_turns: Vec<String> = second.turns.iter().map(opened_on).collect();

        assert!(!second_turns.is_empty(), "asking for more turns returns some");
        let above = all
            .iter()
            .position(|held| *held == opened_on(&first.turns[0]))
            .and_then(|at| at.checked_sub(1))
            .map(|at| all[at].clone());
        assert_eq!(
            second_turns.last().cloned(),
            above,
            "the page above ends on the turn directly above this one, so no turn is skipped",
        );
    }

    /// A client walking back through history reaches EVERY turn, and the walk
    /// TERMINATES.
    ///
    /// **This is the assertion the keyed cursor failed.** A cursor the fold
    /// could not supply is `None` on every page, and `None` is the one signal
    /// a client reads as "nothing above this page" - so the walk stopped after
    /// one page and every turn above it was unreachable. Not an error and not
    /// a gap a client can see: a conversation that appeared to begin where the
    /// page did. `pages > 1` is the half that catches it.
    #[test]
    fn walking_back_through_history_sees_every_turn_and_ends() {
        let rows = turns_of(25);
        let borrowed: Vec<&str> = rows.iter().map(String::as_str).collect();
        let (messages, rendered) = a_conversation(&borrowed);
        let every = every_turn(&messages, &rendered);

        let mut seen: Vec<String> = Vec::new();
        let mut cursor: Option<String> = None;
        let mut pages = 0;
        loop {
            let asked = page(&messages, &rendered, 0, 0, cursor.as_deref(), 5);
            pages += 1;
            assert!(pages < every.len() + 2, "the walk terminates rather than cycling");
            seen.extend(asked.turns.iter().map(opened_on));
            match asked.cursor {
                Some(next) => cursor = Some(next),
                None => break,
            }
        }

        assert!(pages > 1, "the walk goes past the first page, or nothing above it is reachable");
        for turn in &every {
            assert!(seen.contains(turn), "every turn is reached, and {turn} was not");
        }
        assert!(
            seen.contains(&every[0]),
            "and the walk ends only once the OLDEST turn has been seen: {pages} pages",
        );
    }

    use std::collections::BTreeSet;
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Mutex};

    use super::*;
    use crate::surface::PendingKind;
    use crate::testing::ViewFacts;
    use crate::transport::Heartbeat;
    use crate::work::WorkCache;
    use forge_primitives::git_diff::GitDiffSnapshot;

    /// Where a run builds its fixture fleet: a directory of its own, so two
    /// runs overlapping on one machine share neither the store, the
    /// single-instance lock, nor the tree the record reads. Under `/tmp`
    /// rather than the ambient temp dir, which is the root the usage
    /// fixture's own label rule folds to (`/tmp` paths read as `scratch`) -
    /// a `$TMPDIR` root would pin a different label. Every form of the root
    /// stands in as `<fixture>` before a fixture is compared - see
    /// [`volatile`].
    ///
    /// The guard removes the directory when the test is done, and it does that
    /// only because of where it is declared: locals drop in reverse order, so a
    /// `dir` taken before `state` releases the tree after the store opened
    /// under it has closed.
    fn fixture_dir() -> tempfile::TempDir {
        tempfile::Builder::new().tempdir_in("/tmp").expect("the fixture's directory")
    }

    /// The fixture's own repository at `path`: `main` with one commit, a
    /// `worktree-pr` branch one commit ahead of it, and uncommitted work -
    /// so the record's content read carries a worktree layer and a
    /// branch-ahead layer, both populated.
    ///
    /// Deterministic on purpose: the fixture pins the hunks this builds,
    /// word for word.
    fn fixture_repo(path: &Path) {
        let git = |args: &[&str]| {
            let out = std::process::Command::new("git")
                .env_remove("GIT_DIR")
                .env_remove("GIT_WORK_TREE")
                .env_remove("GIT_COMMON_DIR")
                // The dates are pinned because the record now carries the
                // branch's commit chain, and a sha covers the commit's
                // timestamps: unpinned, every run minted a different sha and
                // the fixture could not hold one.
                .env("GIT_AUTHOR_DATE", "2026-01-01T00:00:00+00:00")
                .env("GIT_COMMITTER_DATE", "2026-01-01T00:00:00+00:00")
                .arg("-C")
                .arg(path)
                .args(args)
                .output()
                .expect("run git");
            assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
        };
        std::fs::create_dir_all(path).expect("the fixture's directory");
        git(&["init", "-q"]);
        git(&["symbolic-ref", "HEAD", "refs/heads/main"]);
        git(&["config", "user.email", "fixture@example.test"]);
        git(&["config", "user.name", "Fixture"]);
        git(&["config", "commit.gpgsign", "false"]);
        std::fs::write(path.join("kept.txt"), "one\ntwo\n").expect("write kept.txt");
        git(&["add", "-A"]);
        git(&["commit", "-qm", "base"]);
        git(&["checkout", "-q", "-b", "worktree-pr"]);
        std::fs::write(
            path.join("branch.txt"),
            "committed on the branch\nsecond line\nthird line\n",
        )
        .expect("write");
        git(&["add", "-A"]);
        git(&["commit", "-qm", "branch"]);
        // Uncommitted: a modification, and an addition staged so it is a
        // tracked change against HEAD.
        std::fs::write(path.join("kept.txt"), "one\ntwo\nthree\n").expect("write kept.txt");
        std::fs::write(path.join("added.txt"), "added one\n").expect("write added.txt");
        git(&["add", "added.txt"]);
    }

    /// The surface the fixtures are produced from, and both halves of the
    /// reason are deliberate.
    ///
    /// DETERMINISTIC, because every form of `root` stands in as `<fixture>`
    /// before anything is compared against a committed fixture - see
    /// [`volatile`] - so which directory the fleet lives in does not pin the
    /// shape. POPULATED, because an empty fleet pins almost nothing: most of
    /// a session's fields would be `null`, and a field renamed to `null`
    /// would still pass.
    async fn fixture_state(root: &Path) -> TransportState {
        fixture_state_seeded(root, |_| {}).await
    }

    /// [`fixture_state`] with one more seed of the test's own, laid down
    /// before the surface is built - for a test that pins a shape the default
    /// fixture deliberately leaves empty.
    async fn fixture_state_seeded(
        root: &Path,
        seed: impl FnOnce(&crate::testing::Fleet),
    ) -> TransportState {
        let fleet =
            crate::testing::Fleet::in_dir(root, &[("TestOrg", &["proj"])]).expect("the fleet");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.set_cli_version(Some("1.0.0"), Some("1.1.0"));
        fleet.set_user_preferences(serde_json::json!({}));
        fleet.add_worker("TestOrg", "proj", "w1").expect("the worker is added");
        fleet.install_agent("TestOrg", "proj", "lead");
        fleet
            .seed_transcript(
                "TestOrg",
                "proj",
                "lead",
                &[
                    r#"{"type":"user","message":{"role":"user","content":"hello"},"session_id":"s"}"#,
                    // A tool call, so the fixture pins a group's rows: the
                    // class it keys on, the word it draws and the tool's own
                    // name.
                    r#"{"type":"assistant","uuid":"a1","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"tool_use","id":"tu1","name":"Bash","input":{"command":"ls","description":"List files"}}]}}"#,
                    r#"{"type":"user","uuid":"u2","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"tu1","content":"a\nb"}]}}"#,
                    // A counted turn, so the usage fixture carries real
                    // tokens rather than a pool that reads as empty. The
                    // stamp is fixed in the past: a "now" stamp would put
                    // the record in the rolling windows, and which of them
                    // hold it flips at midnight.
                    r#"{"type":"assistant","uuid":"a2","timestamp":"2025-01-02T03:04:05.000Z","message":{"id":"m-usage","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"counted"}],"usage":{"input_tokens":7,"output_tokens":11,"cache_read_input_tokens":13,"cache_creation_input_tokens":17}}}"#,
                    // A sub-agent dispatch, so the record carries a
                    // POPULATED card rather than an empty list a rename
                    // could cross unseen.
                    r#"{"type":"assistant","uuid":"a3","message":{"id":"m-sub","role":"assistant","model":"claude-opus-5","content":[{"type":"tool_use","id":"tu-sub","name":"Task","input":{"description":"map the calls","subagent_type":"general-purpose","run_in_background":false,"prompt":"do the thing"}}]}}"#,
                    r#"{"type":"result","subtype":"success","duration_ms":1,"duration_api_ms":1,"is_error":false,"num_turns":1,"session_id":"s"}"#,
                ],
            )
            .expect("the transcript seeds");
        fleet.seed_test_pending_interaction(&fixture_seat(), PendingKind::Permission);
        // A rostered background task, so the fixture carries a populated
        // registry rather than an empty list a reader cannot tell from an
        // unwired field - and the two sub-agent facts, which the session
        // task's one fold raises and this fixture has no task to run: the
        // dispatch the transcript seeded, as the card list and the flag a
        // seat that ran it would hold.
        fleet.seed_view_facts(
            &fixture_seat(),
            ViewFacts {
                background_tasks: vec![forge_workspace::BackgroundTask {
                    task_id: "t-fixture".to_owned(),
                    is_backgrounded: true,
                    task_type: "local_bash".to_owned(),
                    description: "gh run watch".to_owned(),
                    command: Some("gh run watch 123 --exit-status".to_owned()),
                    tool_use_id: Some("tu-fixture".to_owned()),
                }],
                cards: vec![forge_primitives::runtime::SubagentCard {
                    name: "map the calls".to_owned(),
                    dispatch_id: "tu-sub".to_owned(),
                    agent_type: Some("general-purpose".to_owned()),
                    running: false,
                    failed: false,
                    backgrounded: false,
                    ended_at_ms: Some(1_750_000_000_000),
                    calls: 1,
                    tail: vec![forge_primitives::runtime::SubagentCall {
                        name: "Read".to_owned(),
                        title: "Read src/lib.rs".to_owned(),
                        status: forge_primitives::runtime::SubagentCallStatus::Completed,
                    }],
                    usage: Some(forge_primitives::messages::TaskUsage {
                        total_tokens: 9_714,
                        tool_uses: 1,
                        duration_ms: 2_716,
                        extras: serde_json::Map::new(),
                    }),
                }],
                ..ViewFacts::default()
            },
        );
        seed(&fleet);
        // A scan the fixture pins the working tree, the PR row and the
        // diff content from. It is taken of a repository of its own beside
        // the fleet rather than of the seat's project directory: building
        // one AT that path would also make the path exist, and the home
        // fixture's rows - pinned with that directory gone - would move
        // with it. The scan is real, so the record's content read has both
        // layers to carry rather than pinning a failed layer; the PR row
        // is stamped on top, since a temp repo has no remote for `gh` to
        // find one through.
        let surface = fleet.surface();
        let work = Arc::new(WorkCache::new());
        let repo = root.join("diff-repo");
        fixture_repo(&repo);
        let mut diff =
            forge_workspace::env::git_diff::scan(&repo, &std::collections::HashMap::new(), None)
                .await;
        diff.pr = Some(forge_primitives::git::GitPrInfo {
            number: 1249,
            url: "https://example.test/pull/1249".to_owned(),
            draft: false,
        });
        diff.closes = vec![forge_primitives::git::GitIssueRef {
            number: 1215,
            url: "https://example.test/issues/1215".to_owned(),
        }];
        diff.pushed_sha = Some("abc123".to_owned());
        diff.pr_fetched_at = None;
        surface.store_work_snapshot(
            &fixture_seat(),
            forge_workspace::work::WorkSnapshot {
                diff,
                cwd: repo,
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );

        let state = TransportState {
            surface,
            work,
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        // The transport does not read a transcript, so the seat's
        // conversation is put where a `Connected` would have left it.
        fleet
            .hold_conversation(&state, "TestOrg", "proj", "lead")
            .expect("the fixture's conversation is held");
        state
    }

    /// The substrings a fixture must not pin: the directory the fleet
    /// happens to live in, in every form the wire spells it.
    ///
    /// Three forms, because the wire carries all three. A path arrives raw
    /// (`ProjectView::path`), canonically (anything that resolved a symlink)
    /// and SLUGGED (`ProjectKey`, every non-alphanumeric turned into `-`).
    /// The slug form is why replacing the plain path alone is not enough:
    /// the key is a temp directory spelled with dashes, and it differs per
    /// machine just as much as the path does.
    fn volatile(root: &Path) -> Vec<(String, &'static str)> {
        let raw = root.to_string_lossy().into_owned();
        let canonical = root
            .canonicalize()
            .unwrap_or_else(|_| root.to_path_buf())
            .to_string_lossy()
            .into_owned();
        let mut forms = vec![
            (raw.clone(), "<fixture>"),
            (canonical.clone(), "<fixture>"),
            (slug(&raw), "<fixture>"),
            (slug(&canonical), "<fixture>"),
            // The build stamp carries the commit the binary was built from, so
            // it moves on every commit and a fixture pinning it would fail on
            // the next one. Each gets its OWN stand-in: collapsing them to one
            // placeholder would leave a client unable to tell which field is
            // which.
            (crate::FORGE_VERSION.to_owned(), "<forge-version>"),
            (crate::FORGE_VERSION_SHORT.to_owned(), "<forge-version-short>"),
            // The run's own identity differs per process, so a fixture that
            // pinned it would fail on the next run.
            (boot_id().to_owned(), "<boot-id>"),
        ];
        forms.sort_by_key(|(form, _)| std::cmp::Reverse(form.len()));
        forms.dedup_by(|left, right| left.0 == right.0);
        forms
    }

    /// The slug `ProjectKey` builds: every non-alphanumeric becomes `-`.
    fn slug(path: &str) -> String {
        path.chars().map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }).collect()
    }

    fn normalise(value: &mut Value, forms: &[(String, &'static str)]) {
        match value {
            Value::String(text) => {
                for (form, stands_in) in forms {
                    if let Some(rest) = text.strip_prefix(form.as_str()) {
                        *text = format!("{stands_in}{rest}");
                        break;
                    }
                }
            }
            Value::Array(items) => items.iter_mut().for_each(|item| normalise(item, forms)),
            Value::Object(fields) => fields.values_mut().for_each(|field| normalise(field, forms)),
            _ => {}
        }
    }

    /// **The home's attention item is the whole ask.** The default fixture
    /// pins an empty list, so a shape no test ever draws could be renamed
    /// unseen; this seeds one waiting row and reads the item the fleet row
    /// encodes - the task, the seat, the subject, the ask's words and the
    /// moment the wait began.
    #[tokio::test]
    async fn a_decision_wait_crosses_with_its_ask() {
        let dir = fixture_dir();
        let state = fixture_state_seeded(dir.path(), |fleet| {
            fleet.seed_task(forge_primitives::tasks::Task {
                id: forge_primitives::tasks::TaskId::from("t-wait"),
                project_name: "proj".to_owned(),
                subject: "ship the parser".to_owned(),
                active_form: None,
                detail: None,
                status: forge_primitives::tasks::TaskStatus::Waiting,
                owner: Some(forge_primitives::SessionSlot::worker("TestOrg", "proj", "w1")),
                parent: None,
                waiting_on: Some(forge_primitives::tasks::Waiting {
                    kind: Some(forge_primitives::tasks::WaitingKind::Decision),
                    detail: Some("sign this off".to_owned()),
                    on: None,
                    verification: false,
                    parked: false,
                    since: Some(
                        std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(5),
                    ),
                }),
                estimate: None,
                rank: None,
                verify: None,
                attested: None,
                reserved: None,
                links: Vec::new(),
                attempt: 0,
                archived_at: None,
                created_at: std::time::SystemTime::UNIX_EPOCH,
                // An edit after the wait began must not move the item's date.
                updated_at: std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(9),
            });
        })
        .await;

        let mut encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        normalise(&mut encoded, &volatile(dir.path()));
        let row = encoded["fleet"]
            .as_array()
            .expect("the fleet")
            .iter()
            .find(|row| row["project"] == "proj")
            .expect("the project's row");
        assert_eq!(row["waiting_on_user"], 1, "the wait counts");
        let item = &row["attention"][0];
        assert_eq!(item["task"], "t-wait", "the row's id");
        assert_eq!(item["owner"], "w1", "the seat that raised it");
        assert_eq!(item["subject"], "ship the parser", "the row's subject");
        assert_eq!(item["words"], "sign this off", "the ask's own words");
        assert_eq!(
            item["at"],
            serde_json::json!({ "secs_since_epoch": 5, "nanos_since_epoch": 0 }),
            "the moment the wait began",
        );
    }

    /// Every subject a client can watch, with the fixture that pins its wire
    /// shape.
    fn fixtures() -> Vec<(Subject, PathBuf)> {
        let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/wire_fixtures");
        vec![
            (Subject::Home, dir.join("home.json")),
            (Subject::Board, dir.join("board.json")),
            (Subject::Session(fixture_seat()), dir.join("session.json")),
            (Subject::Usage, dir.join("usage.json")),
        ]
    }

    /// Writes each subject's fixture from the encoder. Run deliberately:
    /// `cargo nextest run -p forge-server --run-ignored ignored-only write_the_fixtures`.
    ///
    /// **What this writes is only what the encoder currently emits**, so a
    /// fixture produced here pins the present shape and a wrong one equally.
    /// Each has to be READ against the intended shape before it is
    /// committed; generating one is not the work.
    #[tokio::test]
    #[ignore = "writes the fixtures; run deliberately"]
    async fn write_the_fixtures() {
        let dir = fixture_dir();
        let state = fixture_state(dir.path()).await;
        let forms = volatile(dir.path());
        std::fs::create_dir_all(fixtures()[0].1.parent().expect("a directory")).expect("mkdir");
        for (subject, fixture) in fixtures() {
            let mut encoded = encode_subject(&state, &subject).await.expect("encode");
            normalise(&mut encoded, &forms);
            std::fs::write(&fixture, serde_json::to_string_pretty(&encoded).expect("render"))
                .expect("write");
        }
    }

    /// A seat's own working tree crosses on its row, which is what lets a
    /// worker's row draw its own branch rather than its project's.
    ///
    /// Read through the same shared cache the project rows use, so this is one
    /// read drawn twice rather than two reads - and it is a read a page cannot
    /// derive from the project's, because the two name different directories.
    #[tokio::test]
    async fn a_home_agents_row_carries_its_own_tree() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        let row = encoded["agents"]
            .as_array()
            .expect("agents is a list")
            .iter()
            .find(|row| row["slot"]["label"] == "lead")
            .expect("the started project's lead is a row");

        assert!(
            row.get("work").is_some(),
            "the seat's own tree crosses on the row, not the project's read: {row}"
        );
    }

    /// A seat between heal attempts crosses the home with its state and its
    /// attempt, which is what a row draws `restarting (1 of 3)` from.
    /// Catches a wire that spells `AgentRow`'s fields out and drops the two
    /// new ones, which would draw the seat as something it is not.
    #[tokio::test]
    async fn a_home_agents_row_carries_the_heal_attempt() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let lead = forge_primitives::SessionSlot::lead("TestOrg", "proj");
        let next_at =
            std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_790_000_000);
        fleet.workspace().set_heal_progress(&lead, 1, next_at);
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        let row = encoded["agents"]
            .as_array()
            .expect("agents is a list")
            .iter()
            .find(|row| row["slot"]["label"] == "lead")
            .expect("the started project's lead is a row");

        assert_eq!(row["lifecycle"], serde_json::json!("Restarting"), "the state the row draws");
        assert_eq!(row["heal_attempt"], serde_json::json!(1), "the attempt it names");
        assert_eq!(row["heal_attempts"], serde_json::json!(3), "the cap the row's 'of 3' reads");
    }

    /// **The home carries the fleet glance and the board carries the board**
    /// (#2517): the three fields that were most of the home's ~708 KB payload
    /// moved to their own subject, and the board's answer carries every one.
    #[tokio::test]
    async fn the_home_carries_no_board_and_the_board_carries_all_of_it() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.add_task("TestOrg", "proj", "a counted row", "w1", None).expect("the task is seeded");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let home = encode_subject(&state, &Subject::Home).await.expect("the home encodes");
        let home_project = &home["projects"][0];
        for field in ["rows", "crons", "connectors"] {
            assert!(
                home_project.get(field).is_none(),
                "the home still carries the board's `{field}`: {home_project}",
            );
        }

        let board = encode_subject(&state, &Subject::Board).await.expect("the board encodes");
        let board_project = &board["projects"][0];
        for field in ["rows", "crons", "connectors"] {
            assert!(
                board_project.get(field).is_some(),
                "the board's snapshot does not carry `{field}`, which its updates move: \
                 {board_project}",
            );
        }

        // **The header's total is the board's own row count**, on a project
        // that holds one: the fleet row must count it, not count it minus one.
        let board_rows = board_project["rows"].as_array().map(Vec::len);
        assert_eq!(board_rows, Some(1), "the seeded row did not reach the board: {board}");
        assert_eq!(
            home["fleet"][0]["rows"].as_u64(),
            Some(1),
            "the fleet row's total does not match the board's rows: {home}",
        );
    }

    /// **A board row's owner-gone chip is read from the seat registry on the
    /// encode**, not from the task: an owner nothing holds is `stale`, and a
    /// seat that comes up clears it on the next read (#2517).
    #[tokio::test]
    async fn a_board_row_marks_an_owner_whose_seat_is_not_live() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.add_task("TestOrg", "proj", "an owned row", "w1", None).expect("the task is seeded");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let board = encode_subject(&state, &Subject::Board).await.expect("encode");
        let gone_chip = |board: &Value| board["projects"][0]["rows"][0]["marks"]["stale"].clone();
        assert_eq!(
            gone_chip(&board),
            Value::Bool(true),
            "an owner no seat holds must draw the gone chip: {board}",
        );

        // The seat comes up: the next read clears the chip.
        let _commands = fleet.install_agent("TestOrg", "proj", "w1");
        let board = encode_subject(&state, &Subject::Board).await.expect("encode");
        assert_eq!(
            gone_chip(&board),
            Value::Bool(false),
            "a live owner must not draw the gone chip: {board}",
        );
    }

    /// The home's fatal carries the words the server prints, not the serde
    /// tag: a view drawing the line words it from the field (#1638).
    #[tokio::test]
    async fn a_home_fatal_carries_the_terminals_words() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.set_fatal_error(forge_primitives::error::AppError::ConnectionFailed);
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        assert_eq!(
            encoded["fatal_error"],
            serde_json::json!("Failed to establish or maintain the Agent SDK bridge connection."),
            "the home's fatal is the terminal's own words"
        );
    }

    /// The rows the processes feed leads with: the CLI's background-task
    /// registry, which no read answered.
    ///
    /// Without it a client drew the OS walk and nothing else, so a
    /// `setsid`-detached bash - the case the registry exists for - was a
    /// missing row rather than an empty field.
    #[tokio::test]
    async fn a_session_snapshot_carries_the_background_registry() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.seed_view_facts(
            &fixture_seat(),
            ViewFacts {
                background_tasks: vec![forge_workspace::BackgroundTask {
                    task_id: "t1".to_owned(),
                    is_backgrounded: true,
                    task_type: "local_bash".to_owned(),
                    description: "gh run watch".to_owned(),
                    command: Some("gh run watch 123 --exit-status".to_owned()),
                    tool_use_id: Some("tu-1".to_owned()),
                }],
                ..ViewFacts::default()
            },
        );
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");

        assert_eq!(
            encoded["background_tasks"][0]["task_id"], "t1",
            "the seat's registry crosses: {encoded}"
        );
        assert_eq!(
            encoded["background_tasks"][0]["description"], "gh run watch",
            "with the line the row leads with: {encoded}"
        );
        assert_eq!(
            encoded["background_tasks"][0]["command"], "gh run watch 123 --exit-status",
            "and the command the scan adopts by: {encoded}"
        );
    }

    /// The occupant's id crosses on the header: the `Connected` that named it
    /// reaches nobody who was not attached when it fired, so a page opened on
    /// a running seat has this rather than the frame. The worker registry
    /// carries the same id for a live dynamic worker, and nothing else does.
    #[tokio::test]
    async fn a_session_snapshot_carries_the_occupants_id() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.seed_view_facts(
            &fixture_seat(),
            ViewFacts {
                session_id: Some(forge_primitives::SessionId::new("d4f70669-1f2a")),
                ..ViewFacts::default()
            },
        );
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");

        assert_eq!(
            encoded["header"]["session_id"], "d4f70669-1f2a",
            "the occupant the core named crosses on the header: {encoded}"
        );
    }

    /// `/resume`'s list rides the record, because the composer's other sources
    /// do: a field that never reached the JSON would leave the popover with
    /// nothing to offer and nothing on screen saying why.
    #[tokio::test]
    async fn a_session_snapshot_carries_the_sessions_the_seat_can_resume() {
        let dir = tempfile::tempdir().expect("fleet dir");
        let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
            .expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let cwd =
            fleet.workspace().cwd_for_session(&fixture_seat()).expect("the project's directory");
        let key = forge_workspace::userdata::catalog::scan::project_key_for_directory(Some(&cwd));
        let project_dir = dir.path().join("projects").join(key);
        std::fs::create_dir_all(&project_dir).expect("the project's transcripts");
        for (id, summary, secs) in [
            ("aaaa-0000", "the older session", 1_700_000_100_u64),
            ("bbbb-1111", "the newer session", 1_700_000_200),
        ] {
            let path = project_dir.join(format!("{id}.jsonl"));
            std::fs::write(
                &path,
                format!(
                    "{{\"type\":\"user\",\"cwd\":\"{cwd}\",\"message\":{{\"content\":\"{summary}\"}}}}\n"
                ),
            )
            .expect("write a transcript");
            let file = std::fs::File::options().write(true).open(&path).expect("open it");
            file.set_modified(std::time::UNIX_EPOCH + std::time::Duration::from_secs(secs))
                .expect("pin the transcript's mtime");
        }
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");

        assert_eq!(
            encoded["resume_sessions"][0]["session_id"], "bbbb-1111",
            "the newest session leads, by the id a pick writes: {encoded}"
        );
        assert_eq!(
            encoded["resume_sessions"][0]["summary"], "the newer session",
            "with the transcript's own title beside it: {encoded}"
        );
        assert_eq!(
            encoded["resume_sessions"][1]["session_id"], "aaaa-0000",
            "and the older one follows: {encoded}"
        );
    }

    /// **The instance list the record carries is the workspace's**, not a
    /// fold the record takes over the conversation.
    ///
    /// It was folded here until the cap landed - over the whole held
    /// conversation, on every read - and the two answers part company the
    /// moment the conversation a record can see is shorter than the one the
    /// session has run: a section that shrank with the cap would be one no
    /// viewer of the session's own frames agrees with. The seat below
    /// dispatches in its conversation, so a fold here would answer a card,
    /// and what the record carries has to be the workspace's rather than it.
    #[tokio::test]
    async fn the_record_reads_the_instance_list_from_the_workspace() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet
            .seed_transcript(
                "TestOrg",
                "proj",
                "lead",
                &[
                    r#"{"type":"user","message":{"role":"user","content":"map it"},"session_id":"s"}"#,
                    r#"{"type":"assistant","uuid":"a1","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"tool_use","id":"tu-sub","name":"Task","input":{"description":"from the conversation","subagent_type":"Explore"}}]}}"#,
                ],
            )
            .expect("the transcript seeds");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        fleet
            .hold_conversation(&state, "TestOrg", "proj", "lead")
            .expect("the fixture's conversation is held");

        let before =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");
        assert_eq!(
            before["subagent_instances"].as_array().expect("a list").len(),
            0,
            "the conversation's own dispatch is not folded by the read: {before}",
        );

        fleet.seed_view_facts(
            &fixture_seat(),
            ViewFacts {
                cards: vec![forge_primitives::runtime::SubagentCard {
                    name: "from the workspace".to_owned(),
                    dispatch_id: "tu-sub".to_owned(),
                    agent_type: Some("Explore".to_owned()),
                    running: true,
                    failed: false,
                    backgrounded: false,
                    ended_at_ms: None,
                    calls: 0,
                    tail: Vec::new(),
                    usage: None,
                }],
                ..ViewFacts::default()
            },
        );

        let after =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");
        let cards = after["subagent_instances"].as_array().expect("a list");
        assert_eq!(
            cards.len(),
            1,
            "and the list the workspace holds is the one that crosses: {after}"
        );
        assert_eq!(
            cards[0]["name"], "from the workspace",
            "named by the fold that announces it rather than by a second one here: {after}",
        );
    }

    /// **A seat whose loop never ran is walked at the read**, the same
    /// fallback the work row takes. A fixture holds no seat, so no loop runs
    /// for one: the record's index is the tree's rather than an empty list
    /// standing in for a tree nobody looked at.
    #[tokio::test]
    async fn the_record_walks_the_index_when_the_store_holds_none() {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path().join("proj");
        std::fs::create_dir_all(&root).expect("mkdir");
        std::fs::write(root.join("found.rs"), "").expect("write");
        let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
            .expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");

        let entries = encoded["file_index"]["entries"].as_object().expect("an index crosses");
        assert!(
            entries.contains_key("found.rs"),
            "a seat nothing has walked is walked at the read: {encoded}",
        );
    }

    /// The PR row the inspector leads its GIT section with: the open pull
    /// request, and the issues it closes.
    ///
    /// The keys are asserted PRESENT rather than populated, because a
    /// populated one needs `gh`, a pushed branch and an open PR - which this
    /// fixture cannot produce. The key's presence is the property that matters
    /// here: a client drawing the section reads a field that is there and
    /// null, rather than finding no field at all.
    ///
    /// The POPULATED shape is pinned next door, from a seeded scan, and in
    /// the committed fixture; the live walk that reached a real PR is in the
    /// plan's ledger at
    /// `docs/superpowers/ledgers/2026-09-28-forge-server-socket/` (git-excluded,
    /// so no repo-visible artifact carries it).
    #[tokio::test]
    async fn a_session_snapshot_carries_the_pr_row() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");

        assert!(encoded.get("pr").is_some(), "the session carries a PR field: {encoded}");
        assert_eq!(
            encoded["closes"],
            serde_json::json!([]),
            "and the issues it closes, empty when there is no PR: {encoded}"
        );
    }

    /// **The record answers the machine turn a page attaching mid-handoff
    /// needs.** The live page learns it from the update; this is what a
    /// second client reads while the judgment turn runs.
    #[tokio::test]
    async fn a_record_attached_mid_handoff_carries_the_machine_turn() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let workspace = fleet.workspace();
        let seat = fixture_seat();
        let domain = match workspace.domain_session_for(&seat) {
            Some(domain) => domain,
            None => workspace.register_domain_session(seat.clone(), None),
        };
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        let subject = Subject::Session(seat);

        let idle = encode_subject(&state, &subject).await.expect("the record encodes");
        assert_eq!(idle["composer"]["machine"], Value::Null, "an idle seat takes input");

        domain.lock().handoff.awaiting = true;
        let mid = encode_subject(&state, &subject).await.expect("the record encodes");
        assert_eq!(
            mid["composer"]["machine"],
            serde_json::json!("handoff"),
            "a page attaching mid-handoff reads the machine turn",
        );
    }

    /// The record carries the changed files' hunks, bounded and flagged: a
    /// review surface has no other read of them.
    ///
    /// Its own fleet and its own repository rather than `fixture_state`'s: what
    /// this pins is the hunk shape of a three-file diff, and the shared
    /// fixture's record is a tree of a different shape.
    #[tokio::test]
    async fn the_record_carries_the_changed_files_with_their_hunks() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let surface = fleet.surface();
        let repo = tempfile::tempdir().expect("a repository directory");
        fixture_repo(repo.path());
        let diff = forge_workspace::env::git_diff::scan(
            repo.path(),
            &std::collections::HashMap::new(),
            None,
        )
        .await;
        surface.store_work_snapshot(
            &seat,
            forge_workspace::work::WorkSnapshot {
                diff,
                cwd: repo.path().to_owned(),
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );
        let state = TransportState {
            surface,
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Session(seat)).await.expect("encode");

        let files = encoded["diff"]["worktree"]["populated"]
            .as_array()
            .expect("the worktree layer is populated");
        let added = files.iter().find(|file| file["path"] == "added.txt").expect("added.txt");
        assert_eq!(added["status"], "added", "the status crosses: {encoded}");
        let lines = added["hunks"][0]["lines"].as_array().expect("hunks carry their lines");
        assert!(
            lines.iter().any(|line| line["kind"] == "added" && line["text"] == "added one"),
            "the hunk's own lines cross: {encoded}",
        );
        assert_eq!(encoded["diff"]["truncated"], false, "nothing was withheld");
        let ahead = encoded["diff"]["branch_ahead"]["populated"]
            .as_array()
            .expect("the branch layer is populated too");
        assert!(
            ahead.iter().any(|file| file["path"] == "branch.txt"),
            "and the branch's own commit is there: {encoded}",
        );
    }

    /// The record resolves each all-changes thread against the content it
    /// ships, and leaves a commit-scoped thread unresolved.
    ///
    /// The landing rides the read because a client has no working tree to
    /// re-scan.
    ///
    /// **A commit-scoped thread is placed here too, when the branch's
    /// content still carries it** - the all-changes view draws it beside
    /// the same code, and the recorded line is never the evidence: it
    /// counts in the commit's numbering. A thread the branch no longer
    /// carries answers outdated, and a cross-scope landing carries no
    /// `moved_from`, whose number belongs to the other scope.
    ///
    /// **A thread written against another base is refused the landing too**:
    /// its line numbers belong to a diff this record does not carry, and a
    /// placement would be a guess (the placement refuses exactly that).
    #[tokio::test]
    async fn the_record_carries_each_threads_landing() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let surface = fleet.surface();
        let repo = tempfile::tempdir().expect("a repository directory");
        fixture_repo(repo.path());
        let diff = forge_workspace::env::git_diff::scan(
            repo.path(),
            &std::collections::HashMap::new(),
            None,
        )
        .await;
        surface.store_work_snapshot(
            &seat,
            forge_workspace::work::WorkSnapshot {
                diff,
                cwd: repo.path().to_owned(),
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );
        let thread = |id: &str,
                      path: &str,
                      text: &str,
                      line: u32,
                      context: &[&str],
                      commit: Option<&str>,
                      base: &str|
         -> forge_primitives::review::ReviewThread {
            forge_primitives::review::ReviewThread {
                id: id.to_owned(),
                anchor: forge_primitives::review::ReviewAnchor {
                    path: path.to_owned(),
                    side: forge_primitives::review::ReviewSide::New,
                    line,
                    content_hash: forge_workspace::env::git_diff::resolver::anchor_hash(text),
                    context: context.iter().map(|held| (*held).to_owned()).collect(),
                    base_ref: base.to_owned(),
                },
                comments: vec![forge_primitives::review::ReviewComment {
                    author: forge_primitives::review::ReviewAuthor::User,
                    text: "c".to_owned(),
                    at: "2026-01-01T00:00:00Z".to_owned(),
                    review_id: None,
                }],
                status: forge_primitives::review::ReviewStatus::Open,
                created_at: "2026-01-01T00:00:00Z".to_owned(),
                updated_at: "2026-01-01T00:00:00Z".to_owned(),
                commit: commit.map(str::to_owned),
            }
        };
        for (id, text, line, context, commit, base) in [
            ("placed", "committed on the branch", 1, &[][..], None, "main"),
            ("gone", "a line the branch does not carry", 1, &[][..], None, "main"),
            (
                "scoped",
                "second line",
                2,
                &["committed on the branch", "third line"][..],
                Some("abc123"),
                "main",
            ),
            ("scoped-moved", "third line", 1, &["second line"][..], Some("abc123"), "main"),
            ("scoped-gone", "not on the branch at all", 1, &[][..], Some("abc123"), "main"),
            ("other-base", "committed on the branch", 1, &[][..], None, "HEAD"),
        ] {
            assert!(
                fleet.workspace().upsert_review_thread(
                    "proj",
                    "worktree-pr",
                    thread(id, "branch.txt", text, line, context, commit, base),
                ),
                "the {id} thread stores",
            );
        }
        let state = TransportState {
            surface,
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Session(seat)).await.expect("encode");

        let threads =
            encoded["reviews"]["threads"]["value"].as_array().expect("the threads read answered");
        let of = |id: &str| threads.iter().find(|held| held["id"] == id).expect(id);
        assert_eq!(
            of("placed")["resolution"],
            serde_json::json!({"state": "placed", "line": 1, "moved_from": null}),
            "an anchored line still carrying its text lands where it was written: {encoded}",
        );
        assert_eq!(
            of("gone")["resolution"],
            serde_json::json!({"state": "outdated", "reason": "gone"}),
            "code the branch no longer carries is outdated, in the resolver's own words",
        );
        assert_eq!(
            of("scoped")["resolution"],
            serde_json::json!({"state": "placed", "line": 2, "moved_from": null}),
            "a commit-scoped thread the branch still carries lands there, by content: {encoded}",
        );
        assert_eq!(
            of("scoped-moved")["resolution"],
            serde_json::json!({"state": "placed", "line": 3, "moved_from": null}),
            "and the recorded line is not named beside this scope's: it counts in the commit's",
        );
        assert_eq!(
            of("scoped-gone")["resolution"],
            serde_json::json!({"state": "outdated", "reason": "gone"}),
            "a commit-scoped thread the branch no longer carries is outdated there",
        );
        assert_eq!(
            of("other-base")["resolution"],
            serde_json::Value::Null,
            "nor has one numbered against another base, whose lines address other code here",
        );
    }

    /// A branch and its base sharing a wide file, so a change in its middle
    /// leaves the ends outside the branch-ahead diff's hunks.
    fn wide_repo(path: &Path) {
        let git = |args: &[&str]| {
            let out = std::process::Command::new("git")
                .env_remove("GIT_DIR")
                .env_remove("GIT_WORK_TREE")
                .env_remove("GIT_COMMON_DIR")
                .env("GIT_AUTHOR_DATE", "2026-01-01T00:00:00+00:00")
                .env("GIT_COMMITTER_DATE", "2026-01-01T00:00:00+00:00")
                .arg("-C")
                .arg(path)
                .args(args)
                .output()
                .expect("run git");
            assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
        };
        let wide = |ten: &str| -> String {
            (1..=20)
                .map(|n| if n == 10 { format!("{ten}\n") } else { format!("wide {n}\n") })
                .collect()
        };
        std::fs::create_dir_all(path).expect("the fixture's directory");
        git(&["init", "-q"]);
        git(&["symbolic-ref", "HEAD", "refs/heads/main"]);
        git(&["config", "user.email", "fixture@example.test"]);
        git(&["config", "user.name", "Fixture"]);
        git(&["config", "commit.gpgsign", "false"]);
        std::fs::write(path.join("wide.txt"), wide("wide ten")).expect("write wide.txt");
        git(&["add", "-A"]);
        git(&["commit", "-qm", "base"]);
        git(&["checkout", "-q", "-b", "worktree-pr"]);
        std::fs::write(path.join("wide.txt"), wide("wide ten changed")).expect("write wide.txt");
        git(&["add", "-A"]);
        git(&["commit", "-qm", "branch"]);
    }

    /// **A comment written on a revealed line lands in the whole file.**
    /// The line exists in the branch, but outside the branch-ahead diff's
    /// bounded hunk; the record holds the whole file for exactly that case,
    /// at the revision the anchor's side is numbered in. Text the file does
    /// not carry is still outdated.
    #[tokio::test]
    async fn the_record_places_a_thread_outside_the_hunks_in_the_whole_file() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let surface = fleet.surface();
        let repo = tempfile::tempdir().expect("a repository directory");
        wide_repo(repo.path());
        let diff = forge_workspace::env::git_diff::scan(
            repo.path(),
            &std::collections::HashMap::new(),
            None,
        )
        .await;
        surface.store_work_snapshot(
            &seat,
            forge_workspace::work::WorkSnapshot {
                diff,
                cwd: repo.path().to_owned(),
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );
        let thread = |id: &str, side, text: &str, context: &[&str], commit: Option<&str>| {
            forge_primitives::review::ReviewThread {
                id: id.to_owned(),
                anchor: forge_primitives::review::ReviewAnchor {
                    path: "wide.txt".to_owned(),
                    side,
                    line: 2,
                    content_hash: forge_workspace::env::git_diff::resolver::anchor_hash(text),
                    context: context.iter().map(|held| (*held).to_owned()).collect(),
                    base_ref: "main".to_owned(),
                },
                comments: vec![forge_primitives::review::ReviewComment {
                    author: forge_primitives::review::ReviewAuthor::User,
                    text: "c".to_owned(),
                    at: "2026-01-01T00:00:00Z".to_owned(),
                    review_id: None,
                }],
                status: forge_primitives::review::ReviewStatus::Open,
                created_at: "2026-01-01T00:00:00Z".to_owned(),
                updated_at: "2026-01-01T00:00:00Z".to_owned(),
                commit: commit.map(str::to_owned),
            }
        };
        for held in [
            thread("revealed-new", ReviewSide::New, "wide 2", &["wide 1", "wide 3"], None),
            thread("revealed-old", ReviewSide::Old, "wide 2", &["wide 1", "wide 3"], None),
            thread("revealed-gone", ReviewSide::New, "not in the file", &[], None),
            // The same path and side as an eligible read, but its line counts
            // in its own commit: it must not borrow that read.
            thread(
                "scoped-borrow",
                ReviewSide::New,
                "wide 2",
                &["wide 1", "wide 3"],
                Some("abc123"),
            ),
        ] {
            assert!(
                fleet.workspace().upsert_review_thread("proj", "worktree-pr", held),
                "the thread stores",
            );
        }
        let state = TransportState {
            surface,
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        let encoded = encode_subject(&state, &Subject::Session(seat)).await.expect("encode");
        let threads =
            encoded["reviews"]["threads"]["value"].as_array().expect("the threads read answered");
        let of = |id: &str| threads.iter().find(|held| held["id"] == id).expect(id);
        assert_eq!(
            of("revealed-new")["resolution"],
            serde_json::json!({"state": "placed", "line": 2, "moved_from": null}),
            "a new-side line outside the hunks lands in the branch tip's file: {encoded}",
        );
        assert_eq!(
            of("revealed-old")["resolution"],
            serde_json::json!({"state": "placed", "line": 2, "moved_from": null}),
            "and an old-side one in the base's, whose numbering is its own: {encoded}",
        );
        assert_eq!(
            of("revealed-gone")["resolution"],
            serde_json::json!({"state": "outdated", "reason": "gone"}),
            "code the file does not carry is still outdated: {encoded}",
        );
        assert_eq!(
            of("scoped-borrow")["resolution"],
            serde_json::json!({"state": "outdated", "reason": "gone"}),
            "a commit-scoped thread never borrows the branch's read: {encoded}",
        );
    }

    /// Run one git command in a fixture repository, with pinned dates.
    fn git_in(path: &Path, args: &[&str]) {
        let out = std::process::Command::new("git")
            .env_remove("GIT_DIR")
            .env_remove("GIT_WORK_TREE")
            .env_remove("GIT_COMMON_DIR")
            .env("GIT_AUTHOR_DATE", "2026-01-01T00:00:00+00:00")
            .env("GIT_COMMITTER_DATE", "2026-01-01T00:00:00+00:00")
            .arg("-C")
            .arg(path)
            .args(args)
            .output()
            .expect("run git");
        assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
    }

    /// Init a fixture repository on `main` with `user`/`email`/no signing.
    fn init_repo(path: &Path) {
        std::fs::create_dir_all(path).expect("the fixture's directory");
        git_in(path, &["init", "-q"]);
        git_in(path, &["symbolic-ref", "HEAD", "refs/heads/main"]);
        git_in(path, &["config", "user.email", "fixture@example.test"]);
        git_in(path, &["config", "user.name", "Fixture"]);
        git_in(path, &["config", "commit.gpgsign", "false"]);
    }

    /// Twenty numbered lines, with `ten` at the tenth.
    fn wide_lines(ten: &str) -> String {
        (1..=20).map(|n| if n == 10 { format!("{ten}\n") } else { format!("wide {n}\n") }).collect()
    }

    /// Store `diff` as the seat's scan and encode the record.
    async fn encode_repo(
        fleet: &crate::testing::Fleet,
        seat: &SessionSlot,
        repo: &Path,
    ) -> serde_json::Value {
        let surface = fleet.surface();
        let diff =
            forge_workspace::env::git_diff::scan(repo, &std::collections::HashMap::new(), None)
                .await;
        surface.store_work_snapshot(
            seat,
            forge_workspace::work::WorkSnapshot {
                diff,
                cwd: repo.to_owned(),
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );
        let state = TransportState {
            surface,
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        encode_subject(&state, &Subject::Session(seat.clone())).await.expect("encode")
    }

    /// One thread anchored on `wide.txt` at `line`, on `side`.
    fn wide_thread(
        id: &str,
        side: ReviewSide,
        line: u32,
        text: &str,
        context: &[&str],
        commit: Option<&str>,
    ) -> ReviewThread {
        ReviewThread {
            id: id.to_owned(),
            anchor: ReviewAnchor {
                path: "wide.txt".to_owned(),
                side,
                line,
                content_hash: forge_workspace::env::git_diff::resolver::anchor_hash(text),
                context: context.iter().map(|held| (*held).to_owned()).collect(),
                base_ref: "main".to_owned(),
            },
            comments: vec![forge_primitives::review::ReviewComment {
                author: forge_primitives::review::ReviewAuthor::User,
                text: "c".to_owned(),
                at: "2026-01-01T00:00:00Z".to_owned(),
                review_id: None,
            }],
            status: forge_primitives::review::ReviewStatus::Open,
            created_at: "2026-01-01T00:00:00Z".to_owned(),
            updated_at: "2026-01-01T00:00:00Z".to_owned(),
            commit: commit.map(str::to_owned),
        }
    }

    /// The landing of one thread in an encoded record.
    fn landing(encoded: &serde_json::Value, id: &str) -> serde_json::Value {
        let threads =
            encoded["reviews"]["threads"]["value"].as_array().expect("the threads read answered");
        threads.iter().find(|held| held["id"] == id).expect(id)["resolution"].clone()
    }

    /// **The old side is numbered at the merge-base, not at the base's
    /// current tip.** A base that has advanced since the branch forked shifts
    /// every line in its tip, so a revealed old-side comment read there would
    /// land one line off (or park Gone). A once-forked, then-advanced base is
    /// the ordinary shape of a long-lived branch.
    #[tokio::test]
    async fn the_record_reads_the_old_side_at_the_merge_base() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let repo = tempfile::tempdir().expect("a repository directory");
        init_repo(repo.path());
        std::fs::write(repo.path().join("wide.txt"), wide_lines("wide 10")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "base"]);
        git_in(repo.path(), &["checkout", "-q", "-b", "worktree-pr"]);
        std::fs::write(repo.path().join("wide.txt"), wide_lines("wide 10 changed")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "branch"]);
        // The base advances under the fork: a line inserted at the top shifts
        // every line number below it.
        git_in(repo.path(), &["checkout", "-q", "main"]);
        let advanced: String = std::iter::once("wide 0\n".to_owned())
            .chain((1..=20).map(|n| format!("wide {n}\n")))
            .collect();
        std::fs::write(repo.path().join("wide.txt"), advanced).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "advance"]);
        git_in(repo.path(), &["checkout", "-q", "worktree-pr"]);
        for held in [
            // The old side numbers at the merge-base.
            wide_thread("revealed-old", ReviewSide::Old, 2, "wide 2", &["wide 1", "wide 3"], None),
            // And the new side at the branch tip: the base's own numbering
            // shifted the same line, so a read there would place it one down.
            wide_thread(
                "revealed-new",
                ReviewSide::New,
                16,
                "wide 16",
                &["wide 15", "wide 17"],
                None,
            ),
        ] {
            assert!(
                fleet.workspace().upsert_review_thread("proj", "worktree-pr", held),
                "the thread stores",
            );
        }
        let encoded = encode_repo(&fleet, &seat, repo.path()).await;
        assert_eq!(
            landing(&encoded, "revealed-old"),
            serde_json::json!({"state": "placed", "line": 2, "moved_from": null}),
            "the merge-base's line 2 is the comment's own, though main's tip shifted it: {encoded}",
        );
        assert_eq!(
            landing(&encoded, "revealed-new"),
            serde_json::json!({"state": "placed", "line": 16, "moved_from": null}),
            "and the branch tip's line 16 is the new side's own: {encoded}",
        );
    }

    /// **The old side resolves against the base the diff was pinned to, not
    /// the ref's name.** The name can have moved to a commit the branch now
    /// contains - the base-only move - and the merge-base resolved at that
    /// name is a different commit from the one the diff was computed against.
    #[tokio::test]
    async fn the_old_side_reads_the_pinned_base_not_the_moved_name() {
        let repo = tempfile::tempdir().expect("a repository directory");
        init_repo(repo.path());
        std::fs::write(repo.path().join("wide.txt"), wide_lines("wide 10")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "base"]);
        git_in(repo.path(), &["checkout", "-q", "-b", "worktree-pr"]);
        std::fs::write(repo.path().join("wide.txt"), wide_lines("wide 10 changed")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "branch"]);
        let base = forge_workspace::env::git_diff::file::merge_base(repo.path(), "main")
            .await
            .expect("the fork point");
        // The base advances with a line inserted at the top, which shifts every
        // line number below it, and the branch then merges that advance: the
        // committed-to name IS the branch's history, while the pinned base the
        // diff was computed against is the fork.
        git_in(repo.path(), &["checkout", "-q", "main"]);
        let advanced: String = std::iter::once("wide 0\n".to_owned())
            .chain((1..=20).map(|n| format!("wide {n}\n")))
            .collect();
        std::fs::write(repo.path().join("wide.txt"), advanced).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "advance"]);
        let moved = std::process::Command::new("git")
            .arg("-C")
            .arg(repo.path())
            .args(["rev-parse", "main"])
            .output()
            .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_owned())
            .expect("git rev-parse");
        git_in(repo.path(), &["checkout", "-q", "worktree-pr"]);
        git_in(repo.path(), &["merge", "-q", "--no-edit", "-X", "theirs", "main"]);

        let snapshot = forge_workspace::env::git_diff::scan(
            repo.path(),
            &std::collections::HashMap::new(),
            None,
        )
        .await;
        let content =
            forge_workspace::env::git_diff::content::scan(repo.path(), &snapshot, None, None, None)
                .await;
        let threads = vec![wide_thread(
            "revealed-old",
            ReviewSide::Old,
            2,
            "wide 2",
            &["wide 1", "wide 3"],
            None,
        )];
        let key = ("wide.txt".to_owned(), ReviewSide::Old);

        let pinned =
            whole_files_for(repo.path(), 1 << 20, &threads, &content, Some("main"), Some(&base))
                .await;
        assert!(pinned.contains_key(&key), "the pinned base places the anchor");
        assert!(
            pinned[&key].lines.iter().any(|line| line.as_str() == "wide 2"),
            "and the file it read is the merge-base's",
        );

        let by_name =
            whole_files_for(repo.path(), 1 << 20, &threads, &content, Some("main"), Some(&moved))
                .await;
        assert_eq!(
            by_name[&key].lines.get(1).map(String::as_str),
            Some("wide 1"),
            "the moved base reads a shifted line 2, which is the anchor's own proof it moved",
        );
    }

    /// **A line the hunks do not carry is answered by the whole file alone.**
    /// A repeated block inside a hunk offers a content-and-neighbours match
    /// for an outside line that reads the same; the bounded content must not
    /// place the comment there.
    #[tokio::test]
    async fn the_whole_file_wins_for_a_line_no_hunk_carries() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let repo = tempfile::tempdir().expect("a repository directory");
        init_repo(repo.path());
        let body = |target: &str| -> String {
            let mut lines: Vec<String> = (1..=30).map(|n| format!("filler {n}\n")).collect();
            lines[1] = "alpha\n".to_owned();
            lines[2] = "beta\n".to_owned();
            lines[3] = "gamma\n".to_owned();
            lines[17] = "alpha\n".to_owned();
            lines[18] = "beta\n".to_owned();
            lines[19] = "gamma\n".to_owned();
            lines[20] = format!("{target}\n");
            lines.concat()
        };
        std::fs::write(repo.path().join("wide.txt"), body("target")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "base"]);
        git_in(repo.path(), &["checkout", "-q", "-b", "worktree-pr"]);
        std::fs::write(repo.path().join("wide.txt"), body("target changed")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "branch"]);
        for held in [
            // Outside the hunk (which covers 18..=24), and its text plus
            // neighbours also match the repeated block inside it.
            wide_thread("outside", ReviewSide::New, 3, "beta", &["alpha", "gamma"], None),
            // And the hunk's own copy still places there.
            wide_thread("inside", ReviewSide::New, 19, "beta", &["alpha", "gamma"], None),
        ] {
            assert!(
                fleet.workspace().upsert_review_thread("proj", "worktree-pr", held),
                "the thread stores",
            );
        }
        let encoded = encode_repo(&fleet, &seat, repo.path()).await;
        assert_eq!(
            landing(&encoded, "outside"),
            serde_json::json!({"state": "placed", "line": 3, "moved_from": null}),
            "the outside line is its own, not the hunk's lookalike: {encoded}",
        );
        assert_eq!(
            landing(&encoded, "inside"),
            serde_json::json!({"state": "placed", "line": 19, "moved_from": null}),
            "a line the hunks carry still places in the hunks: {encoded}",
        );
    }

    /// **The old side's read follows the rename.** A file renamed with edits
    /// keeps its old content at the old path in the base tree; reading the
    /// new path there would park an old-side comment Gone though its line is
    /// untouched.
    #[tokio::test]
    async fn the_record_reads_a_renamed_files_old_side_at_its_old_path() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let repo = tempfile::tempdir().expect("a repository directory");
        init_repo(repo.path());
        std::fs::write(repo.path().join("wide.txt"), wide_lines("wide 10")).expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "base"]);
        git_in(repo.path(), &["checkout", "-q", "-b", "worktree-pr"]);
        git_in(repo.path(), &["mv", "wide.txt", "renamed.txt"]);
        std::fs::write(repo.path().join("renamed.txt"), wide_lines("wide 10 changed"))
            .expect("write");
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "rename"]);
        let mut held =
            wide_thread("revealed-old", ReviewSide::Old, 2, "wide 2", &["wide 1", "wide 3"], None);
        held.anchor.path = "renamed.txt".to_owned();
        assert!(fleet.workspace().upsert_review_thread("proj", "worktree-pr", held), "stores");
        let encoded = encode_repo(&fleet, &seat, repo.path()).await;
        assert_eq!(
            landing(&encoded, "revealed-old"),
            serde_json::json!({"state": "placed", "line": 2, "moved_from": null}),
            "the old side lives at the old path, unchanged: {encoded}",
        );
    }

    /// **The whole-file reads are bounded per record.** Ten comments outside
    /// their hunks ask for ten files; the encode reads eight and leaves the
    /// rest to the bounded content, so a heavily commented branch cannot cost
    /// an unbounded pile of git subprocesses on every seat read.
    #[tokio::test]
    async fn the_whole_file_reads_are_bounded_per_record() {
        let repo = tempfile::tempdir().expect("a repository directory");
        init_repo(repo.path());
        let mut files = Vec::new();
        let mut threads = Vec::new();
        for n in 0..10 {
            let path = format!("f{n}.txt");
            std::fs::write(repo.path().join(&path), "one\ntwo\nthree\n").expect("write");
            files.push(FileContent {
                path: path.clone(),
                old_path: None,
                status: forge_workspace::env::git_diff::hunks::FileStatus::Modified,
                binary: false,
                submodule: false,
                truncated: false,
                hunks: vec![forge_workspace::env::git_diff::hunks::Hunk {
                    old_start: 3,
                    old_count: 1,
                    new_start: 3,
                    new_count: 1,
                    lines: vec![forge_workspace::env::git_diff::hunks::DiffLine {
                        kind: forge_workspace::env::git_diff::hunks::DiffLineKind::Context,
                        text: "three".to_owned(),
                        old_line: Some(3),
                        new_line: Some(3),
                    }],
                }],
            });
            threads.push(ReviewThread {
                id: format!("t{n}"),
                anchor: ReviewAnchor {
                    path,
                    side: ReviewSide::New,
                    line: 2,
                    content_hash: forge_workspace::env::git_diff::resolver::anchor_hash("two"),
                    context: vec!["one".to_owned(), "three".to_owned()],
                    base_ref: "main".to_owned(),
                },
                comments: Vec::new(),
                status: forge_primitives::review::ReviewStatus::Open,
                created_at: "2026-01-01T00:00:00Z".to_owned(),
                updated_at: "2026-01-01T00:00:00Z".to_owned(),
                commit: None,
            });
        }
        git_in(repo.path(), &["add", "-A"]);
        git_in(repo.path(), &["commit", "-qm", "base"]);
        let content = DiffContent {
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Populated(files),
            base: Some(forge_primitives::git_diff::BaseState::Current),
            base_error: None,
            truncated: false,
        };
        let whole = whole_files_for(
            repo.path(),
            10 * 1024 * 1024,
            &threads,
            &content,
            Some("main"),
            Some("main"),
        )
        .await;
        assert_eq!(whole.len(), 8, "the encode stops at its eight-file bound");
        assert!(
            !whole.contains_key(&("f9.txt".to_owned(), ReviewSide::New)),
            "and the threads past it keep the bounded content's answer",
        );
    }

    /// A read cut before the anchored line cannot answer: the record carries
    /// no landing rather than calling the code gone.
    #[test]
    fn a_read_cut_before_the_anchor_answers_no_landing() {
        let anchor = ReviewAnchor {
            path: "wide.txt".to_owned(),
            side: ReviewSide::New,
            line: 5,
            content_hash: forge_workspace::env::git_diff::resolver::anchor_hash("wide 5"),
            context: Vec::new(),
            base_ref: "main".to_owned(),
        };
        let cut = WholeFiles::from([(
            ("wide.txt".to_owned(), ReviewSide::New),
            WholeFile { lines: vec!["wide 1".to_owned()], truncated: true },
        )]);
        assert!(
            resolution_wire(&anchor, &[], true, Some(&cut)).is_none(),
            "a line past the cut is unresolved, not gone",
        );
        let whole = WholeFiles::from([(
            ("wide.txt".to_owned(), ReviewSide::New),
            WholeFile { lines: vec!["wide 1".to_owned()], truncated: false },
        )]);
        assert!(
            matches!(
                resolution_wire(&anchor, &[], true, Some(&whole)),
                Some(ThreadResolutionWire::Outdated { reason: OutdatedReasonWire::Gone })
            ),
            "a complete read that does not carry the line is gone",
        );
    }

    /// A commit-scoped thread must not borrow the branch's whole-file read:
    /// its line counts in its own commit's numbering.
    #[test]
    fn a_commit_scoped_thread_does_not_borrow_the_branch_read() {
        let anchor = wide_thread("commit", ReviewSide::New, 2, "wide 2", &["wide 1"], None).anchor;
        let whole = WholeFiles::from([(
            ("wide.txt".to_owned(), ReviewSide::New),
            WholeFile { lines: vec!["wide 1".to_owned(), "wide 2".to_owned()], truncated: false },
        )]);
        assert!(
            matches!(
                resolution_wire(&anchor, &[], false, None),
                Some(ThreadResolutionWire::Outdated { reason: OutdatedReasonWire::Gone })
            ),
            "with no read of its own, the commit's thread is not placed by the branch's",
        );
        assert!(
            matches!(
                resolution_wire(&anchor, &[], false, Some(&whole)),
                Some(ThreadResolutionWire::Placed { line: 2, .. })
            ),
            "and the same anchor would have been placed had it been eligible",
        );
    }

    /// A POPULATED PR row, from a seeded scan rather than from `gh`.
    ///
    /// The key's presence is not enough on its own: a regression that answers
    /// no PR - a wrong directory handed to the read, say - keeps the key and
    /// passes that. This drives the values a client draws.
    #[tokio::test]
    async fn the_pr_row_carries_the_scans_pr_and_its_closing_issues() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let surface = fleet.surface();
        let cwd = surface.roster().cwd_for(&seat).expect("the seat has a directory");
        surface.store_work_snapshot(
            &seat,
            forge_workspace::work::WorkSnapshot {
                diff: scanned(),
                cwd: std::path::PathBuf::from(&cwd),
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );
        let work = Arc::new(WorkCache::new());
        let state = TransportState {
            surface,
            work,
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Session(seat)).await.expect("encode");

        assert_eq!(encoded["pr"]["number"], 1249, "the open PR crosses with its number: {encoded}");
        assert_eq!(encoded["closes"][0]["number"], 1215, "and the issues it closes: {encoded}");
        assert_eq!(
            encoded["work"]["branch"], "worktree-pr",
            "and the branch comes from the SAME scan as the PR, so the two cannot disagree: \
             {encoded}"
        );
        assert_eq!(encoded["work"]["changed"], 3, "with the count that scan took: {encoded}");
    }

    /// The pushed row and the record's own fields are one answer.
    ///
    /// Both are the seat's store read the same way, so a client that applies
    /// the update lands on exactly what a client that read the record holds -
    /// a second derivation is the only way the two could ever disagree.
    #[tokio::test]
    async fn the_pushed_row_is_the_records_own_fields() {
        // Its OWN fleet rather than `fixture_state`'s: what this pins is a
        // snapshot built by `scanned()`, not the fixture's own tree.
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let seat = fixture_seat();
        let surface = fleet.surface();
        let cwd = surface.roster().cwd_for(&seat).expect("the seat has a directory");
        surface.store_work_snapshot(
            &seat,
            forge_workspace::work::WorkSnapshot {
                diff: scanned(),
                cwd: std::path::PathBuf::from(&cwd),
                read_at: std::time::Instant::now(),
                base: None,
                base_at: None,
                base_error: None,
            },
        );
        let state = TransportState {
            surface,
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded =
            encode_subject(&state, &Subject::Session(seat.clone())).await.expect("encode");
        let held = state.surface.work(&seat, std::path::Path::new(&cwd), false).await;
        let pushed = forge_workspace::work::work_from_scan(&held.diff, &held.cwd);

        assert_eq!(
            encoded["work"],
            serde_json::to_value(&pushed).expect("encode"),
            "the row the seat pushes is the row its record answers",
        );
        assert_eq!(
            encoded["pr"],
            serde_json::to_value(&held.diff.pr).expect("encode"),
            "and so is the PR beside it",
        );
        assert_eq!(
            encoded["closes"],
            serde_json::to_value(&held.diff.closes).expect("encode"),
            "and the issues it closes",
        );
        assert_eq!(
            encoded["git"],
            serde_json::to_value(forge_workspace::work::git_work_view(&held.diff, None, None))
                .expect("encode"),
            "and the tree behind the row's depth",
        );
    }

    /// A scan as a view's inspector reads it: one branch, one worktree
    /// layer, one chain ahead, one PR and one closing issue. Both depth
    /// layers are populated so the record-vs-constructor guard covers both
    /// axes - a derivation omitting either reddens.
    fn scanned() -> GitDiffSnapshot {
        use forge_primitives::git::{GitBranch, GitCommit, GitIssueRef, GitPrInfo};
        use forge_primitives::git_diff::{GitBranchAhead, GitDiffStats, LayerState, RepoGate};
        GitDiffSnapshot {
            branch: GitBranch::Named("worktree-pr".to_owned()),
            pushed_sha: Some("abc123".to_owned()),
            pr_fetched_at: None,
            default_branch: Some("main".to_owned()),
            repo_gate: RepoGate::InRepo,
            worktree: LayerState::Populated(GitDiffStats {
                files: Vec::new(),
                total_files: 3,
                total_added: 9,
                total_removed: 2,
            }),
            branch_ahead: LayerState::Populated(GitBranchAhead {
                commit_count: 1,
                stats: GitDiffStats {
                    files: Vec::new(),
                    total_files: 0,
                    total_added: 0,
                    total_removed: 0,
                },
                commits: vec![GitCommit {
                    sha: "a1b2c3d".to_owned(),
                    subject: "the branch's own commit".to_owned(),
                    stats: GitDiffStats::default(),
                    time: 1_766_000_000,
                }],
            }),
            pr: Some(GitPrInfo {
                number: 1249,
                url: "https://example.test/pull/1249".to_owned(),
                draft: false,
            }),
            closes: vec![GitIssueRef {
                number: 1215,
                url: "https://example.test/issues/1215".to_owned(),
            }],
        }
    }

    /// The token/cost report, on the `usage` subject.
    /// Nothing carried it: the pool is scanned off-thread by the server,
    /// and a client had no way to ask for the same numbers.
    ///
    /// Its own fleet rather than `fixture_state`'s, whose transcript pool the
    /// usage scan reads for a whole record: what this pins is a pool of one
    /// transcript.
    #[tokio::test]
    async fn a_usage_subscription_is_answered_with_the_pools_report() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        // A stamp in the past on purpose: a record stamped "now" flips which
        // rolling windows hold it at midnight, so the assertion below would be
        // a coin toss once a day.
        fleet
            .seed_transcript(
                "TestOrg",
                "proj",
                "lead",
                &[r#"{"type":"assistant","timestamp":"2025-01-02T03:04:05.000Z","message":{"id":"m-usage","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"counted"}],"usage":{"input_tokens":7,"output_tokens":11}}}"#],
            )
            .expect("the transcript seeds");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Usage).await.expect("encode");

        assert_eq!(
            encoded["lifetime"]["total"]["input"], 7,
            "the pool's own count crosses, so the read is the scanner's: {encoded}"
        );
        assert_eq!(
            encoded["today"]["total"]["input"], 0,
            "and a record from a past day is in no rolling window: {encoded}"
        );
        assert!(
            encoded.get("pricing_available").is_some(),
            "and the flag that blanks costs rather than reading a misleading zero: {encoded}"
        );
    }

    /// **The served page caps an oversized `tool_use_result` (#2374), and
    /// only there.** The held copy keeps the record whole - the transcript
    /// and the visible block are untouched - while the page replaces it
    /// with an `omitted` marker; a record under the cap crosses as it is.
    #[tokio::test]
    async fn the_served_page_caps_an_oversized_tool_record() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        let huge = "x".repeat(100_000);
        let frame = |uuid: &str, record: Value| -> Message {
            serde_json::from_value(serde_json::json!({
                "type": "user",
                "uuid": uuid,
                "message": {"role": "user", "content": "the call"},
                "tool_use_result": record,
                "session_id": "s",
            }))
            .expect("a user frame")
        };
        let big = frame("b1", serde_json::json!({"payload": huge}));
        let small = frame("s1", serde_json::json!({"answers": {"q": "a"}}));
        state.conversations.insert(
            &fixture_seat(),
            crate::transport::conversation::Conversation::new(
                vec![big.clone(), small.clone()],
                0,
                0,
            ),
        );

        let (value, _) = encode_subject_timed(&state, &Subject::Session(fixture_seat()), false)
            .await
            .expect("the seat encodes");
        let turns = value["conversation"]["turns"].as_array().expect("the page carries turns");
        let frame = |uuid: &str| -> &Value {
            turns
                .iter()
                .flat_map(|turn| turn["messages"].as_array().into_iter().flatten())
                .find(|frame| frame["uuid"] == uuid)
                .unwrap_or_else(|| panic!("the page carries the frame {uuid}"))
        };
        let capped = &frame("b1")["tool_use_result"];
        assert_eq!(capped["omitted"], true, "the big record is marked: {capped}");
        assert!(
            capped["bytes"].as_u64().is_some_and(|bytes| bytes > 100_000),
            "and says how much it weighed: {capped}",
        );
        assert_eq!(
            frame("s1")["tool_use_result"],
            serde_json::json!({"answers": {"q": "a"}}),
            "a record under the cap crosses as it is",
        );

        // The page is bounded by the cap rather than by the record.
        let conversation = serde_json::to_vec(&value["conversation"]).expect("serialises");
        assert!(
            conversation.len() < 30_000,
            "the served conversation is bounded: {} bytes",
            conversation.len(),
        );

        // **And the held copy still carries it whole**: the cap is the
        // page's, not the transcript's.
        let held = state.conversations.get(&fixture_seat()).expect("the seat is held");
        let kept = held.lock().messages()[0].clone();
        let Message::User { tool_use_result, .. } = kept else {
            panic!("the held frame is the user frame it was seeded with");
        };
        assert_eq!(
            tool_use_result.expect("the held copy keeps its record")["payload"]
                .as_str()
                .map(str::len),
            Some(100_000),
            "the held copy is untouched by the page's cap",
        );
    }

    /// **The cap keeps what a client draws before it drops the record**
    /// (#2374). A Write update carries the whole file in
    /// `content`/`originalFile`, which nothing draws; keeping the patch and
    /// the outside-edit mark leaves the row intact under the cap.
    #[test]
    fn an_oversized_record_keeps_the_keys_a_client_draws() {
        let patch = serde_json::json!([
            {"oldStart": 1, "oldLines": 1, "newStart": 1, "newLines": 1, "lines": ["-one", "+two"]}
        ]);
        let mut record = serde_json::json!({
            "content": "x".repeat(100_000),
            "originalFile": "x".repeat(100_000),
            "structuredPatch": patch,
            "userModified": true,
        });
        let report = cap_tool_record(&mut record, 64 * 1024).expect("the record is over the cap");
        assert_eq!(
            report.kept, "structuredPatch,userModified",
            "the drawn keys fit, so the record was reduced to them",
        );
        assert_eq!(record["structuredPatch"], patch, "the drawn patch survives the cap: {record}");
        assert_eq!(record["userModified"], true, "and the outside-edit mark with it: {record}");
        assert!(record.get("content").is_none(), "the unread whole file is dropped: {record}");
        assert!(record.get("omitted").is_none(), "nothing was omitted: {record}");
    }

    /// The reduction keeps a question's drawn keys too: `answers` and
    /// `annotations` survive an over-cap record, so the card still draws what
    /// was picked and typed (#2374).
    #[test]
    fn an_oversized_question_record_keeps_its_answers() {
        let mut record = serde_json::json!({
            "answers": {"Which one?": "the second"},
            "annotations": {"Which one?": {"notes": "because"}},
            "debug": "x".repeat(100_000),
        });
        cap_tool_record(&mut record, 64 * 1024);
        assert_eq!(
            record["answers"]["Which one?"], "the second",
            "the picked answer survives the cap: {record}",
        );
        assert_eq!(
            record["annotations"]["Which one?"]["notes"], "because",
            "and the typed note with it: {record}",
        );
        assert!(record.get("omitted").is_none(), "nothing was omitted: {record}");
    }

    /// **A question whose notes are what pushed it over keeps its answer**
    /// (#2374): the drawn keys do not fit together, so the reduction keeps
    /// `answers` alone and the live and transcript pages draw the same card.
    #[test]
    fn a_question_over_the_cap_keeps_its_answer_and_drops_the_notes() {
        let mut record = serde_json::json!({
            "answers": {"Which one?": "the second"},
            "annotations": {"Which one?": {"notes": "x".repeat(100_000)}},
        });
        let report = cap_tool_record(&mut record, 64 * 1024).expect("the record is over the cap");
        assert_eq!(
            record["answers"]["Which one?"], "the second",
            "the answer survives the cap: {record}",
        );
        assert!(record.get("annotations").is_none(), "the notes are dropped: {record}");
        assert_eq!(report.kept, "answers", "and the log says what was kept");
    }

    /// **A capped record is logged against the call it capped** (#2374): the
    /// seat, the call's id, the record's own size, the cap and what survived,
    /// so "where did this payload go" is answered by one line. Both entry
    /// points are pinned, and a record under the cap logs nothing.
    #[test]
    fn a_capped_record_is_logged_with_its_seat_call_and_size() {
        use crate::test_support::logged;

        let seat = fixture_seat();
        let record = serde_json::json!({
            "structuredPatch": [
                {"oldStart": 1, "oldLines": 1, "newStart": 1, "newLines": 1,
                 "lines": ["-one", format!("+{}", "x".repeat(100_000))]}
            ],
            "userModified": true,
        });
        let bytes = serde_json::to_string(&record).expect("serialises").len();
        let mut turns = vec![TurnWire {
            key: None,
            messages: vec![serde_json::json!({
                "type": "user",
                "message": {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": "toolu_edit", "content": "ok"}
                ]},
                "tool_use_result": record.clone(),
            })],
        }];
        let events = logged(|| cap_tool_records(&mut turns, 64 * 1024, &seat));
        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::DEBUG
                    && fields.contains(r#"event_name="tool_record_capped""#)
                    && fields.contains(&format!("slot={}", seat.display()))
                    && fields.contains(r#"tool_use_id="toolu_edit""#)
                    && fields.contains(&format!("bytes={bytes}"))
                    && fields.contains("cap=65536")
                    && fields.contains(r#"disposition="omitted""#)
            }),
            "the page's cap is recorded against the call it capped: {events:?}",
        );

        // The live path has its own call site, so dropping one and keeping the
        // other must not read green.
        let live = |record: &Value| -> Message {
            serde_json::from_value(serde_json::json!({
                "type": "user",
                "message": {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": "toolu_live", "content": "ok"}
                ]},
                "session_id": "s1",
                "tool_use_result": record.clone(),
            }))
            .expect("the frame decodes")
        };
        let mut msg = live(&record);
        let events = logged(|| cap_message_record(&mut msg, 64 * 1024, &seat));
        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::DEBUG
                    && fields.contains(r#"event_name="tool_record_capped""#)
                    && fields.contains(r#"tool_use_id="toolu_live""#)
            }),
            "the live path records its own cap: {events:?}",
        );

        let mut under = live(&serde_json::json!({"payload": "x"}));
        let events = logged(|| cap_message_record(&mut under, 64 * 1024, &seat));
        assert!(
            events.is_empty(),
            "a record under the cap says nothing, and so logs nothing: {events:?}",
        );
    }

    /// **A record exactly at the cap crosses whole**: the bound is `<=`, so a
    /// page does not mark a record it fully carries (#2374).
    #[test]
    fn a_record_at_the_cap_is_not_capped() {
        let record = serde_json::json!({"payload": "x".repeat(100_000)});
        let exact = serde_json::to_string(&record).expect("serialises").len();
        let mut at = record.clone();
        assert!(
            cap_tool_record(&mut at, exact).is_none(),
            "a record at the cap is served as it is, and nothing is logged",
        );
        assert_eq!(at, record, "a record at the cap is served as it is");
        let mut over = record.clone();
        let report = cap_tool_record(&mut over, exact - 1).expect("one byte over is capped");
        assert_eq!(report.kept, "", "nothing drawable fits, so the marker replaced it");
        assert_eq!(report.bytes, exact, "the log line carries the record's own size");
        assert_eq!(over["omitted"], true, "one byte over and it is marked: {over}");
    }

    /// When even the drawn keys are over the cap the record becomes the
    /// marker, and `userModified` rides it so the row still says the file
    /// changed outside the edit (#2374).
    #[test]
    fn a_record_over_the_cap_after_reduction_is_the_marker() {
        let mut record = serde_json::json!({
            "structuredPatch": [{"lines": ["x".repeat(100_000)]}],
            "userModified": true,
        });
        cap_tool_record(&mut record, 64 * 1024);
        assert_eq!(record["omitted"], true, "the record is the marker: {record}");
        assert_eq!(record["userModified"], true, "and keeps the outside-edit mark: {record}");
        assert!(
            record["bytes"].as_u64().is_some_and(|bytes| bytes > 100_000),
            "and says how much it weighed: {record}",
        );
    }

    /// **The open path carries its own numbers (#2374).** A session encode is
    /// where a client's switch waits on this half, so the slices of that wait
    /// - the page's cut and the record's encode - ride back with the value;
    /// with the client's own marks, a slow switch is then placed from the
    /// artefacts rather than guessed at.
    #[tokio::test]
    async fn the_open_path_carries_its_own_timings() {
        let root = tempfile::tempdir().expect("a temp dir");
        let state = fixture_state(root.path()).await;

        let (_, timings) = encode_subject_timed(&state, &Subject::Session(fixture_seat()), false)
            .await
            .expect("the seat encodes");

        assert!(timings.page_ms > 0.0, "the page's cut is work on the path: {timings:?}");
        assert!(timings.encode_ms > 0.0, "the record's encode is work on the path: {timings:?}");
        assert_eq!(timings.seed_ms, 0.0, "a HELD copy pays no transcript read: {timings:?}");
        // **An ordinary read neither refreshes the base nor re-scans for one.**
        // Both slices are on the path and both are empty here; a read that
        // reached the network or re-scanned would say so in them.
        assert!(
            timings.base_ms < 1.0,
            "no child runs for an ordinary read's base (a git call costs whole ms): {timings:?}",
        );
        assert!(timings.rescan_ms < 1.0, "and nothing re-scans the tree for one: {timings:?}");
    }

    /// The other half: a seat the transport holds nothing for pays the
    /// transcript read, and the number says so.
    ///
    /// **A transcript big enough that the read dominates its own call**, so
    /// the assertion below has teeth: the seed is a whole file's read and the
    /// rest of `conversation_for` is a lookup and an ask, and the two must
    /// not overlap - a total that swallowed the read is short by the whole
    /// read on exactly the slow cold open the line exists to explain.
    #[tokio::test]
    async fn a_cold_seats_seed_is_timed_outside_the_conversation_total() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        let rows: Vec<String> = (0..3_000).map(a_turns_rows).collect();
        let rows: Vec<&str> = rows.iter().map(String::as_str).collect();
        fleet.seed_transcript("TestOrg", "proj", "lead", &rows).expect("the transcript seeds");
        // An empty transport over that fleet: the seed path is what a seat
        // nobody holds takes.
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let (_, timings) = encode_subject_timed(&state, &Subject::Session(fixture_seat()), false)
            .await
            .expect("the seat encodes");

        assert!(
            timings.seed_ms > 0.0,
            "the transcript read a cold seat takes is timed: {timings:?}"
        );
        assert!(
            timings.conversation_ms < timings.seed_ms,
            "the conversation total EXCLUDES the seed, so a read that took real time must \
             dominate the lookup around it - the two overlapping is the bug this pins: {timings:?}"
        );
    }

    /// The Extensions read crosses with the inventory, the PATH checks and
    /// the run, and the inventory half is **the same shape the live frames
    /// carry** - a client that took either would be drawing the same bytes
    /// (rule 25).
    #[tokio::test]
    async fn an_extensions_subscription_is_answered_with_the_inventory() {
        let fleet = crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("fleet");
        let mut components = forge_primitives::plugins::PluginComponents {
            plugin: "demo@probe-market".to_owned(),
            marketplace: "probe-market".to_owned(),
            installed: true,
            enabled: true,
            version: Some("1.0.0".to_owned()),
            available_version: Some("1.1.0".to_owned()),
            skills: vec!["one".to_owned()],
            ..forge_primitives::plugins::PluginComponents::default()
        };
        components.lsp_servers.insert("absent-server".to_owned(), "not-a-binary-xyz".to_owned());
        let snapshot = forge_primitives::plugins::PluginsInventorySnapshot {
            components: vec![components],
            ..forge_primitives::plugins::PluginsInventorySnapshot::default()
        };
        let pinned = snapshot.clone();
        fleet.workspace().set_plugin_scan_for_test(move || Some(pinned.clone()));
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Extensions).await.expect("encode");

        assert_eq!(
            encoded["snapshot"]["components"][0]["plugin"], "demo@probe-market",
            "the inventory crosses: {encoded}"
        );
        assert_eq!(
            encoded["snapshot"]["components"][0]["available_version"], "1.1.0",
            "with the version the marketplace reports"
        );
        assert_eq!(
            encoded["snapshot"],
            serde_json::to_value(&snapshot).expect("value"),
            "the snapshot half is byte-for-byte the object the Plugins frames embed"
        );
        assert_eq!(
            encoded["lsp"][0]["server"], "absent-server",
            "the PATH check crosses beside it"
        );
        assert_eq!(
            encoded["lsp"][0]["on_path"], false,
            "and reflects this machine rather than guessing"
        );
        assert!(encoded["update_run"].is_null(), "no run has happened this process");

        // The read's own type accepts its own bytes back.
        let wire: ExtensionsWire = serde_json::from_value(encoded).expect("round trip");
        assert_eq!(wire.snapshot.expect("snapshot").components.len(), 1);
        assert_eq!(wire.lsp.len(), 1);
    }

    /// What one fixture pins, and what it only appears to.
    ///
    /// The byte-compare above makes a renamed field fail, and it is silent
    /// about how much it is covering. A key the fixture carries as `null`
    /// pins the key's existence and nothing behind it; an ARRAY it carries
    /// empty pins nothing about the element shape at all, so a field renamed
    /// inside one is invisible to every pin in the tree. `projects[]/tasks[]`
    /// and `projects[]/crons[]` are on that list: named on every run and not
    /// yet seeded, which is #1367.
    ///
    /// Counted rather than listed, and printed on every run: the number is
    /// what says whether the fixture is the instrument it is taken for.
    #[derive(Default)]
    struct Coverage {
        keys: usize,
        nulls: BTreeSet<String>,
        empty_arrays: BTreeSet<String>,
    }

    fn coverage(value: &Value, path: &str, out: &mut Coverage) {
        match value {
            Value::Object(fields) => {
                for (key, held) in fields {
                    out.keys += 1;
                    let here = format!("{path}/{key}");
                    if held.is_null() {
                        out.nulls.insert(here.clone());
                    }
                    coverage(held, &here, out);
                }
            }
            Value::Array(items) => {
                if items.is_empty() {
                    out.empty_arrays.insert(format!("{path}[]"));
                }
                for item in items {
                    coverage(item, &format!("{path}[]"), out);
                }
            }
            _ => {}
        }
    }

    #[tokio::test]
    async fn every_subject_round_trips_through_its_fixture() {
        let dir = fixture_dir();
        let state = fixture_state(dir.path()).await;
        let forms = volatile(dir.path());
        let mut keys = 0usize;
        let mut nulls = 0usize;
        let mut empty: BTreeSet<String> = BTreeSet::new();

        for (subject, fixture) in fixtures() {
            let mut encoded = encode_subject(&state, &subject).await.expect("encode");
            normalise(&mut encoded, &forms);
            let expected: Value =
                serde_json::from_str(&std::fs::read_to_string(&fixture).expect("read"))
                    .expect("parse");
            assert_eq!(encoded, expected, "{} changed shape", fixture.display());

            let mut pin = Coverage::default();
            coverage(&encoded, "", &mut pin);
            keys += pin.keys;
            nulls += pin.nulls.len();
            empty.extend(pin.empty_arrays);
        }

        // The floor: an empty fixture compares equal to a snapshot of nothing
        // and would otherwise report as clean as a populated one.
        assert!(keys > 0, "no fixture carries a key, so there is nothing to pin");
        eprintln!(
            "wire fixtures: {keys} key slots carried, {nulls} null (the key is named, nothing \
             behind it is pinned), {} empty collections (no element shape pinned anywhere): {}",
            empty.len(),
            empty.iter().cloned().collect::<Vec<_>>().join(", "),
        );
    }

    /// The push-to-talk key and its press mapping reach the composer's record.
    ///
    /// A client draws the affordance the keyboard drives, and this is its only
    /// read of what `forge.toml` configured: without it the key is whatever
    /// the client hardcodes, which is the same for every install.
    #[tokio::test]
    async fn the_composer_carries_the_configured_bind_and_mode() {
        let dir = tempfile::tempdir().expect("tempdir");
        let forge = dir.path().join("forge");
        std::fs::create_dir_all(&forge).expect("forge/");
        let project = dir.path().join("proj");
        std::fs::create_dir_all(&project).expect("the project's directory");
        std::fs::write(
            forge.join("forge.toml"),
            format!(
                "\
[[orgs]]
name = \"TestOrg\"
accounts = [\"Acct\"]

[[orgs.projects]]
name = \"proj\"
path = \"{}\"

[[accounts]]
display_name = \"Acct\"
token = \"t\"
models = [\"claude-sonnet-5\"]
provider = \"anthropic\"

[dictate]
enabled = true
bind = \"left_cmd\"
mode = \"toggle\"
",
                project.display()
            ),
        )
        .expect("write forge.toml");
        let workspace = Arc::new(
            forge_workspace::Workspace::new_for_test(dir.path().to_owned()).expect("workspace"),
        );
        let state = TransportState {
            surface: Arc::new(ViewSurface::new(workspace)),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let session =
            encode_subject(&state, &Subject::Session(fixture_seat())).await.expect("encode");
        assert_eq!(
            session["composer"]["bind"], "left_cmd",
            "the composer carries the configured push-to-talk key, not a default: {session}",
        );
        assert_eq!(
            session["composer"]["mode"], "toggle",
            "and how a press maps onto a take, not a default either: {session}",
        );
    }

    /// **A landed take nobody has sent is re-handed by the record** (#1360).
    ///
    /// The landing reaches its dictating connection live, and nothing
    /// retained it: a page that reloaded between the landing and the send
    /// lost the words. The seat holds them now, and the record says they are
    /// the seat's - `carried` - so any page may place them without reading
    /// them as a landing of its own.
    #[tokio::test]
    async fn the_composer_re_hands_a_landing_nobody_has_sent() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };
        let seat = fixture_seat();
        let landing = crate::SessionUpdate::DictateEnded {
            key: seat.clone(),
            outcome: crate::surface::DictateOutcome::Landed {
                text: "the unsent words".to_owned(),
                truncated: true,
            },
            generation: 7,
            surface: Some("review".to_owned()),
            initiator: None,
        };

        let session =
            encode_subject(&state, &Subject::Session(seat.clone())).await.expect("encode");
        assert!(session["composer"]["notice"].is_null(), "nothing landed yet: {session}");

        crate::live::Live::lock(&state.live).apply(&landing);
        let session =
            encode_subject(&state, &Subject::Session(seat.clone())).await.expect("encode");
        assert_eq!(
            session["composer"]["notice"]["text"], "the unsent words",
            "the unsent landing is on the record: {session}",
        );
        assert_eq!(
            session["composer"]["notice"]["generation"], 7,
            "with the take's own number, the landing memory's discriminator",
        );
        assert_eq!(
            session["composer"]["notice"]["truncated"], true,
            "and whether the take was partial, which the client draws a warn line from",
        );
        assert_eq!(
            session["composer"]["notice"]["carried"], true,
            "and marked as the seat's re-hand rather than a landing of this page's own",
        );
        assert_eq!(
            session["composer"]["notice"]["surface"], "review",
            "and naming the surface that began it, so only that surface places it",
        );

        let msg: forge_primitives::Message = serde_json::from_value(serde_json::json!({
            "type": "user",
            "session_id": "s1",
            "message": {"role": "user", "content": "about to be sent"},
        }))
        .expect("a user frame");
        crate::live::Live::lock(&state.live).apply(&crate::SessionUpdate::ChatAppended {
            key: seat.clone(),
            msg,
            source: Some(forge_workspace::PromptSource::You),
        });
        let session = encode_subject(&state, &Subject::Session(seat)).await.expect("encode");
        assert!(
            session["composer"]["notice"].is_null(),
            "and a prompt takes the words with it, so sent words are never handed back: {session}",
        );
    }

    /// The account a project's row chips, and its state.
    ///
    /// The state is derived from the project's model and the account pool, so
    /// `has_model` is not enough to draw it from - a client could draw the row
    /// and not the chip on it.
    ///
    /// The fixture's projects declare no model, so nothing binds and the chip
    /// is absent here: what this pins is the field being on the row at all,
    /// and the POPULATED case is the acceptance walk's, against a config whose
    /// projects do declare one.
    #[tokio::test]
    async fn a_projects_row_carries_the_chip_field() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        let rows = encoded["projects"].as_array().expect("the home carries project rows");

        assert!(rows[0].get("chip").is_some(), "the row carries a chip: {}", rows[0]);
        assert_eq!(
            rows[0]["project"]["has_model"], false,
            "and this fixture's project declares no model, which is why the chip is absent",
        );
    }

    /// A view draws a project's SCHEDULES section from a read the wire
    /// never carried, so a client could create a cron and never see it again.
    #[tokio::test]
    async fn a_projects_schedules_ride_its_board_row() {
        let fleet =
            crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
        fleet.add_cron("proj", "a nightly sweep").expect("the cron is added");
        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Board).await.expect("encode");
        let rows = encoded["projects"].as_array().expect("the board carries project rows");
        let crons = rows[0]["crons"].as_array().expect("and each row carries its schedules");

        assert_eq!(crons.len(), 1, "the project's own cron reached the wire: {crons:?}");
        assert_eq!(crons[0]["prompt"], "a nightly sweep");
    }

    /// A project's connector subscriptions ride its row, which is the read the
    /// section draws.
    ///
    /// The home's own `connectors` carried a `subscriptions` list that no read
    /// could fill - it was built with no project scope, and `None` means "the
    /// liveness facts only" - so a page drawing that list showed the connector
    /// section empty however long it waited, and would have flip-flopped
    /// against the next re-read once a frame patched it.
    #[tokio::test]
    async fn a_projects_connector_subscriptions_ride_its_board_row() {
        let (workspace, _dir) = crate::surface::testing::workspace_with_connector_subs();
        let state = TransportState {
            surface: Arc::new(crate::surface::ViewSurface::new(Arc::clone(&workspace))),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Board).await.expect("encode");
        let rows = encoded["projects"].as_array().expect("the board carries project rows");
        let row = rows
            .iter()
            .find(|row| row["project"]["name"] == "forge")
            .unwrap_or_else(|| panic!("the project has a row: {encoded}"));

        assert_eq!(
            row["connectors"]["gotify"].as_array().map(Vec::len),
            Some(1),
            "the project's own gotify subscription is on its row: {encoded}",
        );
        assert_eq!(
            row["connectors"]["gotify"][0]["applications"][0], "forge",
            "as the subscription itself, not a placeholder",
        );
        assert_eq!(
            row["connectors"]["slack"].as_array().map(Vec::len),
            Some(1),
            "and so is its slack subscription: {encoded}",
        );

        // The home-level `connectors` is the liveness facts and nothing else:
        // a subscription field a stale reader still looks for is a section
        // that draws empty forever.
        let home = encode_subject(&state, &Subject::Home).await.expect("encode");
        assert!(
            home["connectors"]["gotify"].get("subscriptions").is_none(),
            "the home's own connectors carries no subscription list: {home}",
        );
        assert!(
            home["connectors"]["slack"].get("subscriptions").is_none(),
            "for either connector: {home}",
        );
        assert!(
            home["connectors"]["slack"]["connected_workspaces"].is_array(),
            "the facts it does carry are still there: {home}",
        );
    }

    /// A worker's row draws ITS OWN working tree, not its project's.
    ///
    /// The home carried one `work` per project, read at the project's own path
    /// from the lead's seat, so every worker row under it drew that branch and
    /// that count. On a fleet with a worker in a worktree that is a plausible
    /// wrong answer rather than an obviously missing one - `main` on a worker's
    /// row reads as the worker's branch - which is why the client drew no
    /// branch at all on a worker's row rather than another seat's.
    #[tokio::test]
    async fn a_workers_row_carries_its_own_tree_and_not_the_projects() {
        let dir = tempfile::tempdir().expect("tempdir");
        // The trees go up BEFORE the fleet does. A project's registry key is
        // derived from its path and the derivation canonicalises, so a key
        // taken while the directory is absent is not the key taken once it is
        // there - and a worker registered under the first is registered under
        // one nothing looks up again.
        let root = dir.path().join("proj");
        let worktree = root.join(".claude/worktrees/w1");
        repo_at(&root, "main", 1);
        repo_at(&worktree, "worktree-em-dash-sweep", 2);
        // The composed directory is canonical, as the CLI's own worktree
        // root is; a `/var` tempdir would otherwise compare a spelling
        // claude never uses.
        let worktree = std::fs::canonicalize(&worktree).expect("canonical worktree");

        let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
            .expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.add_git_worker("TestOrg", "proj", "w1").expect("the worker is added");

        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        let rows = encoded["agents"].as_array().expect("the home carries agent rows");
        let row_of = |label: &str| {
            rows.iter().find(|row| row["label"] == label).expect("the seat has a row")
        };

        // The premise, so a row with no tree at all cannot pass as a row with
        // the right one: the worker's seat resolves to its own worktree.
        assert_eq!(
            state
                .surface
                .roster()
                .cwd_for(&SessionSlot::worker("TestOrg", "proj", "w1"))
                .as_deref(),
            Some(worktree.as_path()),
            "precondition: the worker's own directory is the worktree",
        );
        assert_eq!(
            row_of("lead")["work"]["branch"],
            "main",
            "the lead's row draws the project's own tree: {encoded}",
        );
        assert_eq!(row_of("lead")["work"]["changed"], 1, "with the count that tree has: {encoded}");
        assert_eq!(
            row_of("w1")["work"]["branch"],
            "worktree-em-dash-sweep",
            "and the worker's row draws its own, which the project's read cannot answer: \
             {encoded}",
        );
        assert_eq!(
            row_of("w1")["work"]["changed"],
            2,
            "with the worker's own count beside it: {encoded}",
        );
    }

    /// A seat forge holds no directory for draws NO tree, rather than the
    /// project's.
    ///
    /// The row is real: a despawned worker leaves its label behind and is
    /// drawn until the label goes, and the registry has nothing to compose a
    /// directory from. `None` is what says so, and the tempting repair - fall
    /// back to `ProjectWire::work` so the cell is not blank - is the bug this
    /// whole read exists to remove, one row narrower.
    #[tokio::test]
    async fn a_seat_with_no_directory_borrows_no_tree() {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path().join("proj");
        repo_at(&root, "main", 1);

        let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
            .expect("the fleet builds");
        fleet.start("TestOrg", "proj").expect("the project starts");
        fleet.add_despawned_worker("proj", "gone").expect("the label is left behind");

        let state = TransportState {
            surface: fleet.surface(),
            work: Arc::new(WorkCache::new()),
            conversations: Arc::new(crate::transport::conversation::Conversations::new()),
            live: Mutex::new(crate::live::Live::new()),
            settings: crate::transport::settings_channel(),
            browser_host: None,
            heartbeat: Heartbeat::default(),
        };

        let encoded = encode_subject(&state, &Subject::Home).await.expect("encode");
        let rows = encoded["agents"].as_array().expect("the home carries agent rows");
        let row_of = |label: &str| {
            rows.iter().find(|row| row["label"] == label).expect("the seat has a row")
        };

        assert_eq!(
            state.surface.roster().cwd_for(&SessionSlot::worker("TestOrg", "proj", "gone")),
            None,
            "precondition: forge holds no directory for the despawned label",
        );
        assert_eq!(
            row_of("gone")["work"],
            Value::Null,
            "so its row draws no tree rather than borrowing the project's: {encoded}",
        );
        assert_eq!(
            row_of("lead")["work"]["branch"],
            "main",
            "while the project's own row still draws the project's tree: {encoded}",
        );
    }

    /// A git repository at `dir` on `branch`, with `changed` uncommitted files.
    ///
    /// `.claude/` is ignored so a worker's worktree, nested inside the
    /// project, does not land in the project's own count as an untracked
    /// directory.
    fn repo_at(dir: &Path, branch: &str, changed: usize) {
        std::fs::create_dir_all(dir).expect("mkdir");
        git(dir, &["init", "-q"]);
        git(dir, &["symbolic-ref", "HEAD", &format!("refs/heads/{branch}")]);
        git(dir, &["config", "user.email", "t@e.com"]);
        git(dir, &["config", "user.name", "T"]);
        git(dir, &["config", "commit.gpgsign", "false"]);
        std::fs::write(dir.join(".gitignore"), ".claude/\n").expect("write");
        std::fs::write(dir.join("tracked.txt"), "one\n").expect("write");
        git(dir, &["add", ".gitignore", "tracked.txt"]);
        git(dir, &["commit", "-qm", "init"]);
        for file in 0..changed {
            std::fs::write(dir.join(format!("changed-{file}.txt")), "x\n").expect("write");
        }
    }

    /// Spawn git with the ambient repo-location variables scrubbed, as the
    /// product's own constructor does: a fixture that skipped it would answer
    /// about a foreign repository when the suite runs under a git hook.
    fn git(dir: &Path, args: &[&str]) {
        let out = std::process::Command::new("git")
            .env_remove("GIT_DIR")
            .env_remove("GIT_WORK_TREE")
            .env_remove("GIT_COMMON_DIR")
            .arg("-C")
            .arg(dir)
            .args(args)
            .output()
            .expect("run git");
        assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
    }
}