scv_tools/

workbench.rs

//! The daemon's workbench: one place that knows every delegated agent, background
//! job, and session of the instance, and lets the owner act on them.
//!
//! What it shows comes from the durable records every SCV process of the
//! instance writes (the job ledger, each run's activity log and state, and
//! the delegation records), so jobs and runs of `scv exec` servers, and of
//! processes that stopped, are listed too. What it does needs the live
//! objects, so only sessions of this process can be acted on: each
//! registers with [`Workbench::register_session`], and its `agent` tool
//! attaches its job store when the registry builds it.
//!
//! The owner acts on:
//!
//! - a conversation: [`Workbench::send`] gives it the owner's prompt as a
//!   new turn, queued behind a running one, or steered into it;
//! - a job: [`Workbench::cancel`] stops it, [`Workbench::resume`] runs an
//!   interrupted one again, [`Workbench::dismiss`] forgets a finished one;
//! - an approval request a background job relayed while the owner watches
//!   ([`Workbench::watch`]): it waits for [`Workbench::approve`] up to
//!   [`ATTENDED_WAIT`], then gets the session's unattended answer;
//! - a conversation taken over in its agent's own terminal
//!   ([`Workbench::attach_terminal`], or [`Workbench::interrupt_and_attach`],
//!   which stops the job running its turn once its lane is held): the owner
//!   reads its screen and types into it ([`Workbench::attach_io`]), detaches
//!   while it keeps running ([`Workbench::detach`]), and ends it
//!   ([`Workbench::end_takeover`]). A takeover ends with its session, and
//!   with the daemon. The snapshot says of each conversation whether it can
//!   be taken over now, and why not.

use std::{
    collections::{BTreeMap, HashMap, HashSet},
    path::{Path, PathBuf},
    sync::{
        Arc, Mutex, Weak,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use async_trait::async_trait;
use scv_client::Layout;
use scv_core::{AgentError, ApprovalGate, ApprovalRequest};
use scv_protocol::{
    AgentState, AttachOffer, AttachPage, AttachState, JobResult, MAX_JOB_RESULT_BYTES, RunLogPage,
    SendMode, WorkbenchApproval, WorkbenchJob, WorkbenchRun, WorkbenchSession, WorkbenchSnapshot,
};
use serde_json::{Value, json};
use tokio::sync::oneshot;
use tokio_util::sync::CancellationToken;

use crate::{
    delegate::{
        activity::{self, RunState},
        background::{
            BackgroundCapable, BackgroundJobs, JobMeta, LiveJob, OwnerCall,
            ledger::{self, JobRecord},
        },
        records::{DelegationRegistry, ProcessIdentity},
        takeover::{self, Takeover},
    },
    sync::lock,
};

/// How long a background job's approval request waits for the owner while
/// the owner watches, before it gets the session's unattended answer.
pub const ATTENDED_WAIT: Duration = Duration::from_secs(120);
/// How long one `watch` keeps the owner counted as watching.
pub const WATCH_TTL: Duration = Duration::from_secs(15);
/// Jobs and runs a snapshot lists at most: every unfinished one, then the
/// newest finished ones.
const MAX_LISTED: usize = 200;
/// Finished jobs and ended runs are kept this long.
pub const KEEP: Duration = Duration::from_secs(7 * 24 * 3600);
/// Resume refused to start a fresh conversation for a follow-up. The message
/// stays pending; the owner is told once.
pub const UNRESUMABLE_FOLLOW_UP: &str = "the subscribed job's conversation could not be resumed, so the follow-up was not started as new work";
/// Ended runs kept at most.
pub const MAX_RUNS: usize = 500;
/// How long a client attached to a taken-over terminal counts as watching
/// after its last read: one that stops reading (its dashboard closed or
/// died) is detached, and the agent keeps running.
pub const ATTACH_LEASE: Duration = Duration::from_secs(30);
/// How long a takeover whose terminal closed is kept, so a client still
/// attached reads how it ended.
const CLOSED_KEPT: Duration = Duration::from_secs(60);
/// Takeovers that may run at once.
const MAX_TAKEOVERS: usize = 8;

/// The daemon's workbench.
pub struct Workbench {
    runs: PathBuf,
    jobs: PathBuf,
    subscriptions: PathBuf,
    worktrees: PathBuf,
    delegations: Arc<DelegationRegistry>,
    inner: Mutex<Inner>,
    watched_until: Mutex<Option<Instant>>,
    /// Jobs being resumed now, `<session>/<job>`: each is resumed once.
    resuming: Mutex<HashSet<String>>,
    /// Conversations taken over in their agents' terminals, by
    /// `<session>/<conversation>`.
    takeovers: Mutex<HashMap<String, Held>>,
    /// Held while a takeover starts, so two `attach`es of one conversation
    /// do not both start one. Attaching to one that runs never waits for it.
    attaching: tokio::sync::Mutex<()>,
    /// SCV is stopping: no takeover starts any more. Set and read with
    /// `takeovers` locked, so none is kept after they all were ended.
    closing: AtomicBool,
}

/// A takeover, and the client attached to it.
struct Held {
    takeover: Arc<Takeover>,
    /// The attached client's hold, and when it last read.
    lease: Option<(String, Instant)>,
    /// When the terminal was first seen closed.
    closed: Option<Instant>,
}

impl std::fmt::Debug for Workbench {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.debug_struct("Workbench").finish_non_exhaustive()
    }
}

#[derive(Default)]
struct Inner {
    sessions: HashMap<String, Slot>,
    approvals: BTreeMap<String, Pending>,
    next: u64,
}

/// A live session of this process.
struct Slot {
    token: u64,
    label: String,
    workspace: PathBuf,
    state: AgentState,
    detail: Option<String>,
    agent: Option<Weak<BackgroundCapable>>,
    jobs: Option<Weak<BackgroundJobs>>,
}

struct Pending {
    info: WorkbenchApproval,
    answer: Option<oneshot::Sender<bool>>,
}

/// A session's place in the workbench while it lives; dropping it removes
/// the session.
pub struct SessionHandle {
    workbench: Weak<Workbench>,
    id: String,
    token: u64,
}

impl std::fmt::Debug for SessionHandle {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SessionHandle")
            .field("id", &self.id)
            .finish()
    }
}

impl SessionHandle {
    /// What the session is doing: a turn runs, an approval waits for its
    /// client, or it is idle.
    pub fn set_state(&self, state: AgentState, detail: Option<String>) {
        let Some(workbench) = self.workbench.upgrade() else {
            return;
        };
        if let Some(slot) = lock(&workbench.inner).sessions.get_mut(&self.id)
            && slot.token == self.token
        {
            slot.state = state;
            slot.detail = detail;
        }
    }
}

impl Drop for SessionHandle {
    fn drop(&mut self) {
        let Some(workbench) = self.workbench.upgrade() else {
            return;
        };
        let mut inner = lock(&workbench.inner);
        if inner
            .sessions
            .get(&self.id)
            .is_some_and(|slot| slot.token == self.token)
        {
            inner.sessions.remove(&self.id);
            drop(inner);
            // Its conversations go with it, taken over or not.
            workbench.end_takeovers_of(Some(&self.id));
        }
    }
}

/// What [`Workbench::send`] did with the owner's prompt.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Sent {
    /// A background job of the session runs it: `<session>/<job>`.
    Job(String),
    /// It went into the conversation's running turn.
    Steered,
}

/// What a maintenance pass did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Maintenance {
    /// Jobs found interrupted: their SCV process stopped.
    pub interrupted_jobs: Vec<String>,
    /// Runs found interrupted.
    pub interrupted_runs: Vec<String>,
    /// Old records removed.
    pub pruned: usize,
}

impl Workbench {
    /// The workbench of the instance at `layout`, whose delegated runs
    /// `delegations` records.
    pub fn new(layout: &Layout, delegations: Arc<DelegationRegistry>) -> Arc<Self> {
        Arc::new(Self {
            runs: layout.runs(),
            jobs: layout.jobs(),
            subscriptions: layout.subscriptions(),
            worktrees: layout.worktrees(),
            delegations,
            inner: Mutex::default(),
            watched_until: Mutex::new(None),
            resuming: Mutex::default(),
            takeovers: Mutex::default(),
            attaching: tokio::sync::Mutex::const_new(()),
            closing: AtomicBool::new(false),
        })
    }

    /// Where the job ledger is.
    pub fn jobs_dir(&self) -> &Path {
        &self.jobs
    }

    /// Register a live session; `label` says who it serves.
    pub fn register_session(
        self: &Arc<Self>,
        id: &str,
        label: &str,
        workspace: &Path,
    ) -> SessionHandle {
        let mut inner = lock(&self.inner);
        inner.next += 1;
        let token = inner.next;
        inner.sessions.insert(
            id.to_owned(),
            Slot {
                token,
                label: label.to_owned(),
                workspace: workspace.to_owned(),
                state: AgentState::Idle,
                detail: None,
                agent: None,
                jobs: None,
            },
        );
        SessionHandle {
            workbench: Arc::downgrade(self),
            id: id.to_owned(),
            token,
        }
    }

    /// The `agent` tool and job store of live session `session`.
    pub(crate) fn attach(
        &self,
        session: &str,
        agent: Weak<BackgroundCapable>,
        jobs: Weak<BackgroundJobs>,
    ) {
        if let Some(slot) = lock(&self.inner).sessions.get_mut(session) {
            slot.agent = Some(agent);
            slot.jobs = Some(jobs);
        }
    }

    /// The owner is watching: background approval requests wait for an
    /// answer for the next [`WATCH_TTL`].
    pub fn watch(&self) {
        *lock(&self.watched_until) = Some(Instant::now() + WATCH_TTL);
    }

    fn watched(&self) -> bool {
        lock(&self.watched_until).is_some_and(|until| Instant::now() < until)
    }

    /// The approval gate for background jobs of `session`: `unattended`
    /// decides, except that a request it would deny waits up to
    /// [`ATTENDED_WAIT`] for the owner while the owner watches. It never
    /// denies what `unattended` allows.
    pub fn attended_gate(
        self: &Arc<Self>,
        session: &str,
        unattended: Arc<dyn ApprovalGate>,
    ) -> Arc<dyn ApprovalGate> {
        Arc::new(AttendedGate {
            workbench: Arc::downgrade(self),
            session: session.to_owned(),
            unattended,
        })
    }

    /// Hold `request` of a job of `session` for the owner's answer.
    async fn ask(
        &self,
        session: &str,
        request: ApprovalRequest,
        cancellation: &CancellationToken,
    ) -> Result<bool, AgentError> {
        let (sender, receiver) = oneshot::channel();
        let now = unix_now();
        let id = format!("ap-{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
        lock(&self.inner).approvals.insert(
            id.clone(),
            Pending {
                info: WorkbenchApproval {
                    id: id.clone(),
                    session: session.to_owned(),
                    job: request.call_id.clone(),
                    name: request.name.clone(),
                    risk: request.risk.as_str().to_owned(),
                    cwd: request.cwd.display().to_string(),
                    summary: request.summary.clone(),
                    since_unix_seconds: now,
                    expires_unix_seconds: now + ATTENDED_WAIT.as_secs(),
                },
                answer: Some(sender),
            },
        );
        // An answer that lands as the deadline fires is still the answer.
        // Closing the receiver and taking anything already sent stops the
        // waiter from reporting a denial after `approve` has sent `true`.
        let mut receiver = receiver;
        let answer = tokio::select! {
            biased;
            answer = &mut receiver => Ok(answer.unwrap_or(false)),
            () = tokio::time::sleep(ATTENDED_WAIT) => {
                receiver.close();
                Ok(receiver.try_recv().unwrap_or(false))
            }
            () = cancellation.cancelled() => {
                receiver.close();
                match receiver.try_recv() {
                    Ok(approved) => Ok(approved),
                    Err(_) => Err(AgentError::Cancelled),
                }
            }
        };
        lock(&self.inner).approvals.remove(&id);
        answer
    }

    /// Answer approval request `id`.
    pub fn approve(&self, id: &str, approved: bool) -> Result<String, String> {
        let pending = lock(&self.inner).approvals.get_mut(id).and_then(|pending| {
            pending
                .answer
                .take()
                .map(|answer| (answer, pending.info.clone()))
        });
        let Some((answer, info)) = pending else {
            return Err(format!(
                "no approval request {id:?} waits; it may have been answered or timed out"
            ));
        };
        if answer.send(approved).is_err() {
            return Err(format!("{} no longer waits for this answer", info.job));
        }
        Ok(format!(
            "{} {}: {}",
            if approved { "Approved" } else { "Denied" },
            info.job,
            bounded(&info.summary, 120)
        ))
    }

    /// Everything the workbench shows now. `restorable` is left for the
    /// server, which keeps session queues.
    pub fn snapshot(&self) -> WorkbenchSnapshot {
        self.expire_takeovers();
        let now = unix_now();
        let live_handles = self.live_handles();
        let (slots, approvals): (Vec<_>, Vec<_>) = {
            let inner = lock(&self.inner);
            (
                inner
                    .sessions
                    .iter()
                    .map(|(id, slot)| SlotView {
                        id: id.clone(),
                        label: slot.label.clone(),
                        workspace: slot.workspace.clone(),
                        state: slot.state,
                        detail: slot.detail.clone(),
                        agent: slot.agent.as_ref().and_then(Weak::upgrade),
                        jobs: slot.jobs.as_ref().and_then(Weak::upgrade),
                    })
                    .collect(),
                inner
                    .approvals
                    .values()
                    .map(|pending| pending.info.clone())
                    .collect(),
            )
        };
        let live_jobs: HashMap<(String, String), LiveJob> = slots
            .iter()
            .filter_map(|slot| {
                slot.jobs
                    .as_ref()
                    .map(|jobs| (slot.id.clone(), jobs.live()))
            })
            .flat_map(|(session, jobs)| {
                jobs.into_iter()
                    .map(move |job| ((session.clone(), job.id.clone()), job))
            })
            .collect();
        let live_sessions: HashSet<&str> = slots.iter().map(|slot| slot.id.as_str()).collect();
        let awaiting: HashMap<(String, String), &WorkbenchApproval> = approvals
            .iter()
            .map(|approval| ((approval.session.clone(), approval.job.clone()), approval))
            .collect();

        // A conversation handle is shown only while the live session that
        // made it still has it, so the owner can send it a prompt: one an
        // earlier daemon, or an earlier session under an ID that came back
        // (a workbench session), left names nothing now. Such a session
        // numbers its conversations after the ones its records name, so a
        // handle never comes to mean another conversation.
        let me = ProcessIdentity::current();
        let held = |owner: &ProcessIdentity, session: &str, handle: &str| {
            me.as_ref() == Some(owner)
                && slots.iter().any(|slot| {
                    slot.id == session
                        && slot.agent.as_ref().is_some_and(|agent| agent.knows(handle))
                })
        };
        // Whether the owner can take each such conversation over now.
        let taken_over = self.running_takeovers();
        let mut offers: HashMap<(String, String), Option<AttachOffer>> = HashMap::new();
        let mut offer = |session: &str, handle: &str| {
            offers
                .entry((session.to_owned(), handle.to_owned()))
                .or_insert_with(|| {
                    let slot = slots.iter().find(|slot| slot.id == session)?;
                    let offer = slot.agent.as_ref()?.attach_offer(handle)?;
                    let approval = offer.job.as_ref().and_then(|job| {
                        awaiting
                            .get(&(session.to_owned(), job.clone()))
                            .map(|approval| approval.id.as_str())
                    });
                    Some(settle_offer(offer, approval, taken_over))
                })
                .clone()
        };
        let mut states = activity::list_states(&self.runs);
        states
            .sort_by_key(|state| std::cmp::Reverse((state.last_activity_unix, state.started_unix)));
        let mut runs: Vec<WorkbenchRun> = Vec::new();
        let mut job_runs: HashMap<(String, String), &RunState> = HashMap::new();
        for state in &states {
            for job in state.jobs.iter().chain(&state.job) {
                job_runs
                    .entry((state.session.clone(), job.clone()))
                    .or_insert(state);
            }
        }
        let mut ended_listed = 0;
        for state in &states {
            if state.ended {
                if ended_listed >= MAX_LISTED {
                    continue;
                }
                ended_listed += 1;
            }
            let (mut current, mut detail) = activity::current(state, now);
            if let Some(job) = &state.job
                && let Some(approval) = awaiting.get(&(state.session.clone(), job.clone()))
                && !state.ended
            {
                current = AgentState::AwaitingInput;
                detail = Some(format!(
                    "waiting for your approval: {}",
                    bounded(&approval.summary, 200)
                ));
            }
            let conversation = state
                .conversation
                .clone()
                .filter(|handle| held(&state.owner, &state.session, handle));
            runs.push(WorkbenchRun {
                handle: state.handle.clone(),
                agent: state.agent.clone(),
                session: state.session.clone(),
                transport: state.transport.clone(),
                attach: conversation
                    .as_deref()
                    .and_then(|handle| offer(&state.session, handle)),
                conversation,
                turn: state.turn,
                job: state.job.clone(),
                jobs: state.jobs.clone(),
                cwd: state.cwd.display().to_string(),
                state: current,
                detail,
                started_unix_seconds: state.started_unix,
                last_activity_unix_seconds: state.last_activity_unix,
                live: !state.ended && live_handles.contains(&state.handle),
                steerable: state.steerable,
            });
        }

        let all = ledger::list_all(&self.jobs);
        let mut records: Vec<&JobRecord> = all.iter().filter(|record| !record.dismissed).collect();
        records.sort_by_key(|record| std::cmp::Reverse((record.created_unix, record.job.clone())));
        let mut jobs = Vec::new();
        let mut finished_listed = 0;
        // What each job was resumed as, by the record of the job resuming
        // it too, which is written first: a dismissed one's included, or
        // the job would be resumable again.
        let successors: HashMap<&str, String> = all
            .iter()
            .filter_map(|record| {
                let from = record.resumed_from.as_deref()?;
                Some((from, format!("{}/{}", record.session, record.job)))
            })
            .collect();
        for &record in &records {
            let key = (record.session.clone(), record.job.clone());
            let resumed_as = record.resumed_as.clone().or_else(|| {
                successors
                    .get(format!("{}/{}", record.session, record.job).as_str())
                    .cloned()
            });
            let live = live_jobs.get(&key);
            let run = job_runs.get(&key).copied();
            let (state, detail) = job_state(
                record,
                live,
                run,
                awaiting.get(&key).copied(),
                live_sessions.contains(record.session.as_str()),
                now,
            );
            // An interrupted job the owner can still resume stays listed
            // after the cap of other finished jobs. Counting it as finished
            // would hide it once newer completions filled the snapshot.
            let resumable_interrupted =
                state == AgentState::Interrupted && resumed_as.is_none() && !record.reviewed;
            if state.is_final() && !resumable_interrupted {
                if finished_listed >= MAX_LISTED {
                    continue;
                }
                finished_listed += 1;
            }
            let cwd = run
                .map(|run| run.cwd.display().to_string())
                .or_else(|| {
                    record
                        .worktree
                        .as_ref()
                        .map(|worktree| worktree.path.clone())
                })
                .unwrap_or_else(|| {
                    let relative = record.arguments.get("cwd").and_then(Value::as_str);
                    relative.map_or_else(
                        || record.workspace.display().to_string(),
                        |cwd| record.workspace.join(cwd).display().to_string(),
                    )
                });
            let conversation = record
                .conversation
                .clone()
                .or_else(|| run.and_then(|run| run.conversation.clone()))
                .or_else(|| {
                    record
                        .arguments
                        .get("session")
                        .and_then(Value::as_str)
                        .map(str::to_owned)
                })
                .filter(|handle| held(&record.owner, &record.session, handle));
            jobs.push(WorkbenchJob {
                session: record.session.clone(),
                job: record.job.clone(),
                agent: record.agent.clone(),
                task: record.task.clone(),
                workspace: record.workspace.display().to_string(),
                cwd,
                state,
                detail,
                started_by: record.started_by,
                attach: conversation
                    .as_deref()
                    .and_then(|handle| offer(&record.session, handle)),
                conversation,
                run: run.map(|run| run.handle.clone()),
                worktree: record.worktree.clone(),
                reviewed: record.reviewed,
                resumable: state == AgentState::Interrupted
                    && !record.reviewed
                    && resumed_as.is_none(),
                resumed_as,
                created_unix_seconds: record.created_unix,
                updated_unix_seconds: record.updated_unix,
            });
        }

        let mut sessions: Vec<WorkbenchSession> = slots
            .iter()
            .map(|slot| WorkbenchSession {
                id: slot.id.clone(),
                label: slot.label.clone(),
                workspace: slot.workspace.display().to_string(),
                live: true,
                state: slot.state,
                detail: slot.detail.clone(),
            })
            .collect();
        let mut seen: HashSet<String> = sessions.iter().map(|session| session.id.clone()).collect();
        for &record in &records {
            if seen.insert(record.session.clone()) {
                let alive = record.owner.is_alive() && record.owner.pid != std::process::id();
                sessions.push(WorkbenchSession {
                    id: record.session.clone(),
                    label: record.label.clone(),
                    workspace: record.workspace.display().to_string(),
                    live: alive,
                    state: AgentState::Unknown,
                    detail: Some(if alive {
                        format!("runs in SCV process {}", record.owner.pid)
                    } else {
                        "ended".to_owned()
                    }),
                });
            }
        }
        sessions.sort_by(|a, b| b.live.cmp(&a.live).then_with(|| a.id.cmp(&b.id)));
        WorkbenchSnapshot {
            generated_unix_seconds: now,
            sessions,
            jobs,
            runs,
            approvals,
            restorable: Vec::new(),
        }
    }

    /// The runs whose processes still run, in any SCV process of the
    /// instance.
    fn live_handles(&self) -> HashSet<String> {
        self.delegations
            .list(true)
            .into_iter()
            .filter(|entry| entry.processes > 0)
            .map(|entry| entry.record.handle)
            .collect()
    }

    /// Refuse to remove the worktree holding `path` while anything still
    /// works in it ([`Self::worktree_users`]).
    pub fn check_worktree_unused(&self, path: &Path) -> Result<(), String> {
        let users = self.worktree_users(path);
        if users.is_empty() {
            return Ok(());
        }
        Err(format!(
            "{} still work in {}; stop them before removing it",
            users.join(", "),
            path.display()
        ))
    }

    /// What still runs in the SCV worktree holding `path` (its top, or any
    /// directory in it): jobs not yet finished that recorded it, and
    /// delegated runs whose processes work inside it.
    pub fn worktree_users(&self, path: &Path) -> Vec<String> {
        let path = crate::delegate::worktree::worktree_path(&self.worktrees, path)
            .or_else(|| std::fs::canonicalize(path).ok())
            .unwrap_or_else(|| path.to_owned());
        let mut users: Vec<String> = ledger::list(&self.jobs)
            .into_iter()
            .filter(|record| {
                !record.state.is_final()
                    && record.owner.is_alive()
                    && record
                        .worktree
                        .as_ref()
                        .is_some_and(|worktree| Path::new(&worktree.path) == path)
            })
            .map(|record| format!("{}/{}", record.session, record.job))
            .collect();
        users.extend(
            self.delegations
                .list(true)
                .into_iter()
                .filter(|entry| entry.processes > 0 && entry.record.cwd.starts_with(&path))
                .map(|entry| entry.record.handle),
        );
        users
    }

    /// One page of run `handle`'s activity log after byte position `after`.
    pub fn run_log(&self, handle: &str, after: u64, limit: usize) -> Result<RunLogPage, String> {
        activity::read_page(&self.runs, handle, after, limit).map_err(|error| {
            if error.kind() == std::io::ErrorKind::NotFound {
                format!("no activity log for run {handle:?}")
            } else {
                format!("cannot read run {handle:?}: {error}")
            }
        })
    }

    /// The `agent` tool of live session `session`.
    fn agent(&self, session: &str) -> Result<(Arc<BackgroundCapable>, PathBuf), String> {
        let inner = lock(&self.inner);
        let slot = inner.sessions.get(session).ok_or_else(|| {
            format!("session {session} does not run in this daemon (it ended, or another SCV process runs it)")
        })?;
        let agent = slot
            .agent
            .as_ref()
            .and_then(Weak::upgrade)
            .ok_or_else(|| format!("session {session} runs no background agent jobs"))?;
        Ok((agent, slot.workspace.clone()))
    }

    /// Give conversation `conversation` of live session `session` the
    /// owner's `prompt`.
    pub async fn send(
        &self,
        session: &str,
        conversation: &str,
        prompt: &str,
        mode: SendMode,
    ) -> Result<Sent, String> {
        let (agent, workspace) = self.agent(session)?;
        let call = agent
            .owner_call(
                json!({"session": conversation, "prompt": prompt}),
                workspace,
                mode,
                JobMeta::default(),
            )
            .await
            .map_err(|error| error.message)?;
        Ok(match call {
            OwnerCall::Job(started) => Sent::Job(job_name(session, &started)),
            OwnerCall::Steered => Sent::Steered,
        })
    }

    /// Whether live session `session` starts a report turn for a finished
    /// model job. A workbench session does not.
    fn reports_jobs(&self, session: &str) -> bool {
        lock(&self.inner)
            .sessions
            .get(session)
            .and_then(|slot| slot.jobs.as_ref())
            .and_then(Weak::upgrade)
            .is_some_and(|jobs| jobs.has_report_turn())
    }

    /// Origin of a subscription follow-up started in `session`. A session
    /// that reports jobs keeps it for that report turn. One that does not
    /// marks it reported when it finishes.
    fn follow_up_meta(
        &self,
        session: &str,
        resumed_from: Option<String>,
        accepted_follow_up: Option<String>,
    ) -> JobMeta {
        let reports = self.reports_jobs(session);
        JobMeta {
            origin: if reports {
                scv_protocol::JobOrigin::Model
            } else {
                scv_protocol::JobOrigin::Owner
            },
            keep_origin: reports,
            resumed_from,
            accepted_follow_up,
            ..JobMeta::default()
        }
    }

    /// Give conversation `conversation` of live session `session` the owner's
    /// `prompt`, queued behind a running turn. When `resume_from` is
    /// `<session>/<job>`, the new job's record links back and that job records
    /// `resumed_as`. A prompt that went into a running turn writes no link,
    /// because no new job started.
    pub async fn deliver_to_job(
        &self,
        session: &str,
        conversation: &str,
        prompt: &str,
        resume_from: Option<String>,
        accepted_follow_up: Option<String>,
    ) -> Result<Sent, String> {
        let (agent, workspace) = self.agent(session)?;
        // A chat session reports the follow-up the way it reports any
        // background job. A workbench session runs no report turn, so the
        // job is marked reported when it finishes and does not hold
        // `scv restart --when-idle`. The acceptance token is on the job's
        // record before this returns; the subscription cursor moves only
        // after that. Queued delivery does not steer, so the record is the
        // acceptance.
        let meta = self.follow_up_meta(session, resume_from.clone(), accepted_follow_up);
        let prompt = crate::background::subscription::worker_prompt(prompt);
        let call = agent
            .owner_call(
                json!({"session": conversation, "prompt": prompt}),
                workspace,
                SendMode::Queue,
                meta,
            )
            .await
            .map_err(|error| error.message)?;
        let sent = match call {
            OwnerCall::Job(started) => Sent::Job(job_name(session, &started)),
            OwnerCall::Steered => Sent::Steered,
        };
        if let (Some(from), Sent::Job(successor)) = (&resume_from, &sent)
            && let Some((session, job)) = from.split_once('/')
        {
            ledger::update(&self.jobs, session, job, |old| {
                old.resumed_as = Some(successor.clone());
            });
        }
        Ok(sent)
    }

    /// Start the owner's own job in live session `session` with the `agent`
    /// call `arguments`. Returns `<session>/<job>`.
    pub async fn start(&self, session: &str, arguments: Value) -> Result<String, String> {
        self.start_with(session, arguments, JobMeta::default())
            .await
    }

    /// [`Self::start`], with what the job's record holds from the start.
    async fn start_with(
        &self,
        session: &str,
        arguments: Value,
        meta: JobMeta,
    ) -> Result<String, String> {
        let (agent, workspace) = self.agent(session)?;
        match agent
            .owner_call(arguments, workspace, SendMode::Queue, meta)
            .await
            .map_err(|error| error.message)?
        {
            OwnerCall::Job(started) => Ok(job_name(session, &started)),
            OwnerCall::Steered => Err("the prompt went into a running turn".into()),
        }
    }

    /// Run interrupted job `job` of `session` again as the owner's job in
    /// live session `host`, which serves the job's workspace. It continues
    /// the agent's own conversation when the agent saved one and can resume
    /// it; otherwise it starts over, told that an earlier attempt may have
    /// left changes. A queued follow-up has no run of its own: it uses the
    /// conversation's directory and worktree, continues that agent session
    /// only after the interrupted job the run served has been resumed (and
    /// then behind that conversation), and otherwise starts in a new
    /// worktree of the checkout that job recorded. Each job is resumed
    /// once, and not while an agent process of its run still goes. Returns
    /// `<host>/<new job>`.
    pub async fn resume(&self, host: &str, session: &str, job: &str) -> Result<String, String> {
        self.resume_locked(host, session, job, None, None).await
    }

    /// [`Self::resume`] with the owner's `prompt` in place of the note that
    /// the attempt was cut off. A finished job (completed, failed, declined,
    /// or timed out) is continued this way too, still through the ledger's
    /// resume link, and still not a reviewed job. `resume` itself is unchanged:
    /// only an interrupted job, and only with that note.
    pub async fn resume_with(
        &self,
        host: &str,
        session: &str,
        job: &str,
        prompt: &str,
        accepted_follow_up: Option<String>,
    ) -> Result<String, String> {
        self.resume_locked(host, session, job, Some(prompt), accepted_follow_up)
            .await
    }

    async fn resume_locked(
        &self,
        host: &str,
        session: &str,
        job: &str,
        follow_up: Option<&str>,
        accepted_follow_up: Option<String>,
    ) -> Result<String, String> {
        let key = format!("{session}/{job}");
        if !lock(&self.resuming).insert(key.clone()) {
            return Err(format!("{job} is being resumed already"));
        }
        // Released however this ends, a dropped request included.
        let _claim = Claim {
            claims: &self.resuming,
            key: &key,
        };
        self.resume_once(host, session, job, &key, follow_up, accepted_follow_up)
            .await
    }

    async fn resume_once(
        &self,
        host: &str,
        session: &str,
        job: &str,
        key: &str,
        follow_up: Option<&str>,
        accepted_follow_up: Option<String>,
    ) -> Result<String, String> {
        let record = ledger::load(&self.jobs, session, job)
            .ok_or_else(|| format!("no record of {session}/{job}"))?;
        let session_live = lock(&self.inner).sessions.contains_key(session);
        let interrupted = record.state == AgentState::Interrupted
            || (!record.state.is_final()
                && (!record.owner.is_alive()
                    || (record.owner.pid == std::process::id() && !session_live)));
        // A subscribed thread's later message continues a finished job too.
        // `resume` does not: it passes no follow-up prompt.
        let ended = matches!(
            record.state,
            AgentState::Completed | AgentState::Failed | AgentState::Declined | AgentState::Timeout
        );
        let resumable = interrupted || (follow_up.is_some() && ended && !record.reviewed);
        if !resumable {
            return Err(format!(
                "{job} was not interrupted ({}); only an interrupted job can be resumed",
                record.state
            ));
        }
        if record.reviewed {
            return Err(
                "a reviewed job is not resumed: its journal is its record. Start a new \
                 reviewed call whose brief restates the open findings and anything already \
                 landed"
                    .into(),
            );
        }
        if let Some(resumed) = &record.resumed_as {
            return Err(format!("{job} was already resumed as {resumed}"));
        }
        let (agent, _) = self.agent(host)?;
        // Reading every record, run state, and the process table takes a
        // while.
        let (successor, run, live, blocking) = {
            let (jobs, runs) = (self.jobs.clone(), self.runs.clone());
            let delegations = Arc::clone(&self.delegations);
            let continued = record
                .arguments
                .get("session")
                .and_then(Value::as_str)
                .map(str::to_owned);
            let (session, job, key) = (session.to_owned(), job.to_owned(), key.to_owned());
            tokio::task::spawn_blocking(move || {
                // A job that names it as what it resumes settles it, should
                // SCV have stopped before this record said so, even once
                // the owner dismissed that job.
                let successor = ledger::list_all(&jobs)
                    .into_iter()
                    .find(|record| record.resumed_from.as_deref() == Some(key.as_str()))
                    .map(|record| format!("{}/{}", record.session, record.job));
                let states = activity::list_states(&runs);
                let served = |state: &activity::RunState| {
                    state.session == session
                        && (state.job.as_deref() == Some(job.as_str()) || state.jobs.contains(&job))
                };
                let own = states
                    .iter()
                    .filter(|state| served(state))
                    .max_by_key(|state| state.started_unix)
                    .cloned();
                // A queued follow-up has no run of its own. The conversation
                // it continues does, and that run has the directory and the
                // agent's session.
                let fallback = if own.is_none() {
                    continued.as_deref().and_then(|handle| {
                        states
                            .iter()
                            .filter(|state| {
                                state.session == session
                                    && state.conversation.as_deref() == Some(handle)
                            })
                            .max_by_key(|state| state.started_unix)
                            .cloned()
                    })
                } else {
                    None
                };
                let run = own.or_else(|| fallback.clone());
                // Adopting while the job that run served is still interrupted
                // would start a second process in its agent session. The
                // follow-up waits until that job is resumed, then queues
                // behind the conversation that resume adopted.
                let blocking = match &fallback {
                    Some(found) if found.vendor.is_some() && found.cwd.is_dir() => {
                        unresumed_served(&jobs, found, &job)
                    }
                    _ => Vec::new(),
                };
                let live = run.as_ref().is_some_and(|run| {
                    delegations
                        .list(true)
                        .iter()
                        .any(|entry| entry.record.handle == run.handle && entry.processes > 0)
                });
                (successor, run, live, blocking)
            })
            .await
            .map_err(|error| error.to_string())?
        };
        if let Some(resumed) = successor {
            return Err(format!("{job} was already resumed as {resumed}"));
        }
        if live {
            return Err(format!(
                "{job}'s agent ({}) still runs; SCV stops what a stopped SCV process left \
                 within a minute (or `scv agents kill` it now). Resume it after that",
                run.as_ref().map_or("", |run| run.handle.as_str())
            ));
        }
        if !blocking.is_empty() {
            let ahead = if blocking.len() == 1 { "it" } else { "them" };
            return Err(format!(
                "{job} was queued behind {}; resume {ahead} first. This follow-up then waits \
                 behind the same conversation, so one agent session is not continued twice",
                blocking.join(", ")
            ));
        }
        let mut record = record;
        let cwd_live = run.as_ref().is_some_and(|found| found.cwd.is_dir());
        if let Some(found) = &run
            && cwd_live
        {
            remember_cwd(&mut record, &found.cwd);
            if record.worktree.is_none()
                && let Some(worktree) =
                    crate::delegate::worktree::worktree_ref(&self.worktrees, &found.cwd)
            {
                record.worktree = Some(worktree);
            }
        }
        // The directory is gone (the owner removed the worktree; an
        // interrupted job does not hold it). The job that run served still
        // records the checkout, dismissed records included, which is what a
        // new worktree is made from. The run's path still names the
        // subdirectory inside it.
        if record.worktree.is_none()
            && let Some(found) = &run
            && let Some(worktree) = served_worktree(&self.jobs, found)
        {
            record.worktree = Some(worktree);
        }
        if !cwd_live
            && let Some(found) = &run
            && let Some(worktree) = &record.worktree
            && found.cwd.starts_with(Path::new(&worktree.path))
        {
            remember_cwd(&mut record, &found.cwd);
        }
        let cwd = run
            .as_ref()
            .map(|run| run.cwd.clone())
            .filter(|cwd| cwd.is_dir());
        let adopted = run
            .as_ref()
            .and_then(|run| run.vendor.as_deref().zip(cwd.as_deref()))
            .and_then(|(vendor, cwd)| agent.adopt(&record.agent, cwd, vendor));
        let mut arguments = if let Some(handle) = &adopted {
            let prompt = match follow_up {
                Some(text) => format!(
                    "[SCV] The owner sent a follow-up on the subscribed thread. Continue the same \
                     task and keep its constraints. If it is not this job's work, hand it back \
                     with `scv jobs hand-back` and do not act on it. The task was:\n\n{}\n\n\
                     Follow-up:\n\n{text}",
                    record.prompt
                ),
                None => format!(
                    "[SCV] SCV stopped before your last turn on this task finished (a restart \
                     or a crash), so it was cut off. Check where the work stands (files, git \
                     status, anything still running), then carry on. The task was:\n\n{}",
                    record.prompt
                ),
            };
            json!({"session": handle, "prompt": prompt})
        } else if follow_up.is_some() {
            // The agent's own conversation could not be resumed. A follow-up
            // must not start a fresh one that knows only the new message.
            return Err(UNRESUMABLE_FOLLOW_UP.into());
        } else {
            rerun_arguments(&record, |option| agent.takes(&record.agent, option))
        };
        if let Some(object) = arguments.as_object_mut() {
            for key in ["timeout_seconds", "no_timeout"] {
                if let Some(value) = record.arguments.get(key) {
                    object.insert(key.into(), value.clone());
                }
            }
        }
        // The new job's record says what it resumes from its first write.
        // A follow-up into a session with no report turn is marked reported
        // when it finishes. One into a chat session stays for that session's
        // report turn.
        let mut meta = if follow_up.is_some() {
            self.follow_up_meta(host, Some(key.to_owned()), accepted_follow_up)
        } else {
            JobMeta {
                origin: scv_protocol::JobOrigin::Owner,
                resumed_from: Some(key.to_owned()),
                accepted_follow_up,
                ..JobMeta::default()
            }
        };
        meta.worktree = record
            .worktree
            .clone()
            .filter(|worktree| Path::new(&worktree.path).is_dir());
        let resumed = self.start_with(host, arguments, meta).await?;
        ledger::update(&self.jobs, session, job, |old| {
            old.resumed_as = Some(resumed.clone());
            if !old.state.is_final() {
                ledger::interrupt(old, "its session ended before it finished");
            }
        });
        Ok(resumed)
    }

    /// Stop job `job` of live session `session`.
    pub fn cancel(&self, session: &str, job: &str) -> Result<String, String> {
        let jobs = {
            let inner = lock(&self.inner);
            inner
                .sessions
                .get(session)
                .and_then(|slot| slot.jobs.as_ref())
                .and_then(Weak::upgrade)
        };
        let Some(jobs) = jobs else {
            return Err(format!(
                "session {session} does not run in this daemon; a job whose session ended \
                 stopped with it, and `scv agents kill` stops another process's runs"
            ));
        };
        jobs.cancel_for_owner(job)?;
        Ok(format!(
            "Stopping {job}: its agent and every process it started are stopped; work it \
             already wrote stays."
        ))
    }

    /// Forget finished or interrupted job `job` of `session`.
    pub fn dismiss(&self, session: &str, job: &str) -> Result<String, String> {
        let record = ledger::load(&self.jobs, session, job)
            .ok_or_else(|| format!("no record of {session}/{job}"))?;
        let jobs = lock(&self.inner)
            .sessions
            .get(session)
            .and_then(|slot| slot.jobs.as_ref())
            .and_then(Weak::upgrade);
        let running = jobs.as_ref().is_some_and(|jobs| {
            jobs.live()
                .iter()
                .any(|live| live.id == job && live.running)
        });
        if running || (!record.state.is_final() && record.owner.is_alive()) {
            return Err(format!("{job} is still running; stop it first"));
        }
        // A session that has yet to report the result writes its record
        // again when it does: this daemon's says so itself, and one of
        // another live SCV process by its record.
        let elsewhere =
            record.owner.is_alive() && ProcessIdentity::current().as_ref() != Some(&record.owner);
        if jobs.is_some_and(|jobs| jobs.unreported(job)) || (elsewhere && !record.reported) {
            return Err(format!(
                "{job}'s result has not reached its session yet; dismiss it once it has"
            ));
        }
        ledger::dismiss(&self.jobs, session, job).map_err(|error| error.to_string())?;
        Ok(format!("Forgot {session}/{job}."))
    }

    /// The ledger record of `job` in `session`.
    pub fn job_record(&self, session: &str, job: &str) -> Option<JobRecord> {
        ledger::load(&self.jobs, session, job)
    }

    /// The conversation a follow-up can be given to. A job's record names it
    /// once the job has finished; until then the run serving the job does,
    /// which is what a job on its first turn has.
    pub fn job_conversation(&self, session: &str, job: &str) -> Option<String> {
        let record = ledger::load_including_dismissed(&self.jobs, session, job)?;
        if let Some(conversation) = record.conversation.clone().or_else(|| {
            record
                .arguments
                .get("session")
                .and_then(Value::as_str)
                .map(str::to_owned)
        }) {
            return Some(conversation);
        }
        activity::list_states(&self.runs)
            .into_iter()
            .find_map(|state| {
                let serves = state.session == session
                    && (state.job.as_deref() == Some(job)
                        || state.jobs.iter().any(|name| name == job));
                serves.then_some(state.conversation).flatten()
            })
    }

    /// The background job run `handle` is serving, when its state names one.
    pub fn job_for_run(&self, handle: &str) -> Option<(String, String)> {
        activity::list_states(&self.runs)
            .into_iter()
            .find(|state| state.handle == handle)
            .and_then(|state| state.job.map(|job| (state.session, job)))
    }

    /// Job `job` of `session` as the workbench lists it, with its reply or
    /// error once it has ended, each cut to [`MAX_JOB_RESULT_BYTES`].
    pub fn job_result(&self, session: &str, job: &str) -> Result<JobResult, String> {
        let record = ledger::load(&self.jobs, session, job)
            .ok_or_else(|| format!("no record of {session}/{job}"))?;
        let (jobs, awaiting, session_live) = {
            let inner = lock(&self.inner);
            let slot = inner.sessions.get(session);
            (
                slot.and_then(|slot| slot.jobs.as_ref())
                    .and_then(Weak::upgrade),
                inner
                    .approvals
                    .values()
                    .find(|pending| pending.info.session == session && pending.info.job == job)
                    .map(|pending| pending.info.clone()),
                slot.is_some(),
            )
        };
        let live = jobs
            .as_ref()
            .and_then(|jobs| jobs.live().into_iter().find(|live| live.id == job));
        // The run that served it last, chosen as the snapshot chooses it.
        let run = activity::list_states(&self.runs)
            .into_iter()
            .filter(|state| {
                state.session == session
                    && (state.job.as_deref() == Some(job)
                        || state.jobs.iter().any(|served| served == job))
            })
            .min_by_key(|state| std::cmp::Reverse((state.last_activity_unix, state.started_unix)));
        let (state, detail) = job_state(
            &record,
            live.as_ref(),
            run.as_ref(),
            awaiting.as_ref(),
            session_live,
            unix_now(),
        );
        let mut truncated = false;
        let mut cut = |text: Option<String>| {
            text.map(|text| {
                let kept = scv_client::text::utf8_prefix(&text, MAX_JOB_RESULT_BYTES);
                truncated |= kept.len() < text.len();
                kept.to_owned()
            })
        };
        let reply = cut(record.reply);
        let error = cut(record.error);
        Ok(JobResult {
            session: record.session,
            job: record.job,
            agent: record.agent,
            state,
            detail,
            reply,
            error,
            truncated,
            outcome: jobs.and_then(|jobs| jobs.outcome(job)),
            finished_unix_seconds: record.finished_unix,
        })
    }

    /// Take conversation `conversation` of live session `session` over in
    /// its agent's own terminal for the owner, or attach to the takeover of
    /// it already running, and return the whole screen. A second client's
    /// attach takes the terminal from the first, whose hold stops working.
    pub async fn attach_terminal(
        &self,
        session: &str,
        conversation: &str,
        cols: Option<u16>,
        rows: Option<u16>,
    ) -> Result<AttachPage, String> {
        self.take_terminal(session, conversation, cols, rows, false, None)
            .await
            .map(|(page, _)| page)
    }

    /// [`Self::attach_terminal`], stopping first the background job that
    /// runs the conversation's turn. The conversation's lane is held for
    /// the takeover before that job is stopped, so no other prompt starts a
    /// turn in between, and nothing is stopped when it cannot be held: when
    /// the agent's terminal is not offered, a takeover of it runs or is
    /// starting, a reviewed job holds it, it has no agent session yet, the
    /// running turn is no background job, that job waits for the owner's
    /// approval (which a takeover never answers or withdraws), or as many
    /// takeovers as may run do. Given the `job` the owner was told it stops,
    /// nothing is stopped either when another job runs the turn by then. A
    /// conversation running no turn is taken over as by `attach_terminal`.
    /// Also returns SCV's note naming the job it stopped.
    pub async fn interrupt_and_attach(
        &self,
        session: &str,
        conversation: &str,
        cols: Option<u16>,
        rows: Option<u16>,
        job: Option<&str>,
    ) -> Result<(AttachPage, Option<String>), String> {
        self.take_terminal(session, conversation, cols, rows, true, job)
            .await
    }

    async fn take_terminal(
        &self,
        session: &str,
        conversation: &str,
        cols: Option<u16>,
        rows: Option<u16>,
        interrupt: bool,
        named: Option<&str>,
    ) -> Result<(AttachPage, Option<String>), String> {
        self.expire_takeovers();
        let key = format!("{session}/{conversation}");
        let id = format!("at-{}", &uuid::Uuid::new_v4().simple().to_string()[..16]);
        if let Some(page) = self.attach_running(&key, &id, cols, rows) {
            return Ok((page, None));
        }
        let _starting = self.attaching.lock().await;
        // It may have started while this one waited.
        if let Some(page) = self.attach_running(&key, &id, cols, rows) {
            return Ok((page, None));
        }
        if self.closing.load(Ordering::Acquire) {
            return Err(stopping());
        }
        if self.running_takeovers() >= MAX_TAKEOVERS {
            return Err(at_cap());
        }
        let (agent, _) = self.agent(session)?;
        let size = takeover::clamp_size(cols, rows);
        let (takeover, stopped) = if interrupt {
            // Held before anything is stopped; refused, nothing is.
            let reserved = agent.reserve_takeover(conversation)?;
            if let (Some(job), Some(named)) = (&reserved.job, named)
                && job != named
            {
                return Err(not_the_named_job(conversation, named, job));
            }
            if let Some(job) = &reserved.job
                && let Some(approval) = lock(&self.inner)
                    .approvals
                    .values()
                    .find(|pending| pending.info.session == session && &pending.info.job == job)
            {
                return Err(awaits_approval(job, &approval.info.id));
            }
            let stopped = reserved.job.clone();
            (
                agent.take_over_reserved(reserved, session, size).await?,
                stopped,
            )
        } else {
            (agent.take_over(conversation, session, size).await?, None)
        };
        let note = stopped.map(|job| format!("Stopped {job} and took {conversation} over."));
        let page = takeover.page(&id, 0);
        {
            let mut takeovers = lock(&self.takeovers);
            if !self.closing.load(Ordering::Acquire) {
                takeovers.insert(
                    key,
                    Held {
                        takeover,
                        lease: Some((id, Instant::now())),
                        closed: None,
                    },
                );
                return Ok((page, note));
            }
        }
        // SCV began stopping while it started: it ends like the others.
        takeover.end().await;
        Err(stopping())
    }

    /// Attach client `id` to the takeover of `key` if it runs, sized `cols`
    /// by `rows`, and return the whole screen; forget one that closed.
    fn attach_running(
        &self,
        key: &str,
        id: &str,
        cols: Option<u16>,
        rows: Option<u16>,
    ) -> Option<AttachPage> {
        let (takeover, watched) = {
            let mut takeovers = lock(&self.takeovers);
            match takeovers.get_mut(key) {
                Some(held) if !held.takeover.is_finished() => {
                    let watched = held.lease.is_some();
                    held.lease = Some((id.to_owned(), Instant::now()));
                    (Arc::clone(&held.takeover), watched)
                }
                Some(_) => {
                    takeovers.remove(key);
                    return None;
                }
                None => return None,
            }
        };
        if !watched {
            takeover.note_attached(true);
        }
        takeover.resize(cols, rows);
        Some(takeover.page(id, 0))
    }

    /// Type `input` into the terminal client `id` is attached to, resize it,
    /// and return what changed on its screen since generation `after`.
    pub async fn attach_io(
        &self,
        id: &str,
        after: u64,
        input: &str,
        cols: Option<u16>,
        rows: Option<u16>,
    ) -> Result<AttachPage, String> {
        if input.len() > takeover::MAX_INPUT_BYTES {
            return Err(format!(
                "at most {} KiB can be typed at once",
                takeover::MAX_INPUT_BYTES / 1024
            ));
        }
        self.expire_takeovers();
        let takeover = self.leased(id)?;
        let note = takeover.input(input.as_bytes()).await;
        takeover.resize(cols, rows);
        let mut page = takeover.page(id, after);
        page.note = note;
        Ok(page)
    }

    /// Stop client `id` watching its terminal; the agent keeps running.
    pub fn detach(&self, id: &str) -> Result<String, String> {
        let takeover = {
            let mut takeovers = lock(&self.takeovers);
            let held = takeovers
                .values_mut()
                .find(|held| held.lease.as_ref().is_some_and(|(lease, _)| lease == id))
                .ok_or_else(not_attached)?;
            held.lease = None;
            Arc::clone(&held.takeover)
        };
        takeover.note_attached(false);
        Ok(if takeover.is_running() {
            format!(
                "Detached from {}; it keeps running in its terminal.",
                takeover.conversation
            )
        } else {
            format!(
                "Detached from {}, whose terminal has closed.",
                takeover.conversation
            )
        })
    }

    /// End the takeover of `conversation` of `session`: hang its terminal
    /// up, stop its process, and give the conversation back to SCV's
    /// turns. Returns SCV's note and the final screen.
    pub async fn end_takeover(
        &self,
        session: &str,
        conversation: &str,
    ) -> Result<(String, AttachPage), String> {
        let key = format!("{session}/{conversation}");
        let held = lock(&self.takeovers)
            .remove(&key)
            .ok_or_else(|| format!("{conversation} of {session} is not taken over"))?;
        held.takeover.end().await;
        let id = held.lease.map(|(id, _)| id).unwrap_or_default();
        let page = held.takeover.page(&id, 0);
        let how = page.exit.clone().unwrap_or_else(|| "closed".into());
        Ok((
            format!("Ended the takeover of {conversation}: the terminal closed ({how})."),
            page,
        ))
    }

    /// End every takeover, waiting for each: the daemon is stopping, so
    /// none starts any more.
    pub async fn end_takeovers(&self) {
        let held: Vec<Held> = {
            let mut takeovers = lock(&self.takeovers);
            self.closing.store(true, Ordering::Release);
            takeovers.drain().map(|(_, held)| held).collect()
        };
        futures_util::future::join_all(held.iter().map(|held| held.takeover.end())).await;
    }

    /// How many takeovers run.
    fn running_takeovers(&self) -> usize {
        lock(&self.takeovers)
            .values()
            .filter(|held| !held.takeover.is_finished())
            .count()
    }

    /// Sessions with a takeover still running, which count as in use.
    pub fn takeover_sessions(&self) -> HashSet<String> {
        lock(&self.takeovers)
            .values()
            .filter(|held| !held.takeover.is_finished())
            .map(|held| held.takeover.session.clone())
            .collect()
    }

    /// End the takeovers of `session` (every one with `None`) in the
    /// background: their session is gone.
    fn end_takeovers_of(&self, session: Option<&str>) {
        let ending: Vec<Arc<Takeover>> = {
            let mut takeovers = lock(&self.takeovers);
            let keys: Vec<String> = takeovers
                .iter()
                .filter(|(_, held)| session.is_none_or(|session| held.takeover.session == session))
                .map(|(key, _)| key.clone())
                .collect();
            keys.iter()
                .filter_map(|key| takeovers.remove(key))
                .map(|held| held.takeover)
                .collect()
        };
        if ending.is_empty() {
            return;
        }
        match tokio::runtime::Handle::try_current() {
            Ok(runtime) => {
                runtime.spawn(async move {
                    futures_util::future::join_all(ending.iter().map(|takeover| takeover.end()))
                        .await;
                });
            }
            Err(_) => ending.iter().for_each(|takeover| takeover.kill_now()),
        }
    }

    /// The running or recently closed takeover client `id` holds.
    fn leased(&self, id: &str) -> Result<Arc<Takeover>, String> {
        let mut takeovers = lock(&self.takeovers);
        let held = takeovers
            .values_mut()
            .find(|held| held.lease.as_ref().is_some_and(|(lease, _)| lease == id))
            .ok_or_else(not_attached)?;
        if let Some((_, read)) = &mut held.lease {
            *read = Instant::now();
        }
        Ok(Arc::clone(&held.takeover))
    }

    /// Detach clients that stopped reading, and forget takeovers whose
    /// terminal closed a while ago.
    fn expire_takeovers(&self) {
        let now = Instant::now();
        let mut detached = Vec::new();
        let mut takeovers = lock(&self.takeovers);
        takeovers.retain(|_, held| {
            if held.takeover.is_finished() {
                let closed = *held.closed.get_or_insert(now);
                return now.duration_since(closed) < CLOSED_KEPT;
            }
            if held
                .lease
                .as_ref()
                .is_some_and(|(_, read)| now.duration_since(*read) >= ATTACH_LEASE)
            {
                held.lease = None;
                detached.push(Arc::clone(&held.takeover));
            }
            true
        });
        drop(takeovers);
        for takeover in detached {
            takeover.note_attached(false);
        }
    }

    /// Settle what processes that stopped left unfinished, and remove old
    /// records: the daemon runs this at startup and every minute.
    pub fn maintain(&self) -> Maintenance {
        self.expire_takeovers();
        let interrupted_jobs = ledger::settle_orphans(&self.jobs)
            .into_iter()
            .map(|record| format!("{}/{}", record.session, record.job))
            .collect();
        let interrupted_runs = activity::settle_orphans(&self.runs);
        // Read only: the daemon's subscriptions change under their own lock,
        // which this does not hold.
        let retain = crate::delegate::background::subscription::pinned_jobs_in(
            &self.subscriptions,
            &self.jobs,
        );
        let pruned = ledger::prune_keeping(&self.jobs, KEEP, &retain)
            + activity::prune(&self.runs, KEEP, MAX_RUNS);
        Maintenance {
            interrupted_jobs,
            interrupted_runs,
            pruned,
        }
    }
}

fn stopping() -> String {
    "SCV is stopping, so it takes no conversation over now".into()
}

fn at_cap() -> String {
    format!("{MAX_TAKEOVERS} conversations are taken over already; end one first")
}

/// The refusal to stop `job`, which waits for the owner's answer to
/// approval request `id`, for a takeover.
fn awaits_approval(job: &str, id: &str) -> String {
    format!(
        "{job} waits for your approval ({id}); answer it first (`scv jobs approve` or `deny`). \
         Taking its conversation over would withdraw the request, and a takeover never answers \
         or withdraws an approval"
    )
}

/// The refusal to stop `running`, which runs conversation `conversation`'s
/// turn, for a takeover the owner was told stops `named`.
fn not_the_named_job(conversation: &str, named: &str, running: &str) -> String {
    format!(
        "{conversation} is running {running}'s turn now, not {named}'s, so nothing was stopped; \
         attach again to stop {running}"
    )
}

/// `offer` as `attach` would answer it now, with what only the workbench
/// knows: with `running` takeovers at the cap, none starts, and stopping a
/// job that waits for the owner's answer to `approval` is refused.
fn settle_offer(mut offer: AttachOffer, approval: Option<&str>, running: usize) -> AttachOffer {
    if offer.interrupt
        && let (Some(job), Some(id)) = (&offer.job, approval)
    {
        offer.detail = Some(awaits_approval(job, id));
        offer.interrupt = false;
    }
    let starts = offer.state == AttachState::Ready || offer.interrupt;
    if starts && running >= MAX_TAKEOVERS {
        offer.state = AttachState::Full;
        offer.detail = Some(at_cap());
        offer.interrupt = false;
    }
    offer
}

fn not_attached() -> String {
    "this terminal is not attached here any more (another dashboard attached to it, or its \
     takeover ended); attach again"
        .into()
}

/// A job claimed for resuming, released when dropped.
struct Claim<'a> {
    claims: &'a Mutex<HashSet<String>>,
    key: &'a str,
}

impl Drop for Claim<'_> {
    fn drop(&mut self) {
        lock(self.claims).remove(self.key);
    }
}

struct SlotView {
    id: String,
    label: String,
    workspace: PathBuf,
    state: AgentState,
    detail: Option<String>,
    agent: Option<Arc<BackgroundCapable>>,
    jobs: Option<Arc<BackgroundJobs>>,
}

/// What a job is doing, from its record, its live entry when its session
/// runs here, the run serving it, and an approval it waits on.
fn job_state(
    record: &JobRecord,
    live: Option<&LiveJob>,
    run: Option<&RunState>,
    awaiting: Option<&WorkbenchApproval>,
    session_live: bool,
    now: u64,
) -> (AgentState, Option<String>) {
    if record.state.is_final() {
        let detail = match record.state {
            AgentState::Interrupted => record.interrupted.clone(),
            AgentState::Completed => None,
            _ => record
                .error
                .as_deref()
                .and_then(|error| error.lines().find(|line| !line.trim().is_empty()))
                .map(|line| bounded(line, 200)),
        };
        return (record.state, detail);
    }
    if let Some(approval) = awaiting {
        return (
            AgentState::AwaitingInput,
            Some(format!(
                "waiting for your approval: {}",
                bounded(&approval.summary, 200)
            )),
        );
    }
    if let Some(live) = live {
        if live.waiting {
            let behind = record
                .arguments
                .get("session")
                .and_then(Value::as_str)
                .map_or_else(String::new, |handle| format!(" {handle}'s"));
            return (
                AgentState::Queued,
                Some(format!("waits for conversation{behind} running turn")),
            );
        }
        if live.running {
            if let Some(run) = run.filter(|run| !run.ended) {
                return activity::current(run, now);
            }
            return (
                AgentState::Working,
                live.progress
                    .clone()
                    .or_else(|| Some(format!("running for {} s", live.elapsed.as_secs()))),
            );
        }
    }
    if !record.owner.is_alive() {
        return (
            AgentState::Interrupted,
            Some("the SCV process that ran it stopped".into()),
        );
    }
    if record.owner.pid == std::process::id() && !session_live {
        return (
            AgentState::Interrupted,
            Some("its session ended before it finished".into()),
        );
    }
    match run {
        Some(run) if !run.ended => activity::current(run, now),
        _ => (record.state, None),
    }
}

/// The workbench's own session ID for `workspace`: stable, so the owner's
/// jobs in one workspace group under one session across daemon restarts.
pub fn host_session(workspace: &Path) -> String {
    use sha2::{Digest, Sha256};
    let digest = Sha256::digest(workspace.as_os_str().as_encoded_bytes());
    let hex: String = digest[..8]
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect();
    format!("workbench-{hex}")
}

/// `<session>/<job>` of a job an `agent` call started.
fn job_name(session: &str, started: &Value) -> String {
    format!(
        "{session}/{}",
        started.get("job").and_then(Value::as_str).unwrap_or("job")
    )
}

/// Decides a background job's approval request: the session's unattended
/// answer, or, for one it would deny while the owner watches, the owner's.
struct AttendedGate {
    workbench: Weak<Workbench>,
    session: String,
    unattended: Arc<dyn ApprovalGate>,
}

#[async_trait]
impl ApprovalGate for AttendedGate {
    async fn approve(
        &self,
        request: ApprovalRequest,
        cancellation: CancellationToken,
    ) -> Result<bool, AgentError> {
        if self
            .unattended
            .approve(request.clone(), cancellation.clone())
            .await?
        {
            return Ok(true);
        }
        match self.workbench.upgrade() {
            Some(workbench) if workbench.watched() => {
                workbench.ask(&self.session, request, &cancellation).await
            }
            _ => Ok(false),
        }
    }
}

fn bounded(text: &str, max: usize) -> String {
    if text.chars().count() <= max {
        return text.to_owned();
    }
    let mut cut: String = text.chars().take(max.saturating_sub(1)).collect();
    cut.push('…');
    cut
}

fn unix_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_secs())
}

/// Jobs whose turns `run` served, the latest one first.
fn served_job_names(run: &RunState) -> Vec<&str> {
    let mut names = Vec::new();
    if let Some(job) = run.job.as_deref() {
        names.push(job);
    }
    for job in &run.jobs {
        if !names.contains(&job.as_str()) {
            names.push(job.as_str());
        }
    }
    names
}

/// Interrupted jobs `run` served that the owner has not resumed, as
/// `session/job`. `skip` is the follow-up being resumed.
fn unresumed_served(root: &Path, run: &RunState, skip: &str) -> Vec<String> {
    served_job_names(run)
        .into_iter()
        .filter(|name| *name != skip)
        .filter_map(|name| {
            let record = ledger::load(root, &run.session, name)?;
            record
                .resumable()
                .then(|| format!("{}/{}", run.session, name))
        })
        .collect()
}

/// The worktree recorded on a job `run` served. A follow-up has none of
/// its own, and the run's directory may already have been removed. A
/// dismissed record still carries the worktree.
fn served_worktree(root: &Path, run: &RunState) -> Option<scv_protocol::WorktreeRef> {
    served_job_names(run).into_iter().find_map(|name| {
        ledger::load_including_dismissed(root, &run.session, name)
            .and_then(|record| record.worktree)
    })
}

/// Record `cwd` on `record` when the call did not name one.
fn remember_cwd(record: &mut ledger::JobRecord, cwd: &Path) {
    let has_cwd = record
        .arguments
        .get("cwd")
        .and_then(Value::as_str)
        .is_some_and(|cwd| !cwd.is_empty());
    if !has_cwd && let Some(object) = record.arguments.as_object_mut() {
        object.insert("cwd".into(), cwd.display().to_string().into());
    }
}

/// The `agent` call that runs interrupted job `record` again from its
/// prompt in a new conversation: its options but those the host agent does
/// not `take`, and a note that an earlier attempt may have left changes. A
/// job that ran in a worktree runs in the same directory of it while the
/// worktree is there, and in a new worktree of its repository once it was
/// removed, so its work never lands in the repository's own checkout.
fn rerun_arguments(record: &JobRecord, takes: impl Fn(&str) -> bool) -> Value {
    let mut arguments = record.arguments.clone();
    if let Some(object) = arguments.as_object_mut() {
        object.remove("session");
        object.remove("worktree");
        object.remove("on_busy");
        // What this session's agent does not take is left to its own
        // defaults.
        for option in ["model", "effort"] {
            if !takes(option) {
                object.remove(option);
            }
        }
        object.insert("agent".into(), record.agent.clone().into());
        object.insert(
            "prompt".into(),
            format!(
                "[SCV] An earlier attempt at this task was cut off when SCV stopped (a restart \
                 or a crash), and its conversation could not be resumed. It may have left \
                 partial changes here: check them first, then do the task:\n\n{}",
                record.prompt
            )
            .into(),
        );
        if let Some(worktree) = &record.worktree {
            let top = Path::new(&worktree.path);
            // The call's directory inside the worktree, as it was recorded.
            let inside = object
                .get("cwd")
                .and_then(Value::as_str)
                .and_then(|cwd| Path::new(cwd).strip_prefix(top).ok())
                .map(Path::to_path_buf)
                .unwrap_or_default();
            if top.is_dir() {
                let cwd = Some(top.join(&inside))
                    .filter(|cwd| cwd.is_dir())
                    .unwrap_or_else(|| top.to_owned());
                object.insert("cwd".into(), cwd.display().to_string().into());
            } else {
                let repo = Path::new(&worktree.repo);
                let cwd = Some(repo.join(&inside))
                    .filter(|cwd| cwd.is_dir())
                    .unwrap_or_else(|| repo.to_owned());
                object.insert("cwd".into(), cwd.display().to_string().into());
                object.insert("worktree".into(), true.into());
            }
        }
    }
    arguments
}

#[cfg(test)]
mod tests;