kraal 0.2.0

//! The session server: owns every pane, workspace and attached client.
//!
//! All state lives on one thread. PTY readers, process waiters and client sockets run on helper
//! threads that only forward [`Event`]s through a channel, so the state needs no locking.

mod agents;
mod attention;
mod clients;
mod commands;
mod contexts;
mod definitions;
mod edits;
pub mod events;
mod fileman;
mod geometry;
mod graphics;
mod hints;
mod input;
pub mod net;
mod overlays;
mod panes;
mod persistence;
mod reload;
pub mod remote;
mod render;
mod requests;
pub mod runbooks;
mod ssh;
mod tools;
mod ui;
mod worktrees;

use std::collections::HashMap;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;
use std::time::{Duration, Instant};

use anyhow::{Context, Result};
use crossbeam_channel::{Receiver, Sender, TrySendError};
use ratatui::layout::Rect;

use crate::agent::detect_agent;
use crate::ansi::Encoder;
use crate::config::Config;
use crate::input::InputParser;
use crate::keymap::Keymap;
use crate::layout::{Node, PaneId};
use crate::pane::{Pane, PaneEnv, PaneSignal};
use crate::paths;
use crate::persist;
use crate::protocol::ServerMsg;

pub use agents::usage_text;
pub use clients::frame_bytes;
use persistence::spawn_persister;
pub use ssh::ssh_config_hosts;
pub use ui::Mode;

pub type ClientId = u32;

/// Everything the server thread reacts to.
pub enum Event {
    /// A pane has output waiting (see [`Server::pump_output`]).
    PaneReady(PaneId),
    /// Pane id, pid of the process that exited, exit code.
    PaneExited(PaneId, Option<u32>, Option<i32>),
    Connection(UnixStream),
    ClientFrame(ClientId, u8, Vec<u8>),
    ClientGone(ClientId),
    /// A background job (file listing) finished; poll and redraw.
    Wake,
    /// `git status` of a directory, from the git thread.
    Git(PathBuf, Option<Box<crate::git::GitInfo>>),
    /// Something a helper thread wants the user to know.
    Warning(String),
    /// Publish an event once what it describes is on disk (a run's log).
    Emit(&'static str, serde_json::Value),
    /// A job started with [`Server::background`] finished with this result.
    Job(u32, Box<dyn std::any::Any + Send>),
    /// The agents a server's session has now, from the connection with this id.
    RemoteAgents(u32, Vec<remote::RemoteAgent>),
    Shutdown,
}

/// A floating pane's position as fractions of the main area, so it follows resizes.
#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct FloatRect {
    pub x: f32,
    pub y: f32,
    pub w: f32,
    pub h: f32,
}

impl FloatRect {
    pub fn to_rect(self, area: Rect) -> Rect {
        let w = ((self.w * area.width as f32).round() as u16).clamp(10.min(area.width), area.width);
        let h = ((self.h * area.height as f32).round() as u16).clamp(5.min(area.height), area.height);
        let x = area.x + ((self.x * area.width as f32).round() as u16).min(area.width - w);
        let y = area.y + ((self.y * area.height as f32).round() as u16).min(area.height - h);
        Rect::new(x, y, w, h)
    }

    pub fn from_rect(r: Rect, area: Rect) -> FloatRect {
        let fw = area.width.max(1) as f32;
        let fh = area.height.max(1) as f32;
        FloatRect {
            x: (r.x - area.x) as f32 / fw,
            y: (r.y - area.y) as f32 / fh,
            w: r.width as f32 / fw,
            h: r.height as f32 / fh,
        }
    }
}

pub struct Floating {
    pub pane: PaneId,
    pub rect: FloatRect,
}

pub struct Tab {
    pub id: u32,
    pub name: String,
    pub root: Option<Node>,
    pub floating: Vec<Floating>,
    pub focus: Option<PaneId>,
    pub show_floating: bool,
    pub zoom: Option<PaneId>,
    /// Most recently focused first.
    pub recent: Vec<PaneId>,
}

impl Tab {
    pub fn new(id: u32, name: String) -> Tab {
        Tab {
            id,
            name,
            root: None,
            floating: Vec::new(),
            focus: None,
            show_floating: false,
            zoom: None,
            recent: Vec::new(),
        }
    }

    pub fn panes(&self) -> Vec<PaneId> {
        let mut v = self.root.as_ref().map(|r| r.leaves()).unwrap_or_default();
        v.extend(self.floating.iter().map(|f| f.pane));
        v
    }

    pub fn contains(&self, id: PaneId) -> bool {
        self.root.as_ref().is_some_and(|r| r.contains(id)) || self.floating.iter().any(|f| f.pane == id)
    }

    pub fn is_floating(&self, id: PaneId) -> bool {
        self.floating.iter().any(|f| f.pane == id)
    }

    pub fn set_focus(&mut self, id: PaneId) {
        self.focus = Some(id);
        self.recent.retain(|p| *p != id);
        self.recent.insert(0, id);
        if self.is_floating(id) {
            self.show_floating = true;
            // Raise it.
            if let Some(i) = self.floating.iter().position(|f| f.pane == id) {
                let f = self.floating.remove(i);
                self.floating.push(f);
            }
        } else if self.zoom.is_some_and(|z| z != id) {
            self.zoom = Some(id);
        }
    }

    /// Remove a pane; returns false if the tab is now empty.
    fn remove(&mut self, id: PaneId) -> bool {
        self.floating.retain(|f| f.pane != id);
        if let Some(root) = &mut self.root
            && root.contains(id)
            && !root.remove(id)
        {
            self.root = None;
        }
        if self.zoom == Some(id) {
            self.zoom = None;
        }
        self.recent.retain(|p| *p != id);
        if self.focus == Some(id) {
            let visible = |p: &PaneId| self.root.as_ref().is_some_and(|r| r.contains(*p)) || self.show_floating;
            self.focus = self.recent.iter().copied().find(visible).or_else(|| self.panes().first().copied());
            if let Some(f) = self.focus
                && !self.is_floating(f)
            {
                self.show_floating = self.show_floating && !self.floating.is_empty();
            }
        }
        if self.floating.is_empty() {
            self.show_floating = false;
        }
        !self.panes().is_empty()
    }
}

/// Where typed keys go besides the focused pane.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum SyncScope {
    #[default]
    Off,
    /// Every terminal pane of the current tab.
    Tab,
    /// Every terminal pane of the workspace, all tabs.
    Workspace,
    /// The panes the user marked.
    Marked,
}

pub struct Workspace {
    pub id: u32,
    pub name: String,
    pub tabs: Vec<Tab>,
    pub active: usize,
    pub definition: Option<PathBuf>,
    pub cwd: Option<PathBuf>,
    pub sync: SyncScope,
    /// Panes marked for typing in sync.
    pub marked: std::collections::HashSet<PaneId>,
    /// Production panes the user agreed to type into in sync, until sync stops.
    /// The approval holds for the production reason it was given for: a pane that moves to
    /// another production server is asked about again.
    pub sync_approved: HashMap<PaneId, contexts::Production>,
}

impl Workspace {
    /// Panes that receive what is typed in `focused`, besides itself.
    pub fn sync_targets(&self, focused: PaneId) -> Vec<PaneId> {
        let mut out = match self.sync {
            SyncScope::Off => Vec::new(),
            SyncScope::Tab => self.tab().root.as_ref().map(|r| r.leaves()).unwrap_or_default(),
            SyncScope::Workspace => {
                self.tabs.iter().flat_map(|t| t.root.as_ref().map(|r| r.leaves()).unwrap_or_default()).collect()
            }
            SyncScope::Marked => self.panes().into_iter().filter(|p| self.marked.contains(p)).collect(),
        };
        if !out.contains(&focused) {
            out.insert(0, focused);
        }
        out
    }

    pub fn tab(&self) -> &Tab {
        &self.tabs[self.active.min(self.tabs.len() - 1)]
    }

    pub fn tab_mut(&mut self) -> &mut Tab {
        let i = self.active.min(self.tabs.len() - 1);
        &mut self.tabs[i]
    }

    pub fn panes(&self) -> Vec<PaneId> {
        self.tabs.iter().flat_map(|t| t.panes()).collect()
    }
}

pub struct Client {
    pub id: ClientId,
    pub out: Sender<Vec<u8>>,
    pub attached: bool,
    pub cols: u16,
    pub rows: u16,
    pub workspace: u32,
    pub parser: InputParser,
    pub pending_since: Option<Instant>,
    pub mode: Mode,
    pub encoder: Encoder,
    /// The sidebar shows when it has something to say; once toggled, as the client chose.
    pub sidebar: Option<bool>,
    pub dirty: bool,
    pub last_input: Instant,
    pub terminal_focused: bool,
    pub drag: Option<ui::Drag>,
    pub toast: Option<(String, Instant)>,
    /// Click bookkeeping for double/triple click selection.
    pub last_click: Option<(Instant, u16, u16, u8)>,
    /// The first session's welcome note is showing.
    pub welcome: bool,
    /// The client runs on another machine: `localhost` there is not the server's.
    pub remote: bool,
    /// Its terminal reports every pointer motion (mode 1003), for a pane that asked for it.
    pub pointer_motion: bool,
    /// Its terminal can send every key with its place and its text (kitty keyboard protocol):
    /// unknown until it answers.
    pub keys_full: Option<bool>,
    /// Its terminal sends every key with its place, for the leader menu.
    pub keys_by_place: bool,
    /// A paste arriving in pieces.
    pub(in crate::server) paste: Option<input::PasteFlow>,
}

/// A line of the messages log.
pub struct Message {
    pub at: std::time::SystemTime,
    pub text: String,
    pub warning: bool,
}

/// Messages kept for the log.
const MAX_MESSAGES: usize = 200;

/// A control request waiting for a pane condition.
struct PendingWait {
    client: ClientId,
    out: Sender<Vec<u8>>,
    pane: PaneId,
    until: Vec<String>,
    deadline: Option<Instant>,
}

pub struct Server {
    pub session: String,
    pub config: Config,
    pub keymap: Keymap,
    pub env: PaneEnv,
    pub tx: Sender<Event>,
    pub panes: HashMap<PaneId, Pane>,
    pub workspaces: Vec<Workspace>,
    pub clients: HashMap<ClientId, Client>,
    next_id: u32,
    /// The client whose terminal size drives pane sizes.
    pub primary: Option<ClientId>,
    pub layout_area: Rect,
    waits: Vec<PendingWait>,
    persist_dirty: bool,
    last_persist: Instant,
    last_tick: Instant,
    next_render: Option<Instant>,
    quitting: bool,
    /// `kraal kill` requests, answered once the last state is written (or not).
    kill_waiters: Vec<ClientId>,
    /// Colours of the outer terminal, learnt from OSC 10/11 replies.
    pub outer_colors: Option<(String, String)>,
    /// What each client's terminal can show of the panes' images, and has been sent.
    images: HashMap<ClientId, graphics::ClientImages>,
    /// Warnings not shown yet: they go to the clients and the messages log.
    pub warnings: Vec<String>,
    /// Why `accounts.toml` could not be read last time, so that it is said once, not every scan.
    pub accounts_error: Option<String>,
    /// What kraal told the clients, for the messages log.
    pub messages: std::collections::VecDeque<Message>,
    /// Names of the workspace definitions on disk, refreshed periodically.
    pub def_names: Vec<String>,
    last_def_scan: Instant,
    /// Servers from the inventory, workspaces and ~/.ssh/config.
    pub known_servers: Vec<crate::servers::Known>,
    /// ssh destinations already looked at for auto-adding.
    discovered: std::collections::HashSet<String>,
    pub accounts: crate::accounts::Accounts,
    /// Name of the workspace whose definition is being started (for account rules).
    starting_workspace: Option<String>,
    pub usage: crate::accounts::UsageMap,
    last_usage_load: Instant,
    last_codex_poll: Instant,
    limit_alerts: HashMap<PaneId, agents::LimitAlert>,
    /// The limit message each agent pane shows, as already counted: a message staying on
    /// screen is one observation, not a new one at every check.
    screen_limits: HashMap<PaneId, (bool, f64)>,
    /// Agent panes switching accounts, until their agent stops: from, to, and the conversation.
    switches: HashMap<PaneId, (crate::accounts::Account, crate::accounts::Account, Option<String>)>,
    /// Panes whose account is over its threshold, already reported as an event.
    limit_events: std::collections::HashSet<PaneId>,
    /// Branch and changes per directory (None: not a repository).
    pub git: HashMap<PathBuf, Option<crate::git::GitInfo>>,
    git_pending: std::collections::HashSet<PathBuf>,
    git_requests: Sender<PathBuf>,
    last_git_scan: Instant,
    /// Current Kubernetes context per `KUBECONFIG` value in use.
    pub kube: HashMap<String, crate::context::KubeFile>,
    kube_cache: crate::context::KubeCache,
    /// Terraform workspaces of the directories panes are in, and of those Terraform runs worked
    /// on (`-chdir`, `TF_DATA_DIR`).
    tf_dirs: contexts::TfDirs,
    /// The `[hints.patterns]` of the config, compiled.
    pub hint_patterns: crate::hints::Custom,
    /// `kraal events` clients.
    subscribers: Vec<events::Subscriber>,
    events: events::EventLog,
    /// Panes `pane.opened` was said for.
    announced: std::collections::HashSet<PaneId>,
    hooks: events::Hooks,
    /// SSH server and production reason of each pane at the last tick, to notice changes.
    pane_states: events::PaneStates,
    /// Runbooks being prepared, by client.
    drafts: HashMap<ClientId, runbooks::Draft>,
    /// Plans shown by `kraal runbook run`, waiting for the yes that runs them.
    plans: HashMap<String, (Instant, runbooks::Draft)>,
    pub runs: Vec<runbooks::Run>,
    /// Runs written in the ledger whose targets wait for that to be done to start.
    launching: Vec<crate::runbooks::LedgerRun>,
    /// Process groups of timed-out targets whose panes were closed before they had ended.
    stray_groups: Vec<runbooks::StrayGroup>,
    /// What the panes of runs print, streamed to their logs.
    pub run_output: HashMap<PaneId, crate::runbooks::OutputLog>,
    /// Markdown runbooks in progress, by client.
    md_sessions: HashMap<ClientId, runbooks::MdSession>,
    /// What to do with the result of each background job.
    jobs: HashMap<u32, JobDone>,
    /// Files of servers being edited in the local editor.
    edits: Vec<edits::Edit>,
    /// Links to open on a client once their tunnel is up: (client, tunnel id, URL).
    pending_opens: Vec<(ClientId, u32, String)>,
    /// Preview panes being read again.
    preview_loads: std::collections::HashSet<PaneId>,
    /// The thread that writes the session state.
    persister: Sender<persistence::Persist>,
    /// What is saved of each pane's scrollback.
    pub scrollback: HashMap<PaneId, persist::SavedScrollback>,
    /// Starts the names of the scrollback files this server makes, so that none is a file an
    /// earlier server's state names.
    scrollback_tag: String,
    /// Writes run logs and the run ledger, in order, off the event loop.
    run_logs: Sender<crate::runbooks::LogJob>,
    pub tunnels: Vec<net::Tunnel>,
    /// Servers whose kraal session's agents are listed.
    pub remotes: Vec<remote::Watch>,
    /// Ports mentioned in pane output, and those of them something listens on.
    port_candidates: HashMap<u16, PaneId>,
    pub ports: std::collections::BTreeMap<u16, PaneId>,
    /// Panes with output waiting, in round-robin order.
    ready: std::collections::VecDeque<PaneId>,
    ready_set: std::collections::HashSet<PaneId>,
    /// Panes whose process exited, with its pid and code: their last output is read first, for
    /// a bounded time (something left in the background may keep writing).
    exiting: HashMap<PaneId, (Option<u32>, Option<i32>, Instant)>,
    /// Input going to panes whose context is being read again, kept until the read ends, in the
    /// order it was typed, with where it goes (see `Server::hold_input`).
    held_input: Vec<input::Held>,
    /// Panes told they have the focus (mode 1004).
    focus_reported: std::collections::HashSet<PaneId>,
    /// Whether the sidebar had something to show at the last tick.
    sidebar_was_useful: bool,
    /// Whether a client was told how to stop the broadcast its key started.
    pub broadcast_tip_shown: bool,
}

/// Finishes a background job on the server thread with its result.
type JobDone = Box<dyn FnOnce(&mut Server, Box<dyn std::any::Any + Send>)>;

const TICK: Duration = Duration::from_millis(400);
const FRAME: Duration = Duration::from_millis(16);
const INPUT_FLUSH: Duration = Duration::from_millis(30);

/// Run a session server in the foreground until it is told to exit.
pub fn run(session: &str) -> Result<()> {
    let socket = paths::socket_path(session);
    paths::private_runtime_dir().context("checking the runtime directory")?;
    if socket.exists() {
        if UnixStream::connect(&socket).is_ok() {
            anyhow::bail!("session {session:?} is already running");
        }
        let _ = std::fs::remove_file(&socket);
    }
    let listener = UnixListener::bind(&socket).with_context(|| format!("binding {}", socket.display()))?;

    let (config, mut warnings) = match Config::load() {
        Ok(c) => (c, Vec::new()),
        Err(e) => (Config::default(), vec![format!("config: {e:#}")]),
    };
    let (mut keymap, kw) = Keymap::build(&config.keys);
    warnings.extend(kw);
    warnings.extend(keymap.bind_tools(&config.tools));
    warnings.extend(crate::tools::check(&config.tools));
    let (hint_patterns, hw) = crate::hints::Custom::build(&config.hints, &config.tools);
    warnings.extend(hw);
    let (hook_list, hkw) = events::read_hooks(&config.hooks);
    warnings.extend(hkw);
    let (theme, tw) = crate::theme::Theme::from_config(&config.theme);
    crate::theme::set(theme);
    warnings.extend(tw);

    let (tx, rx) = crossbeam_channel::bounded::<Event>(4096);
    let persist_tx = tx.clone();
    let logs_tx = tx.clone();
    {
        let tx = tx.clone();
        std::thread::Builder::new().name("accept".into()).spawn(move || {
            for stream in listener.incoming() {
                match stream {
                    Ok(s) => {
                        if tx.send(Event::Connection(s)).is_err() {
                            break;
                        }
                    }
                    Err(_) => continue,
                }
            }
        })?;
    }
    {
        let tx = tx.clone();
        let mut signals = signal_hook::iterator::Signals::new([
            signal_hook::consts::SIGTERM,
            signal_hook::consts::SIGINT,
            signal_hook::consts::SIGHUP,
        ])?;
        std::thread::Builder::new().name("signals".into()).spawn(move || {
            if signals.forever().next().is_some() {
                let _ = tx.send(Event::Shutdown);
            }
        })?;
    }

    let git_tx = tx.clone();
    let env = PaneEnv {
        socket: socket.clone(),
        session: session.to_string(),
        shell: config.shell(),
        scrollback: config.scrollback,
        graphics: std::sync::Arc::new(crate::graphics::Shared::new(
            config.graphics.enabled,
            config.graphics.memory_mb.saturating_mul(1 << 20),
        )),
    };
    let mut server = Server {
        session: session.to_string(),
        config,
        keymap,
        env,
        tx,
        panes: HashMap::new(),
        workspaces: Vec::new(),
        clients: HashMap::new(),
        next_id: 1,
        primary: None,
        layout_area: Rect::new(0, 1, 120, 39),
        waits: Vec::new(),
        persist_dirty: false,
        last_persist: Instant::now(),
        last_tick: Instant::now(),
        next_render: None,
        quitting: false,
        kill_waiters: Vec::new(),
        outer_colors: None,
        images: HashMap::new(),
        warnings,
        accounts_error: None,
        messages: Default::default(),
        def_names: Vec::new(),
        last_def_scan: Instant::now(),
        known_servers: Vec::new(),
        discovered: Default::default(),
        accounts: crate::accounts::Accounts::empty(),
        starting_workspace: None,
        usage: Default::default(),
        last_usage_load: Instant::now(),
        last_codex_poll: Instant::now() - Duration::from_secs(3600),
        limit_alerts: HashMap::new(),
        screen_limits: HashMap::new(),
        switches: HashMap::new(),
        limit_events: Default::default(),
        ready: Default::default(),
        ready_set: Default::default(),
        exiting: Default::default(),
        focus_reported: Default::default(),
        held_input: Vec::new(),
        sidebar_was_useful: false,
        broadcast_tip_shown: false,
        tunnels: Vec::new(),
        remotes: Vec::new(),
        port_candidates: HashMap::new(),
        ports: Default::default(),
        git: HashMap::new(),
        git_pending: Default::default(),
        git_requests: worktrees::spawn_git_worker(git_tx),
        last_git_scan: Instant::now() - Duration::from_secs(10),
        kube: HashMap::new(),
        kube_cache: Default::default(),
        tf_dirs: Default::default(),
        hint_patterns,
        subscribers: Vec::new(),
        events: Default::default(),
        announced: Default::default(),
        hooks: events::Hooks::new(hook_list),
        pane_states: Default::default(),
        drafts: HashMap::new(),
        plans: HashMap::new(),
        runs: Vec::new(),
        launching: Vec::new(),
        stray_groups: Vec::new(),
        run_output: HashMap::new(),
        md_sessions: HashMap::new(),
        jobs: HashMap::new(),
        edits: Vec::new(),
        preview_loads: Default::default(),
        pending_opens: Vec::new(),
        persister: spawn_persister(persist_tx),
        scrollback: HashMap::new(),
        scrollback_tag: format!("{}-{}", crate::accounts::now(), std::process::id()),
        run_logs: {
            let warn = logs_tx;
            crate::runbooks::spawn_log_writer(move |w| {
                let _ = warn.send(Event::Warning(w));
            })
        },
    };
    server.scan_definitions();
    server.restore();
    let result = server.event_loop(&rx);
    server.shutdown();
    let _ = std::fs::remove_file(&socket);
    result
}

impl Server {
    /// Run `work` on a helper thread (slow git, commands), then `then` with its result on the
    /// server thread, so nothing slow ever blocks the event loop. `false` when no thread could
    /// be started: `then` never runs.
    pub fn background<T: Send + 'static>(
        &mut self,
        work: impl FnOnce() -> T + Send + 'static,
        then: impl FnOnce(&mut Server, T) + 'static,
    ) -> bool {
        let id = self.alloc_id();
        self.jobs.insert(
            id,
            Box::new(move |server, result| {
                if let Ok(value) = result.downcast::<T>() {
                    then(server, *value);
                }
            }),
        );
        let tx = self.tx.clone();
        let spawned = std::thread::Builder::new().name("job".into()).spawn(move || {
            let _ = tx.send(Event::Job(id, Box::new(work())));
        });
        if let Err(e) = spawned {
            self.jobs.remove(&id);
            self.warnings.push(format!("cannot start a background job: {e}"));
            return false;
        }
        true
    }

    /// Show `mode` (a question a background job prepared) unless the client is busy with
    /// another picker or prompt: then the question is dropped (the client is told to ask again).
    pub fn set_mode_if_idle(&mut self, client: ClientId, mode: Mode) {
        let Some(c) = self.clients.get_mut(&client) else { return };
        if !matches!(c.mode, Mode::Normal | Mode::Leader { .. }) {
            c.toast =
                Some(("finish what you are doing, then ask again".into(), Instant::now() + Duration::from_secs(3)));
            return;
        }
        c.mode = mode;
        c.dirty = true;
        self.mark_dirty_client(client);
    }

    pub fn alloc_id(&mut self) -> u32 {
        let id = self.next_id;
        self.next_id += 1;
        id
    }

    fn event_loop(&mut self, rx: &Receiver<Event>) -> Result<()> {
        loop {
            let timeout = if self.ready.is_empty() {
                self.next_deadline().saturating_duration_since(Instant::now())
            } else {
                Duration::ZERO
            };
            match rx.recv_timeout(timeout) {
                Ok(ev) => {
                    self.handle(ev);
                    self.follow_focus();
                }
                Err(crossbeam_channel::RecvTimeoutError::Timeout) => {}
                Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break,
            }
            // Keystrokes, clients and control requests come first. Panes are told of focus changes
            // after each (a pane focused and left within one batch is told both).
            while let Ok(ev) = rx.try_recv() {
                self.handle(ev);
                self.follow_focus();
            }
            // …then a slice of program output, interrupted as soon as something else arrives.
            self.pump_output(Duration::from_millis(6), rx);
            self.timers();
            // Timers close panes too.
            self.follow_focus();
            if self.quitting {
                break;
            }
            if self.next_render.is_some_and(|t| t <= Instant::now()) {
                self.render_clients();
            }
        }
        Ok(())
    }

    /// Parse waiting pane output for up to `budget`: the panes clients are looking at first,
    /// for at most half of it, then the others in turn, one chunk each. Stops early when a
    /// control event arrives, but the others always get a chunk: a pane printing without a
    /// pause keeps the queue busy (its own wake-ups), and must not starve the rest.
    fn pump_output(&mut self, budget: Duration, rx: &Receiver<Event>) {
        let start = Instant::now();
        let focused: Vec<PaneId> =
            self.clients.values().filter(|c| c.attached).filter_map(|c| self.client_focus(c.id)).collect();
        for p in focused {
            while start.elapsed() < budget / 2 && rx.is_empty() {
                let Some(data) = self.panes.get(&p).and_then(|pn| pn.take_output()) else { break };
                self.pane_output(p, &data);
            }
        }
        let mut served = false;
        while !std::mem::replace(&mut served, true) || (start.elapsed() < budget && rx.is_empty()) {
            let Some(id) = self.ready.pop_front() else { break };
            match self.panes.get(&id).and_then(|pn| pn.take_output()) {
                Some(data) => {
                    self.pane_output(id, &data);
                    self.ready.push_back(id);
                }
                None => {
                    self.ready_set.remove(&id);
                }
            }
        }
    }

    fn next_deadline(&self) -> Instant {
        let mut d = self.last_tick + TICK;
        if let Some(r) = self.next_render {
            d = d.min(r);
        }
        for c in self.clients.values() {
            if let Some(p) = c.pending_since {
                d = d.min(p + INPUT_FLUSH);
            }
            if let Some((_, until)) = &c.toast {
                d = d.min(*until);
            }
        }
        for w in &self.waits {
            if let Some(dl) = w.deadline {
                d = d.min(dl);
            }
        }
        for (_, _, until) in self.exiting.values() {
            d = d.min(*until);
        }
        d.max(Instant::now() + Duration::from_millis(1))
    }

    fn timers(&mut self) {
        let now = Instant::now();
        // Escape sequences that never completed.
        let flush: Vec<ClientId> = self
            .clients
            .values()
            .filter(|c| c.pending_since.is_some_and(|p| now >= p + INPUT_FLUSH))
            .map(|c| c.id)
            .collect();
        for id in flush {
            self.flush_client_input(id);
        }
        let expired: Vec<ClientId> =
            self.clients.values().filter(|c| c.toast.as_ref().is_some_and(|(_, t)| *t <= now)).map(|c| c.id).collect();
        for id in expired {
            if let Some(c) = self.clients.get_mut(&id) {
                c.toast = None;
            }
            self.mark_dirty_client(id);
        }
        // Synchronized updates that timed out: what they held is drawn, and what the emulator
        // has to answer or report goes out.
        let flushed: Vec<(PaneId, Vec<PaneSignal>)> =
            self.panes.iter_mut().filter_map(|(id, p)| p.poll_sync_timeout().map(|s| (*id, s))).collect();
        for (id, signals) in flushed {
            self.pane_signals(id, signals);
            self.mark_dirty_pane(id);
        }
        self.finish_exits();
        self.release_held();
        if now >= self.last_tick + TICK {
            self.last_tick = now;
            self.tick();
        }
        self.check_waits();
    }

    /// Periodic work: agent detection, preview reloads, persistence.
    fn tick(&mut self) {
        let mut changed = Vec::new();
        let mut reloads = Vec::new();
        let mut ssh_seen = Vec::new();
        let mut ssh_ended = Vec::new();
        let mut read_contexts = false;
        let ids: Vec<PaneId> = self.panes.keys().copied().collect();
        for id in ids {
            let Some(pane) = self.panes.get_mut(&id) else { continue };
            if let crate::pane::Content::Preview(p) = &mut pane.content {
                if p.modified_since_load() && self.preview_loads.insert(id) {
                    reloads.push((id, p.path().to_path_buf()));
                }
                continue;
            }
            if pane.exit_code.is_some() {
                continue;
            }
            if let crate::pane::Content::Term(t) = &mut pane.content
                && std::mem::take(&mut t.input_dropped)
            {
                let name = pane.name.clone().unwrap_or_else(|| pane.title.clone());
                self.warnings.push(format!(
                    "{name} is not reading its input: what was typed or pasted beyond 16 MiB was dropped"
                ));
            }
            let before = pane.agent.state;
            let argv = pane.foreground_argv();
            if let Some(a) = &argv
                && ssh::runs_ssh(a)
            {
                ssh_seen.push((id, a[1..].to_vec()));
            } else if argv.is_some() && pane.ssh.is_some() {
                ssh_ended.push(id);
            }
            if let Some(a) = &argv
                && pane.ssh.is_none()
            {
                read_contexts |= contexts::observe_tools(&self.config.contexts, pane, a);
            }
            let agent = argv.as_deref().and_then(detect_agent);
            let mut moved = pane.agent.set_process(agent);
            if pane.agent.kind.is_some() {
                let screen = pane.live_text();
                let quiet = pane.last_output.elapsed();
                moved |= pane.agent.observe(&screen, quiet);
            }
            if moved {
                changed.push((id, Some(before)));
            }
        }
        for (id, before) in changed {
            if let Some(before) = before {
                self.on_agent_change(id, before);
            }
            self.mark_dirty();
        }
        for (id, args) in ssh_seen {
            self.observe_ssh(id, &args);
        }
        // A context file just seen in use is read now, not at the next scan.
        if read_contexts {
            self.refresh_kube();
            self.refresh_tf_dirs();
        }
        // Changed previews are read again off the event loop (images take a while to decode).
        for (id, path) in reloads {
            let started = self.background(
                move || crate::preview::Loaded::read(&path),
                move |server, loaded| {
                    server.preview_loads.remove(&id);
                    if let Some(crate::pane::Content::Preview(p)) = server.panes.get_mut(&id).map(|p| &mut p.content) {
                        p.apply(loaded);
                        server.mark_dirty();
                    }
                },
            );
            // No thread to read it: the next tick tries again.
            if !started {
                self.preview_loads.remove(&id);
            }
        }
        self.forget_ended_ssh(ssh_ended);
        self.tick_edits();
        self.forget_ended_reports();
        self.diff_pane_states();
        self.hooks.reap();
        self.check_run_timeouts();
        self.show_warnings();
        // The sidebar appears (or goes) with what it has to show.
        let useful = self.sidebar_useful();
        if useful != self.sidebar_was_useful {
            self.sidebar_was_useful = useful;
            for c in self.clients.values_mut() {
                c.encoder.invalidate();
            }
            self.relayout();
            self.mark_dirty();
        }
        // Wake-ups can be dropped when the event queue is full; catch those panes here.
        let waiting: Vec<PaneId> = self.panes.values().filter(|p| p.has_output()).map(|p| p.id).collect();
        for id in waiting {
            if self.ready_set.insert(id) {
                self.ready.push_back(id);
            }
        }
        // Leave copy mode in panes no client is browsing any more (focus moved, client gone).
        let browsing: Vec<PaneId> = self
            .clients
            .values()
            .filter(|c| matches!(c.mode, Mode::Scroll | Mode::Search { .. }))
            .filter_map(|c| self.client_focus(c.id))
            .collect();
        for p in self.panes.values_mut() {
            if p.in_copy_mode() && !browsing.contains(&p.id) {
                p.copy_mode(false);
            }
        }
        if self.last_usage_load.elapsed() > Duration::from_secs(3) {
            self.reload_usage();
            self.check_limits();
        }
        if self.last_git_scan.elapsed() > Duration::from_secs(3) {
            self.refresh_git();
            self.refresh_kube();
            self.refresh_tf_dirs();
            self.refresh_ports();
        }
        self.check_tunnels();
        self.check_remotes();
        if self.last_codex_poll.elapsed() > Duration::from_secs(300) {
            self.poll_codex_usage();
        }
        let expired: Vec<PaneId> =
            self.panes.values().filter(|p| p.close_at.is_some_and(|t| t <= Instant::now())).map(|p| p.id).collect();
        for id in expired {
            self.close_pane(id);
        }
        self.save_scrollback(false);
        if self.persist_dirty || self.last_persist.elapsed() > Duration::from_secs(30) {
            self.persist();
        }
        if self.last_def_scan.elapsed() > Duration::from_secs(5) {
            self.scan_definitions();
        }
    }

    fn handle(&mut self, ev: Event) {
        match ev {
            Event::PaneReady(id) => {
                if self.ready_set.insert(id) {
                    self.ready.push_back(id);
                }
            }
            Event::PaneExited(id, pid, code) => {
                self.exiting.insert(id, (pid, code, Instant::now() + Duration::from_secs(1)));
                self.finish_exits();
            }
            Event::Connection(stream) => self.new_connection(stream),
            Event::ClientFrame(id, kind, payload) => self.client_frame(id, kind, payload),
            Event::ClientGone(id) => self.drop_client(id),
            Event::Git(dir, info) => self.git_result(dir, info),
            Event::Warning(w) => self.warnings.push(w),
            Event::Emit(ty, fields) => self.emit(ty, fields),
            Event::Job(id, result) => {
                if let Some(done) = self.jobs.remove(&id) {
                    done(self, result);
                }
            }
            Event::RemoteAgents(conn, list) => self.remote_list(conn, list),
            Event::Wake => {
                let mut changed = false;
                for p in self.panes.values_mut() {
                    if let crate::pane::Content::Files(f) = &mut p.content {
                        changed |= f.poll();
                    }
                }
                if changed {
                    self.mark_dirty();
                }
            }
            Event::Shutdown => self.quitting = true,
        }
        // Whatever was to fill a new workspace and could not start leaves no empty workspace
        // behind (one is never shown without a tab).
        self.drop_empty_workspaces();
    }

    fn pane_output(&mut self, id: PaneId, data: &[u8]) {
        self.note_ports(id, data);
        if let Some(out) = self.run_output.get_mut(&id) {
            out.feed(data);
        }
        let Some(pane) = self.panes.get_mut(&id) else { return };
        let signals = pane.feed(data);
        self.pane_signals(id, signals);
        self.mark_dirty_pane(id);
    }

    /// Exits whose last output has been read, or that waited long enough: a little of what is
    /// left is read each time, never the endless output of a program left in the background.
    fn finish_exits(&mut self) {
        let ids: Vec<PaneId> = self.exiting.keys().copied().collect();
        for id in ids {
            let mut left = 16;
            let drained = loop {
                if left == 0 {
                    break false;
                }
                let Some(data) = self.panes.get(&id).and_then(|p| p.take_output()) else { break true };
                self.pane_output(id, &data);
                left -= 1;
            };
            let Some(&(pid, code, until)) = self.exiting.get(&id) else { continue };
            if drained || Instant::now() >= until || !self.panes.contains_key(&id) {
                self.exiting.remove(&id);
                self.pane_exited(id, pid, code);
            }
        }
    }

    /// Act on what feeding output to pane `id` asked for.
    fn pane_signals(&mut self, id: PaneId, signals: Vec<PaneSignal>) {
        for s in signals {
            match s {
                PaneSignal::Notification { title, body } => {
                    let title = if title.is_empty() { self.pane_label(id) } else { title };
                    self.notify(&title, &body, Some(id));
                }
                PaneSignal::Clipboard(text) => {
                    // What a pane copies goes to one clipboard: that of the client looking at its
                    // workspace that was used last (another machine attached elsewhere keeps its own).
                    let bytes = crate::pane::osc52(&text);
                    let ws = self.workspace_of(id).map(|(wi, _)| self.workspaces[wi].id);
                    if let Some(c) = self
                        .clients
                        .values()
                        .filter(|c| c.attached && Some(c.workspace) == ws)
                        .max_by_key(|c| c.last_input)
                    {
                        let _ = c.out.try_send(frame_bytes(&bytes));
                    }
                }
                PaneSignal::CommandDone(done) => self.command_done(id, done),
                PaneSignal::Contexts(contexts) => self.contexts_reported(id, contexts),
                PaneSignal::Bell | PaneSignal::TitleChanged => {}
            }
        }
    }

    // ---- rendering ---------------------------------------------------------------------

    pub fn mark_dirty(&mut self) {
        for c in self.clients.values_mut() {
            c.dirty = true;
        }
        self.schedule_render();
    }

    pub fn mark_dirty_client(&mut self, id: ClientId) {
        if let Some(c) = self.clients.get_mut(&id) {
            c.dirty = true;
        }
        self.schedule_render();
    }

    fn mark_dirty_pane(&mut self, pane: PaneId) {
        let ws = self.workspace_of(pane).map(|(w, _)| self.workspaces[w].id);
        for c in self.clients.values_mut() {
            if Some(c.workspace) == ws {
                c.dirty = true;
            }
        }
        self.schedule_render();
    }

    fn schedule_render(&mut self) {
        if self.next_render.is_none() {
            self.next_render = Some(Instant::now() + Duration::from_millis(2));
        }
    }

    fn render_clients(&mut self) {
        self.next_render = None;
        let ids: Vec<ClientId> = self.clients.values().filter(|c| c.attached && c.dirty).map(|c| c.id).collect();
        let mut again = false;
        for id in ids {
            // Slow client: keep it dirty and retry on the next frame.
            if self.clients[&id].out.len() >= 8 {
                again = true;
                continue;
            }
            let (buf, cursor, links) = self.compose(id);
            // Images the client's terminal does not have yet go first.
            let (mut bytes, more) = self.graphics_updates(id);
            let Some(c) = self.clients.get_mut(&id) else { continue };
            c.encoder.encode(&buf, cursor, &links, &mut bytes);
            c.dirty = more;
            again |= more;
            match c.out.try_send(frame_bytes(&bytes)) {
                Ok(()) => {}
                Err(TrySendError::Full(_)) => {
                    c.encoder.invalidate();
                    c.dirty = true;
                    again = true;
                    if let Some(ci) = self.images.get_mut(&id) {
                        ci.forget();
                    }
                }
                Err(TrySendError::Disconnected(_)) => {}
            }
        }
        if again {
            self.next_render = Some(Instant::now() + FRAME);
        }
    }

    fn shutdown(&mut self) {
        self.end_edits();
        // The last state goes through the writer, after anything it still has to write (the
        // panes' scrollback first), and the server waits for it.
        self.save_scrollback(true);
        self.persist();
        let (done, written) = crossbeam_channel::bounded(1);
        let saved = match self.persister.send(persistence::Persist::Stop(done)) {
            Ok(()) => written
                .recv_timeout(Duration::from_secs(10))
                .unwrap_or_else(|_| Err("its writer did not finish in time".into())),
            Err(_) => Err("its writer is gone".into()),
        };
        if let Err(e) = &saved {
            eprintln!("kraal: the session state was not saved: {e}");
        }
        // `kraal kill` says whether this last state, the one a restart brings back, is saved.
        let answer = match saved {
            Ok(()) => Ok(crate::protocol::Response::Ok),
            Err(e) => Ok(crate::protocol::Response::Text(format!("the layout could not be saved: {e}"))),
        };
        for id in std::mem::take(&mut self.kill_waiters) {
            self.send_msg(id, &ServerMsg::Response { result: answer.clone() });
            self.clients.remove(&id);
        }
        self.interrupt_runs();
        // What the runs wrote is in their logs before the server goes.
        let (done, flushed) = crossbeam_channel::bounded(1);
        if self.run_logs.send(crate::runbooks::LogJob::Flush(done)).is_ok() {
            let _ = flushed.recv_timeout(Duration::from_secs(10));
        }
        self.hooks.stop_all();
        let ids: Vec<ClientId> = self.clients.keys().copied().collect();
        for id in ids {
            self.send_msg(id, &ServerMsg::Exit { reason: "server exited".into() });
        }
        for p in self.panes.values_mut() {
            p.kill();
        }
        self.stop_tunnels();
        self.stop_remotes();
        // Give writer threads a moment to deliver the exit messages.
        std::thread::sleep(Duration::from_millis(50));
    }
}