forge_workspace/

domain_session.rs

//! Workspace-side per-session state.
//!
//! Holds what workspace itself needs: routing metadata (`AgentHandle`
//! slot, claude-issued `session_id`), the pending-interaction mailbox,
//! and the facts every view reads through the view surface. The rest of
//! the operational state a view renders (lifecycle, cwd, account info)
//! lives on the view's own session record.
//!
//! Two classes are held here rather than folded per view. The turn
//! signal [`DomainSession::runtime_state`] is the one the workspace needs
//! authoritatively for its worker-liveness and prompt-interception
//! guards, and [`DomainSession::background_work`],
//! [`DomainSession::available_commands`] and
//! [`DomainSession::available_agents`] are held so a view arriving late
//! reads them from the core.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use forge_agent::AgentHandle;
use forge_primitives::{
    AvailableAgent, AvailableCommand, AvailableModel, CurrentModel, EffortLevel, McpServerStatus,
    MonitorRecord, PermissionMode, RuntimeSessionState, SessionId,
};

use crate::SessionSlot;
use crate::protocol::PendingInteractionSlot;

/// The MCP servers a session's bridge last reported, and the failure
/// standing beside them when the read did not complete.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct McpServers {
    pub servers: Vec<McpServerStatus>,
    /// Why the read failed, when it did. A failed read carries an empty
    /// server list, so a reader that ignored this would report a session
    /// with no servers connected.
    pub error: Option<String>,
}

/// How full the session's context window is, from the bridge's last
/// answer. Both halves are `Option` because the upstream probe reports
/// them independently.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ContextUsage {
    pub percent: Option<u8>,
    pub max_tokens: Option<u64>,
}

/// A continuation the failure left behind: the words the model will
/// receive, taken by the fire as soon as the failed turn is folded.
#[derive(Debug, Clone)]
pub struct PendingAutoContinue {
    pub reason: String,
}

/// The handoff this session is running, if any.
///
/// The request is parked rather than acted on while a turn runs: a handoff
/// never interrupts one. The judgment turn is forge's own prompt, so the
/// reply it captures is the last assistant text before that turn ends - and
/// only that turn's end finishes the handoff.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct HandoffState {
    /// A request that arrived while a turn was running.
    pub parked: bool,
    /// The note the running turn's exit call carried, which becomes the
    /// brief's judgment half. `None` falls back to the turn's last text.
    pub supplied: Option<String>,
    /// Why this handoff runs, which the row and the record both state.
    pub reason: Option<forge_primitives::HandoffReason>,
    /// A fill reading was asked for at the last turn boundary, and the
    /// trigger is waiting to judge it.
    pub probing: bool,
    /// The reading that fired this handoff, carried to the row.
    pub fill: Option<crate::handoff::FillReading>,
    /// The token of the swap gate this handoff opened, so a build that lost
    /// the seat cannot spawn.
    pub swap_token: Option<u64>,
    /// The judgment turn forge sent is running.
    pub awaiting: bool,
    /// The id the judgment prompt was sent under, so the swap waits for
    /// THAT prompt's turn rather than for whatever turn ends next.
    pub judgment_uuid: Option<String>,
    /// The CLI's own lifecycle frame said the judgment prompt started. A
    /// turn that ends without it is not the judgment turn.
    pub judgment_started: bool,
    /// The last assistant text seen while that turn ran.
    pub judgment: Option<String>,
}

impl HandoffState {
    /// Whether a handoff owns this seat's next turn boundary.
    pub fn in_progress(&self) -> bool {
        self.parked || self.awaiting
    }
}

/// Workspace's owned per-session state. One `DomainSession` per
/// active `SessionTask`. Single writer (the `SessionTask`); accessed
/// via `Arc<parking_lot::Mutex<DomainSession>>` so the `Workspace`
/// can route commands without locking the whole pool.
/// The lifecycle states that settle a queued row.
///
/// Written out rather than derived, and it has to agree with the client's own
/// set (`client/src/session/apply.ts`): the core drops what the client drops.
/// A state neither names leaves the row standing on both sides, which is the
/// fail-safe direction - a pile that keeps a delivered prompt is visibly odd,
/// and one that empties on a word nobody reads is silently wrong.
const SETTLED_STATES: [&str; 5] = ["started", "completed", "cancelled", "discarded", "refused"];

/// The interactions a seat is parked on, oldest first.
///
/// **Ordered because the read answers it, and the dock draws the front.** Two
/// AskUserQuestion calls in one assistant message run in parallel with
/// different tool ids, so a seat can hold several at once, and the rule is
/// first-parked-first-drawn. One Vec rather than a map and an
/// order beside it: a seat holds a handful at most, and there is then no
/// second structure to keep in step.
#[derive(Default)]
pub struct PendingInteractions {
    held: Vec<(String, PendingInteractionSlot)>,
}

impl PendingInteractions {
    /// Park `slot` under `tool_id`. A repeat of an id replaces in place, so a
    /// frame folded twice does not park the same interaction twice.
    pub fn insert(&mut self, tool_id: String, slot: PendingInteractionSlot) {
        if let Some(waiting) = self.held.iter_mut().find(|(id, _)| *id == tool_id) {
            waiting.1 = slot;
            return;
        }
        self.held.push((tool_id, slot));
    }

    pub fn get(&self, tool_id: &str) -> Option<&PendingInteractionSlot> {
        self.held.iter().find(|(id, _)| id == tool_id).map(|(_, slot)| slot)
    }

    pub fn remove(&mut self, tool_id: &str) -> Option<PendingInteractionSlot> {
        let at = self.held.iter().position(|(id, _)| id == tool_id)?;
        Some(self.held.remove(at).1)
    }

    pub fn contains_key(&self, tool_id: &str) -> bool {
        self.held.iter().any(|(id, _)| id == tool_id)
    }

    pub fn len(&self) -> usize {
        self.held.len()
    }

    pub fn is_empty(&self) -> bool {
        self.held.is_empty()
    }

    /// Oldest first.
    pub fn values(&self) -> impl Iterator<Item = &PendingInteractionSlot> {
        self.held.iter().map(|(_, slot)| slot)
    }

    /// Oldest first, together with the ids that key them.
    pub fn drain(&mut self) -> impl Iterator<Item = (String, PendingInteractionSlot)> {
        std::mem::take(&mut self.held).into_iter()
    }
}

/// What one frame says about whether a turn is open, when it says anything.
///
/// Derived from the message in [`liveness_of`], and folded in by
/// [`DomainSession::note_liveness`].
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum TurnLiveness {
    /// The model produced output, so a turn is running.
    Working,
    /// The turn's own `Result`, or the transport's `Error`: it is over.
    Ended,
}

/// What one message says about a turn's liveness, or `None` where it says
/// nothing.
pub fn liveness_of(msg: &forge_primitives::Message) -> Option<TurnLiveness> {
    match msg {
        forge_primitives::Message::Assistant { .. }
        | forge_primitives::Message::ThinkingTokens { .. } => Some(TurnLiveness::Working),
        forge_primitives::Message::Result { .. } | forge_primitives::Message::Error { .. } => {
            Some(TurnLiveness::Ended)
        }
        _ => None,
    }
}

pub struct DomainSession {
    pub key: SessionSlot,
    /// Claude-issued session UUID. `None` until the first `Connected`
    /// event from this session's bridge. Workspace consults this when
    /// dispatching `AgentHandle` calls that route by session id.
    pub session_id: Option<SessionId>,
    /// Agent connection handle bound to this session at spawn time.
    /// `None` for pre-spawn / pre-Connect domains (the cron and gotify
    /// delivery paths register a placeholder handle-less domain before
    /// the spawn, which fills the handle in later).
    pub conn: Option<Arc<AgentHandle>>,
    /// Pending permission/question/elicitation oneshots indexed by the
    /// wire `tool_id` / `elicitation_id`, oldest first. `SessionTask` pops
    /// on `Respond*` commands; bridge inserts on every `*Request` event.
    pub pending_interactions: PendingInteractions,
    /// `--new` boot-wave flag, stamped at spawn time from
    /// `SessionLaunchSettings.force_new`. For a project lead it makes
    /// the Connected-time respawn skip the store lookup
    /// (`resume_existing = None` for every worker), so they come
    /// up fresh alongside their fresh lead. `false` for every non-boot
    /// spawn.
    pub spawned_force_new: bool,
    /// A boot spawn's lead restart note, waiting for the Connected hook
    /// to deliver it as the lead's first turn. `None` for every other
    /// spawn, and cleared by the hook so a reconnect repeats nothing.
    pub pending_lead_resume_note: Option<String>,
    /// Whether the spawn that opened this session wrote the worker's
    /// durable row rather than adopting one that was already there.
    /// Stamped at spawn time from the spawn's role, and read by the
    /// tag-write rollback to decide whether the row goes with the spawn
    /// it is discarding. `false` for every lead and every non-worker.
    pub spawn_wrote_row: bool,
    /// Latest runtime liveness mirrored from the session's
    /// `session_state_changed` wire messages. Operational turn state
    /// otherwise lives on the view's own session record; this one signal is
    /// duplicated here so the workspace's worker-liveness and
    /// prompt-interception guards can tell an in-flight turn from an
    /// idle one (a view's own gate alone can race a just-delivered
    /// peer / cron / gotify / slack prompt). `None` until the first
    /// state message.
    pub runtime_state: Option<RuntimeSessionState>,
    /// Turn committed at `Command::Prompt` routing, ahead of the
    /// wire-lagged `runtime_state`; the guards OR it in.
    ///
    /// **Derived from the commit count, not a bare flag.** A Result ends ONE
    /// turn, and a prompt committed before that Result landed belongs to the
    /// next one - so clearing a flag at the Result would erase the next
    /// turn's stamp and make a running seat read idle. `commit_turn` and
    /// `serve_turn` are the only writers.
    pub turn_pending: bool,
    /// Prompts committed as turns on this occupant.
    pub turn_commits: u64,
    /// Results (or transport errors) that closed one of them.
    pub turns_served: u64,
    /// Frame-derived liveness: the model's own output proves a turn is
    /// running, and a `Result`/`Error` ends it.
    ///
    /// The two mirrors above miss a turn's opening. `turn_pending` covers
    /// only a routed `Command::Prompt`, and one `Result` clears it; and
    /// `runtime_state` no longer arrives ("may never arrive", above). So a
    /// turn whose opening is thinking and prose reads NOT in flight on the
    /// wire until its first tool call, and a client drawing from the header
    /// shows no running row for that stretch - measured at 41 seconds on a
    /// resumed worker (2026-10-09), where the terminal showed it running and
    /// the client did not.
    pub turn_open: bool,
    /// Whether the reader asked to stop the in-flight turn. Stamped
    /// where `Command::Cancel` routes and spent when that turn's
    /// `Result` lands: a cancelled turn ends with the same failed
    /// `Result` a genuine error does, so `failed_turn_at` below cannot
    /// tell the two apart without it.
    pub pending_cancel: bool,
    /// When the slot's newest turn ended in failure, for the rail's
    /// mark. Cleared when a new turn is committed, dropped with the
    /// occupant, and never set for a turn the reader cancelled.
    pub failed_turn_at: Option<std::time::SystemTime>,
    /// The continuation the seat's newest failed turn left: what the model
    /// will be told, taken by the fire as soon as that failure is folded.
    /// Cleared by the fire, a new turn, a success, or the occupant changing.
    pub auto_continue: Option<PendingAutoContinue>,
    /// The continuation forge has sent for that failure, until the prompt
    /// runs: its id, and the occupant the failure belonged to. A `/new` or
    /// `/resume` the queue drains first leaves a different occupant under
    /// it, and the words describe a history the replacement does not carry.
    /// Not cleared by `Connected` - the replacement is the case it exists
    /// for.
    pub auto_continue_queued: Option<(String, Option<forge_primitives::SessionId>)>,
    /// Whether a continuation has already gone out for the seat's current
    /// failure episode. Held until a turn completes or a view shows the
    /// seat, so a failure that persists is continued once rather than in a
    /// prompt loop.
    pub auto_continue_spent: bool,
    /// Whether the CLI last reported live background work here - the
    /// `background_tasks_changed` snapshot, which carries the whole set
    /// each change, so an empty one clears. Held on the session rather
    /// than folded per view: two views always agree about it, and a
    /// connection failure clears it because the CLI sends no terminal
    /// snapshot for a session that died.
    pub background_work: bool,
    /// The CLI's background-task registry as it last reported it, each entry
    /// carrying the command its own card named where both have been seen.
    ///
    /// The bool above says that something is running; this is what a view
    /// draws a row from, and it is held for the same reason - the set arrives
    /// whole on each `background_tasks_changed`, and a view that attached
    /// afterwards would otherwise see the flag and not one row.
    pub background_tasks: Vec<crate::BackgroundTask>,
    /// The CLI's long-running calls it announced outside the registry, by
    /// task id: a foreground call that outlives a beat reaches `task_started`
    /// and its own `task_notification`, and never `background_tasks_changed`.
    ///
    /// Held apart from the registry because that snapshot replaces itself
    /// whole and would wipe them; merged where a view reads
    /// ([`Self::task_rows`]), so the row says what the seat is running
    /// whether the model backgrounded the call or not. Each keeps the CLI's
    /// own `is_backgrounded`, so a consumer can tell a foreground hold from a
    /// registry row; the turn's end settles the foreground ones even when no
    /// ending frame came, and a backgrounded hold lasts only until the next
    /// snapshot - that set is the background world, complete on every change.
    pub(crate) live_tasks: Vec<crate::BackgroundTask>,
    /// Prompts in the CLI's queue for this occupant: what a view draws as
    /// waiting, and what the pile read answers with.
    ///
    /// Appended where the prompt is sent - the only place that knows its
    /// source and words - and advanced by the CLI's own lifecycle frames,
    /// which are the only writer of a prompt's state. Belongs to the
    /// occupant, so it is dropped with the background registry.
    pub prompt_queue: Vec<crate::protocol::QueuedPrompt>,
    /// Whether this occupant's CLI advertises `msg_lifecycle_v1`, from its
    /// init frame. `None` until one has been read.
    ///
    /// The pile is drawn FOR it: without the frames nothing would ever settle
    /// a row, so a session that does not advertise them records no rows at all
    /// and falls back to drawing the prompt the way it was drawn before the
    /// pile existed. Unknown reads as present - the first prompt can precede
    /// the first init - and the pinned CLI always advertises it.
    pub lifecycle_frames: Option<bool>,
    /// The command a tool call's card carried, by tool-use id.
    ///
    /// Held rather than resolved on the spot because the card arrives BEFORE
    /// the `task_started` that links it to a task, so the command has to be
    /// kept until a link names it. Cleared with the registry: it belongs to
    /// the occupant that made the calls.
    background_commands: HashMap<String, String>,
    /// The ids whose cards arrived since the last registry change.
    ///
    /// What keeps the map above bounded: a command is kept while a live task
    /// links to it, or for one registry change after its card arrived - which
    /// is exactly the window in which the `task_started` naming it can land.
    staged_commands: HashSet<String>,
    /// The tool call that began a task, by task id, from `task_started`.
    task_tool_use: HashMap<String, String>,
    /// Whether this session is waiting on `/login` before it can run.
    ///
    /// Set from the signals the CLI actually sends: an assistant message
    /// whose error is `AuthenticationFailed`, or a retry frame naming the
    /// same class. A `TurnError` envelope whose text reads like an auth
    /// failure sets it too, as a fallback - that envelope's producers are
    /// forge's own failed write and cancel rather than CLI output. Cleared
    /// by a `Connected` and by a turn that finished, since either proves
    /// the credential works. A fact about the session, so the lifecycle
    /// derivation reads it rather than each view inventing the state from
    /// the event it saw.
    pub awaiting_login: bool,
    /// The `/dictate` overlay's per-session normalizer-axis overrides.
    /// Set by `Command::SetDictateOverride` / `::ResetDictateOverrides`
    /// and consumed when a capture finishes into
    /// `NormalizeOptions`. Session-scoped and volatile: dies with the
    /// session, never persisted.
    pub dictate_overrides: crate::dictate::DictateOverrides,
    /// The slash commands the CLI last advertised for this session, from
    /// the `system/init` frame's `slash_commands` and from
    /// `commands_changed`. Held here so a view that arrived after the
    /// turn started reads the list through the view surface, rather than
    /// waiting a whole turn for the next frame.
    pub available_commands: Vec<AvailableCommand>,
    /// The subagents the CLI last advertised, from the same init frame's
    /// `agents`, on the same terms as [`Self::available_commands`].
    pub available_agents: Vec<AvailableAgent>,
    /// Whether this turn's `system/init` has already been read for the
    /// agent catalogue. The first init of a turn carries it and a
    /// re-fire inside the same turn repeats it, so the read arms once per
    /// turn - the rule this fold applies.
    pub agents_emitted_this_turn: bool,
    /// Hook-observed permission mode: higher fidelity than the
    /// `system/status` mode, because
    /// the CLI can change mode without re-emitting status.
    pub observed_permission_mode: Option<PermissionMode>,
    /// Hook-observed effort level, on the same terms as
    /// [`Self::observed_permission_mode`].
    pub observed_effort: Option<EffortLevel>,
    /// The effort level forge stamped into this session's launch
    /// settings, read back so a session whose hook has not fired yet
    /// still answers with the level it is running at rather than none.
    /// `Max` when the launch asked for nothing, which is forge's default.
    pub configured_effort: EffortLevel,
    /// The permission mode forge stamped into the same launch settings,
    /// on the same terms: the answer until a hook reports one. `None`
    /// when the launch pinned none.
    pub configured_permission_mode: Option<PermissionMode>,
    /// The MCP servers this session last saw, from its own bridge: MCP
    /// is configured per session, so this is not an account-wide fact.
    pub mcp_servers: Option<McpServers>,
    /// How full the context window was at the last poll.
    pub context_usage: Option<ContextUsage>,
    /// The handoff this occupant is running, if any. Belongs to the
    /// occupant: a swap that completes resets it, and a request that was
    /// parked for a seat whose occupant changed describes a session that
    /// is gone.
    pub handoff: HandoffState,
    /// The model the session resolved to, from its connect and from
    /// every later init frame that names a different one.
    pub current_model: Option<CurrentModel>,
    /// The model catalogue the connect resolved against, kept so a frame
    /// naming another model can be resolved the same way the connect's own
    /// name was rather than falling back to the raw id.
    pub available_models: Vec<AvailableModel>,
    /// The monitors this session has running, and the ones that settled
    /// while it did, folded from the wire: the `Monitor` tool call, the
    /// task id the CLI assigns it, and the transition that settles it.
    /// The set drains once every entry in it is terminal, so a session
    /// that ran a monitor and stopped drawing one holds none.
    pub monitors: Vec<MonitorRecord>,
    /// The last OS-level walk of the session's process tree. The scan
    /// runs on a view's tick rather than here, so this is the most
    /// recent one whoever asked last produced.
    pub process_snapshot: Option<forge_agent::env::processes::ProcessSnapshot>,
    /// The last scan of the session's working tree, and when it was taken.
    /// Written by the seat's watch loop while a view is showing it, so a
    /// read answers a value rather than paying for a `git` subprocess.
    pub work_snapshot: Option<crate::work::WorkSnapshot>,
    /// The session's sub-agent instances, folded a frame at a time as they
    /// arrive. Seeded from the history a connect carries, then driven by
    /// each frame.
    pub card_tracker: crate::subagent_cards::CardTracker,
    /// The instance list the last announcement carried, which a push is
    /// gated on: an announcement is a frame about a change, so the frame
    /// that moved nothing announces nothing.
    pub cards_snapshot: Vec<forge_primitives::runtime::SubagentCard>,
    /// The seat's walked file index, and when it was taken. Written by the
    /// seat's own loop while a view is showing it, so the composer's `@`
    /// list reads a value rather than paying for a walk of the tree.
    pub file_index: Option<crate::work::HeldFileIndex>,
}

impl DomainSession {
    /// Whether a permission prompt is still waiting on `tool_id`.
    ///
    /// The session task asks the same question as it pops the slot, and this
    /// asks it where a refusal can still reach a caller: by the time the task
    /// sees the answer, the click has already happened.
    pub fn awaits_permission(&self, tool_id: &str) -> bool {
        matches!(
            self.pending_interactions.get(tool_id),
            Some(PendingInteractionSlot::Permission { .. }),
        )
    }

    /// The same for a question. A prompt answered already, or one that asked
    /// something else, is a dock that is gone.
    pub fn awaits_question(&self, tool_id: &str) -> bool {
        matches!(
            self.pending_interactions.get(tool_id),
            Some(PendingInteractionSlot::Question { .. }),
        )
    }

    /// Replace the registry with the CLI's last snapshot, which carries the
    /// whole set, and fill in whatever commands are already resolvable.
    ///
    /// **A backgrounded hold is redundant by definition**, because this
    /// snapshot is the whole background set; keeping one past a later snapshot
    /// that drops its task is exactly how a `TaskStop`'s row - which sends no
    /// ending and no registry change - resurfaces. So every backgrounded hold
    /// is dropped here. A foreground hold is dropped only when the snapshot
    /// names its task or call: then the registry owns the call.
    ///
    /// Also the place the held commands are pruned after a snapshot: a task
    /// that has left both sets takes its link with it, and the staging window
    /// for a card whose `task_started` has not landed closes.
    pub(crate) fn replace_background_tasks(&mut self, tasks: Vec<crate::BackgroundTask>) {
        self.background_tasks = tasks;
        self.link_background_calls();
        let registered_ids: HashSet<&str> =
            self.background_tasks.iter().map(|task| task.task_id.as_str()).collect();
        let registered_calls: HashSet<&str> =
            self.background_tasks.iter().filter_map(|task| task.tool_use_id.as_deref()).collect();
        self.live_tasks.retain(|task| {
            !task.is_backgrounded
                && !registered_ids.contains(task.task_id.as_str())
                && task.tool_use_id.as_deref().is_none_or(|call| !registered_calls.contains(call))
        });
        // Prune with the staging intact, then close the window: a card that
        // arrived since the last change keeps its command for this one, which
        // is exactly where its `task_started` can land.
        self.prune_task_maps();
        self.staged_commands.clear();
    }

    /// Record a task the CLI announced on `task_started`, which is the only
    /// signal a long foreground call leaves: the registry names backgrounded
    /// calls alone, and a foreground one is announced here and ended by its
    /// own `task_notification` or the turn it ran in.
    ///
    /// Idempotent by task id AND by call: the CLI can announce one call under
    /// two task ids, and a second row for it would double the seat's count. A
    /// backgrounded task its registry already names is not held at all: the
    /// snapshot is the whole background set, and the mirror would be the row
    /// that survives the snapshot dropping it.
    pub(crate) fn note_live_task(&mut self, task: crate::BackgroundTask) {
        let names_it = |held: &crate::BackgroundTask| {
            held.task_id == task.task_id
                || (task.tool_use_id.is_some() && held.tool_use_id == task.tool_use_id)
        };
        let known = self.live_tasks.iter().any(names_it)
            || (task.is_backgrounded && self.background_tasks.iter().any(names_it));
        if known {
            return;
        }
        self.live_tasks.push(task);
        self.link_background_calls();
    }

    /// Withdraw a task its own ending frame settled.
    ///
    /// Matched by the task id **or the call** the frame names: one call can be
    /// announced under two task ids, and the ending may name either.
    pub(crate) fn settle_live_task(&mut self, task_id: &str, tool_use_id: Option<&str>) {
        let settled = |task: &crate::BackgroundTask| {
            task.task_id == task_id
                || tool_use_id.is_some_and(|call| task.tool_use_id.as_deref() == Some(call))
        };
        let calls: Vec<String> = self
            .live_tasks
            .iter()
            .filter(|task| settled(task))
            .filter_map(|task| task.tool_use_id.clone())
            .collect();
        self.live_tasks.retain(|task| !settled(task));
        // The settled call's command is read by nothing now. Other cards stay
        // staged: their `task_started` may still be on its way.
        for call in calls {
            self.staged_commands.remove(&call);
        }
        self.prune_task_maps();
    }

    /// Withdraw what the end of a turn settles: its foreground holds.
    ///
    /// **A foreground call cannot outlive the turn that ran it.** Its own
    /// `task_notification` normally ends it, but the `TaskStop` tool path
    /// sends no ending frame, and a cancelled turn can bury the one it would
    /// have sent, so the turn's terminal frame settles the rest. A background
    /// hold stays: a backgrounded call outlives its turn, and the CLI's
    /// registry is its own source.
    ///
    /// The turn boundary also closes the staging window: every card the turn
    /// drew has been linked by its `task_started` or never named a task.
    pub(crate) fn settle_foreground_holds(&mut self) {
        self.live_tasks.retain(|task| task.is_backgrounded);
        self.staged_commands.clear();
        self.prune_task_maps();
    }

    /// The seat's live tasks as a view reads them: the CLI's background
    /// registry first, then the announced foreground calls still running.
    ///
    /// Deduped by task id and by call, so a hold the registry names never
    /// draws beside it before the next snapshot prunes it.
    pub fn task_rows(&self) -> Vec<crate::BackgroundTask> {
        let registered_ids: HashSet<&str> =
            self.background_tasks.iter().map(|task| task.task_id.as_str()).collect();
        let registered_calls: HashSet<&str> =
            self.background_tasks.iter().filter_map(|task| task.tool_use_id.as_deref()).collect();
        let mut rows = self.background_tasks.clone();
        rows.extend(
            self.live_tasks
                .iter()
                .filter(|task| {
                    !registered_ids.contains(task.task_id.as_str())
                        && task
                            .tool_use_id
                            .as_deref()
                            .is_none_or(|call| !registered_calls.contains(call))
                })
                .cloned(),
        );
        rows
    }

    /// Drop the link of every task that is no longer live, and the command of
    /// every call no live task names.
    ///
    /// A card whose command arrived since the last prune stays while it is
    /// staged, because its `task_started` may still be on its way.
    fn prune_task_maps(&mut self) {
        let running: HashSet<&str> = self
            .background_tasks
            .iter()
            .chain(self.live_tasks.iter())
            .map(|task| task.task_id.as_str())
            .collect();
        self.task_tool_use.retain(|task_id, _| running.contains(task_id.as_str()));
        let linked: HashSet<&str> = self.task_tool_use.values().map(String::as_str).collect();
        self.background_commands
            .retain(|id, _| linked.contains(id.as_str()) || self.staged_commands.contains(id));
    }

    /// Seed the announced holds a fixture cannot drive through a session
    /// actor: the view surface's read test has no wire to produce them.
    #[cfg(any(test, feature = "testing"))]
    pub fn seed_live_tasks(&mut self, tasks: Vec<crate::BackgroundTask>) {
        self.live_tasks = tasks;
        self.link_background_calls();
    }

    /// Record the command a tool call's card carried.
    pub(crate) fn hold_background_command(&mut self, tool_use_id: String, command: String) {
        self.staged_commands.insert(tool_use_id.clone());
        self.background_commands.insert(tool_use_id, command);
        self.link_background_calls();
    }

    /// Record which tool call began a task, which is the link that names the
    /// command, and fill in that task's command if the card is already held.
    pub(crate) fn hold_task_tool_use(&mut self, task_id: String, tool_use_id: String) {
        self.task_tool_use.insert(task_id, tool_use_id);
        self.link_background_calls();
    }

    /// Let go of the registry and everything that resolves a command in it: a
    /// new occupant's calls are its own, and the CLI re-sends the set only
    /// when it changes.
    ///
    /// The queued prompts are handed back rather than dropped: they died with
    /// the process that held them, so the caller owns the row each one still
    /// deserves.
    pub(crate) fn drop_background_tasks(&mut self) -> Vec<crate::protocol::QueuedPrompt> {
        self.background_work = false;
        self.background_tasks.clear();
        self.live_tasks.clear();
        self.background_commands.clear();
        self.staged_commands.clear();
        self.task_tool_use.clear();
        // The queue died with the CLI process the prompts were written to, and
        // so did the promise its init frame made: the flag is re-read from the
        // new occupant's own init, which the CLI re-fires every turn. Left
        // latched, a `/resume` onto an advertising CLI would keep the pile
        // silently off, or latch true beside rows nothing can settle.
        self.lifecycle_frames = None;
        std::mem::take(&mut self.prompt_queue)
    }

    /// Record a prompt as waiting, at the dispatch site.
    pub(crate) fn record_queued_prompt(
        &mut self,
        uuid: &str,
        source: crate::protocol::PromptSource,
        text: &str,
        attachments: Vec<forge_primitives::ImageAttachment>,
    ) {
        self.prompt_queue.push(crate::protocol::QueuedPrompt {
            uuid: uuid.to_owned(),
            source,
            text: text.to_owned(),
            attachments,
        });
    }

    /// The entry a lifecycle frame names, while it is still waiting.
    ///
    /// **Read before the frame advances the queue**, because `started` is a
    /// settled state: the entry leaves the pile as the row it draws arrives.
    pub(crate) fn queued_prompt(&self, uuid: &str) -> Option<&crate::protocol::QueuedPrompt> {
        self.prompt_queue.iter().find(|prompt| prompt.uuid == uuid)
    }

    /// Take every entry still waiting, leaving the pile empty.
    ///
    /// **The capability arrives after the first prompt.** `msg_lifecycle_v1`
    /// is read from the CLI's init frame, and its first user message lands
    /// before that - so a prompt recorded while the flag was `None` can never
    /// be settled by a lifecycle frame. The task takes the entries and draws
    /// them itself, the only carrier they will get.
    pub(crate) fn take_queued_prompts(&mut self) -> Vec<crate::protocol::QueuedPrompt> {
        std::mem::take(&mut self.prompt_queue)
    }

    /// Advance a prompt's state from one of the CLI's lifecycle frames.
    ///
    /// Returns whether the id was known. A state the CLI has SETTLED drops the
    /// row, so the pile holds only prompts still waiting; a state this build
    /// cannot name leaves it standing, because a word the CLI adds later must
    /// not empty a pile nobody can see being wrong. An unknown id is not an
    /// error - a prompt dispatched before this build, or a frame for a prompt
    /// another cohort minted, simply is not in the pile.
    pub(crate) fn advance_queued_prompt(&mut self, uuid: &str, state: &str) -> bool {
        let Some(at) = self.prompt_queue.iter().position(|p| p.uuid == uuid) else {
            return false;
        };
        if SETTLED_STATES.contains(&state) {
            self.prompt_queue.remove(at);
        }
        true
    }

    /// Drop an entry outright - a cancel that the CLI confirmed.
    pub(crate) fn drop_queued_prompt(&mut self, uuid: &str) {
        self.prompt_queue.retain(|p| p.uuid != uuid);
    }

    /// Fill in the call that began each entry, and the command its card
    /// carried, wherever both halves are known. Called after either half
    /// lands, so the order the CLI sends them in does not matter.
    fn link_background_calls(&mut self) {
        for task in self.background_tasks.iter_mut().chain(self.live_tasks.iter_mut()) {
            if task.tool_use_id.is_none() {
                task.tool_use_id = self.task_tool_use.get(&task.task_id).cloned();
            }
            if task.command.is_some() {
                continue;
            }
            task.command = self
                .task_tool_use
                .get(&task.task_id)
                .and_then(|tool_use| self.background_commands.get(tool_use))
                .cloned();
        }
    }

    /// Construct a fresh `DomainSession` bound to `key` with the
    /// given `conn`. Pre-spawn / pre-Connect callers pass `None` to
    /// register a placeholder domain whose handle slot fills in once
    /// the spawn handler runs.
    pub fn new(key: SessionSlot, conn: Option<Arc<AgentHandle>>) -> Self {
        Self {
            key,
            session_id: None,
            conn,
            pending_interactions: PendingInteractions::default(),
            prompt_queue: Vec::new(),
            lifecycle_frames: None,
            spawned_force_new: false,
            pending_lead_resume_note: None,
            spawn_wrote_row: false,
            runtime_state: None,
            turn_pending: false,
            turn_commits: 0,
            turns_served: 0,
            turn_open: false,
            pending_cancel: false,
            failed_turn_at: None,
            auto_continue: None,
            auto_continue_queued: None,
            auto_continue_spent: false,
            background_work: false,
            background_tasks: Vec::new(),
            live_tasks: Vec::new(),
            background_commands: HashMap::new(),
            staged_commands: HashSet::new(),
            task_tool_use: HashMap::new(),
            awaiting_login: false,
            dictate_overrides: crate::dictate::DictateOverrides::default(),
            available_commands: Vec::new(),
            available_agents: Vec::new(),
            agents_emitted_this_turn: false,
            observed_permission_mode: None,
            observed_effort: None,
            configured_effort: EffortLevel::Max,
            configured_permission_mode: None,
            mcp_servers: None,
            context_usage: None,
            handoff: HandoffState::default(),
            current_model: None,
            available_models: Vec::new(),
            monitors: Vec::new(),
            process_snapshot: None,
            work_snapshot: None,
            card_tracker: crate::subagent_cards::CardTracker::default(),
            cards_snapshot: Vec::new(),
            file_index: None,
        }
    }

    /// Whether a turn is in flight. `turn_pending` is the primary
    /// signal: it is stamped synchronously when a `Command::Prompt` is
    /// routed, so it covers the window before any wire echo lands.
    /// `runtime_state` is OR'd in rather than trusted on its own -
    /// `session_state_changed` appears in no wire-conformance baseline
    /// and in none of the captured session JSONL, so it may never
    /// arrive.
    ///
    /// Shared by the prompt-interception guards and the worker
    /// activity derivation so the two cannot drift apart.
    pub fn turn_in_flight(&self) -> bool {
        self.turn_pending
            || self.turn_open
            || matches!(
                self.runtime_state,
                Some(RuntimeSessionState::Running | RuntimeSessionState::RequiresAction)
            )
    }

    /// A prompt was committed as a turn.
    ///
    /// **A prompt queued into a turn already running joins that turn.** The
    /// CLI runs it inside the same turn and emits one Result for both, so
    /// counting it as another turn would leave the outstanding count
    /// permanently ahead and close the handoff's idle gate for the life of
    /// the seat. A prompt sent while the task still reads a turn in flight
    /// but the CLI has started a new one is held busy by `prompt_queue`
    /// until its own Result serves the boundary.
    pub fn commit_turn(&mut self) -> bool {
        if self.turn_in_flight() {
            return false;
        }
        self.turn_commits = self.turn_commits.saturating_add(1);
        self.turn_pending = true;
        true
    }

    /// Undo a commit whose prompt never reached its task: the send failed, so
    /// no Result will ever serve it. Without this the seat reads a turn in
    /// flight for the life of the session, and the header's spinner with it.
    pub fn uncommit_turn(&mut self) {
        self.turn_commits = self.turn_commits.saturating_sub(1);
        self.turn_pending = self.turn_commits > self.turns_served;
    }

    /// One Result (or transport error) closed the oldest outstanding turn.
    ///
    /// **One, not all of them.** A prompt committed before this Result
    /// landed belongs to the next turn, so the flag is derived from what is
    /// still outstanding rather than cleared outright.
    pub fn serve_turn(&mut self) {
        self.turns_served = self.turns_served.saturating_add(1);
        self.turn_pending = self.turn_commits > self.turns_served;
    }

    /// Whether a committed turn is still waiting for its Result.
    pub fn turns_outstanding(&self) -> bool {
        self.turn_commits > self.turns_served
    }

    /// Drop the turn accounting with the occupant it described.
    pub fn reset_turns(&mut self) {
        self.turn_commits = 0;
        self.turns_served = 0;
        self.turn_pending = false;
    }

    /// Whether a handoff may start here: nothing running, nothing waiting in
    /// the CLI's queue, and nothing parked on an answer.
    ///
    /// **All three, not just the turn.** A prompt already queued is a turn
    /// the handoff's own prompt would be folded into, and a parked
    /// interaction is a turn that has not ended; the document names both as
    /// the "otherwise idle" condition.
    pub fn is_idle_for_handoff(&self) -> bool {
        !self.turn_in_flight()
            && self.prompt_queue.is_empty()
            && self.pending_interactions.is_empty()
    }

    /// Fold one frame's own claim about the turn into the liveness mirror.
    ///
    /// The model's output is the proof a turn is running and its Result is the
    /// proof it ended - which is what a view derives from the same
    /// frames. Nothing else is read: a `system` report, a task fact or a hook
    /// line says nothing about whether a turn is open, so it leaves the mirror
    /// where it stands.
    pub fn note_liveness(&mut self, said: TurnLiveness) {
        self.turn_open = said == TurnLiveness::Working;
    }
}

/// The effort level a spawn asked for, read back from the launch
/// settings forge stamps for the CLI.
///
/// The CLI reads `effortLevel` out of its own settings file, so a spawn
/// that pins nothing runs at whatever that file says; forge's default
/// for an unset one is `Max`, the same default the config reader
/// applies.
pub(crate) fn configured_effort_from_settings(
    settings: &crate::SessionLaunchSettings,
) -> EffortLevel {
    settings
        .settings
        .as_ref()
        .and_then(|document| document.get(EFFORT_LEVEL_KEY))
        .and_then(serde_json::Value::as_str)
        .and_then(EffortLevel::from_stored)
        .unwrap_or(EffortLevel::Max)
}

/// The mode a spawn asked for, read back from the same launch settings
/// the effort comes from.
///
/// `None` when the launch pinned nothing and when it pinned a mode forge
/// cannot name: unlike effort there is no default to fall back on here,
/// because the CLI's own default is not a forge setting.
pub(crate) fn configured_permission_mode_from_settings(
    settings: &crate::SessionLaunchSettings,
) -> Option<PermissionMode> {
    settings
        .settings
        .as_ref()
        .and_then(|document| document.get(crate::SessionLaunchSettings::PERMISSIONS_KEY))
        .and_then(|permissions| {
            permissions.get(crate::SessionLaunchSettings::PERMISSIONS_DEFAULT_MODE_KEY)
        })
        .and_then(serde_json::Value::as_str)
        .and_then(PermissionMode::from_wire)
}

/// The CLI settings key a session's effort is stamped under. One const,
/// so the writer and this reader cannot split the seam.
pub(crate) const EFFORT_LEVEL_KEY: &str = "effortLevel";

impl std::fmt::Debug for DomainSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DomainSession")
            .field("key", &self.key)
            .field("session_id", &self.session_id)
            .field("pending_interactions_count", &self.pending_interactions.len())
            .finish_non_exhaustive()
    }
}

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

    /// The level the spawn asked the CLI to run at, so a session whose
    /// hook has not fired yet still names an effort rather than none.
    #[test]
    fn a_launch_that_pins_an_effort_names_that_level() {
        let settings = crate::SessionLaunchSettings {
            settings: Some(serde_json::json!({ "effortLevel": "high" })),
            ..Default::default()
        };

        assert_eq!(
            configured_effort_from_settings(&settings),
            EffortLevel::High,
            "the level the launch stamped is the level the session runs at",
        );
    }

    /// A launch that pins nothing leaves the CLI on its own settings
    /// file, and forge's default for an unset level is `Max`.
    #[test]
    fn a_launch_that_pins_no_effort_reads_forges_default() {
        let settings = crate::SessionLaunchSettings::default();

        assert_eq!(
            configured_effort_from_settings(&settings),
            EffortLevel::Max,
            "an unset effort is forge's default rather than nothing",
        );
    }

    /// A level forge cannot name is a level it cannot report, and the
    /// default is a better answer than a guess.
    #[test]
    fn an_unreadable_launch_effort_reads_forges_default() {
        let settings = crate::SessionLaunchSettings {
            settings: Some(serde_json::json!({ "effortLevel": "turbo" })),
            ..Default::default()
        };

        assert_eq!(
            configured_effort_from_settings(&settings),
            EffortLevel::Max,
            "a level forge cannot read falls back rather than failing",
        );
    }

    /// The mode a spawn asked for, read back from the same launch
    /// settings. Forge stamps its effective default here, so a session
    /// whose hook has not fired yet still names the mode it was launched
    /// in rather than nothing.
    #[test]
    fn a_launch_that_pins_a_mode_names_that_mode() {
        let settings = crate::SessionLaunchSettings {
            settings: Some(serde_json::json!({ "permissions": { "defaultMode": "plan" } })),
            ..Default::default()
        };

        assert_eq!(
            configured_permission_mode_from_settings(&settings),
            Some(PermissionMode::Plan),
            "the mode the launch stamped is the mode the session runs in",
        );
    }

    /// A launch that pinned nothing says so rather than naming a mode
    /// forge did not choose.
    #[test]
    fn a_launch_that_pins_no_mode_names_none() {
        let settings = crate::SessionLaunchSettings::default();

        assert_eq!(
            configured_permission_mode_from_settings(&settings),
            None,
            "no pin is no answer, rather than a mode forged from nothing",
        );
    }

    /// **A prompt the CLI queues inside a running turn shares that turn's
    /// one Result.** Counting it as another turn leaves the outstanding
    /// count permanently ahead, which closes the handoff's idle gate for the
    /// life of the seat.
    #[test]
    fn a_prompt_queued_into_a_running_turn_joins_its_single_result() {
        let mut domain = DomainSession::new(SessionSlot::lead("Org", "proj"), None);
        domain.commit_turn();
        domain.commit_turn();

        assert_eq!(domain.turn_commits, 1, "the queued prompt joins the running turn");
        domain.serve_turn();
        assert!(!domain.turns_outstanding(), "the one Result serves both prompts");
        assert!(!domain.turn_pending, "and the seat reads idle at the boundary");
    }

    /// A mode forge cannot name is a mode it cannot report.
    #[test]
    fn an_unreadable_launch_mode_names_none() {
        let settings = crate::SessionLaunchSettings {
            settings: Some(serde_json::json!({ "permissions": { "defaultMode": "yolo" } })),
            ..Default::default()
        };

        assert_eq!(
            configured_permission_mode_from_settings(&settings),
            None,
            "a mode forge cannot read is not guessed at",
        );
    }
}