magi/

chat.rs

//! The browser interview: `magi plan` for somebody holding a phone.
//!
//! [`crate::plan`] is an interview that works by *handing over the terminal* -
//! stdin, stdout and stderr inherited, the agent's own UI in front of the
//! operator, no timeout. That is the right design and it is not changing. It is
//! also unavailable to the operator who is away from the machine, which is most
//! of the time this repository's operator wants to plan something: there is no
//! terminal in a browser to hand over.
//!
//! So this module is the same interview, arrived at from the other side. magi
//! does host the conversation here, because there is nothing else that can:
//! each operator message is one *headless* [`crate::agent::invoke`], and the
//! transcript lives in a JSON file the phone reads. The end state is identical
//! to `magi plan`'s - a task file checked by [`plan::review_draft`] and filed
//! in [`crate::queue`] - which is deliberate. Two planning paths that accept
//! different task files would be two products.
//!
//! # A turn is cheap because the CLI remembers
//!
//! The thing that makes a turn-per-request affordable is [`SeatState`]: a
//! second [`crate::agent::invoke`] with the same seat resumes the CLI's own
//! conversation (`claude --resume`, `opencode run -s`, `agy --conversation`),
//! so a turn sends the operator's new sentence and nothing else. The model
//! already has the repository it read and the questions it asked. magi does
//! *not* re-send the transcript when the CLI can resume - that would pay for
//! the whole conversation again on every message, and it would let magi's idea
//! of the history drift from the model's. [`transcript`] exists only for the
//! case where resuming is genuinely impossible, and [`turn`] says when.
//!
//! # Shape
//!
//! The same split as [`crate::queue`] and [`crate::ask`]: [`Chat`] is data plus
//! pure helpers, [`Chats`] owns all I/O and is constructed with its root, so
//! every test below drives a real store in a temp directory and none of them
//! touch the operator's home. One conversation is one JSON file, written
//! atomically, because `magi web` and a future `magi chat` are separate
//! processes and a rename is the only cross-process atomic write that needs no
//! coordination between them.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;

use anyhow::{Context, Result, bail};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};

use crate::agent::{self, Invocation, SeatState};
use crate::config::Config;
use crate::plan;
use crate::queue::{self, Queue, Source, Task};

/// On-disk format for a conversation. Bumped when a field's meaning changes.
///
/// The web UI is written against this shape by hand, so a field that changes
/// meaning without a bump here is a front end that lies silently.
pub const SCHEMA: u32 = 1;

/// Wall-clock limit for one agent turn. See [`crate::config::Graph::timeout_chat`].
///
/// An hour by default, not the five minutes this used to be. The short
/// timeout assumed an operator holding a phone with a spinner on it, who
/// needed to be told a turn was wedged while they were still looking at the
/// screen. That is no longer how this gets used: the operator starts another
/// conversation while this one thinks and comes back to it later, so the
/// time a turn takes is no longer time spent waiting - it is only a seat
/// held, which is cheap. What still needs a bound is a genuinely wedged CLI,
/// and an hour is generous enough to not be that.
fn turn_timeout(cfg: &Config) -> Duration {
    Duration::from_secs(cfg.graph.timeout_chat)
}

/// Seat name for the interviewing agent.
///
/// One seat per conversation, so the CLI-side conversation is scoped to this
/// chat and nothing else - the same rule [`crate::agent`] applies to judges.
const SEAT: &str = "plan";

/// Prefix on an agent turn that magi wrote rather than an agent.
///
/// A failed turn has to be *visible*, and the transcript is the only surface
/// the phone renders, so the failure goes in as an agent turn carrying this
/// marker. Two turn authors is what the wire shape allows (`operator` /
/// `agent`), and inventing a third would break every client written against
/// it; a stable prefix the UI can key on costs nothing and loses no
/// information.
pub const MAGI_NOTE: &str = "magi: ";

/// Who said something.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Who {
    /// The person magi is planning for.
    Operator,
    /// The interviewing agent - or magi itself, reporting that the agent
    /// failed. See [`MAGI_NOTE`].
    Agent,
}

/// One image the operator attached to a turn.
///
/// Never carries the bytes themselves: the picture lives on disk under
/// [`Chats::attachments_dir`], named by `id` alone. `name` is the filename
/// the operator's browser reported, kept only for display - it never
/// contributes to a path, which is what keeps an upload from being able to
/// traverse outside its own directory.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Attachment {
    /// Server-minted id; also the file's stem under `attachments_dir`.
    pub id: String,
    /// The operator's own filename, for display only.
    pub name: String,
    /// Validated by `web` at upload time against a closed whitelist:
    /// `image/png`, `image/jpeg`, `image/gif`, `image/webp`.
    pub mime: String,
    /// Size in bytes, so the phone can show it without a second request.
    pub bytes: u64,
}

/// One message in the conversation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Turn {
    /// Who wrote it.
    pub who: Who,
    /// What they said.
    pub body: String,
    /// When it was said.
    pub at: Timestamp,
    /// Images attached to this turn. `#[serde(default)]` so a conversation
    /// recorded before attachments existed still reads - see
    /// `a_chat_recorded_without_a_from_field_still_reads`'s sibling test for
    /// this field.
    #[serde(default)]
    pub attachments: Vec<Attachment>,
}

/// Where a conversation is in its life.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ChatStatus {
    /// Still being talked through.
    Open,
    /// A task was filed from its draft.
    Filed,
    /// Given up on. Kept on disk, because an abandoned interview is still the
    /// record of a decision the operator made.
    Abandoned,
}

impl ChatStatus {
    /// Is this conversation still live?
    pub fn open(self) -> bool {
        matches!(self, Self::Open)
    }

    /// Wire form, for the phone and for logs.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Open => "open",
            Self::Filed => "filed",
            Self::Abandoned => "abandoned",
        }
    }
}

/// One planning conversation.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Chat {
    /// On-disk format version.
    pub schema: u32,
    /// Conversation id, e.g. `20260903-014455-ab12`.
    pub id: String,
    /// Repository the task will be filed against.
    pub repo: PathBuf,
    /// The chat this one was derived from, when it began as a fork into a
    /// different repository. See [`derived_background`]. `#[serde(default)]`
    /// so a conversation recorded before this field existed still reads.
    #[serde(default)]
    pub from: Option<String>,
    /// Roster agent id doing the interviewing.
    pub agent: String,
    /// Current state.
    pub status: ChatStatus,
    /// Everything said, oldest first.
    pub turns: Vec<Turn>,
    /// The task file, once the agent has written one.
    pub draft: Option<String>,
    /// Queue task id, once filed.
    pub task: Option<String>,
    /// When the conversation was opened.
    pub created_at: Timestamp,
    /// Last change to this file.
    pub updated_at: Timestamp,
    /// The CLI-side conversation, which is what makes turn N+1 cost one
    /// sentence instead of the whole transcript.
    ///
    /// Not `pub`: it is magi's bookkeeping, not part of the interview, and a
    /// caller that edited it would silently detach the record from the
    /// conversation the model is actually holding. It is still serialized,
    /// because a chat that survives a restart without its session id resumes
    /// nothing.
    seat: SeatState,
}

impl Chat {
    /// Short form used in lists and notifications, matching a run's short id.
    pub fn short(&self) -> &str {
        short(&self.id)
    }

    /// How many turns the interviewing agent has actually taken.
    ///
    /// Read off the seat rather than counted from [`Chat::turns`], because a
    /// failed turn appends a [`MAGI_NOTE`] message that no agent wrote. The
    /// number names artifacts, so it has to match what was invoked.
    pub fn agent_turns(&self) -> usize {
        self.seat.turns
    }
}

/// A conversation store on disk.
#[derive(Debug, Clone)]
pub struct Chats {
    root: PathBuf,
    /// Serializes the read-modify-write cycle that reads a chat, decides
    /// something from its `status`, and writes the whole record back.
    /// [`abandon`], [`file_draft`] and the tail of [`turn`] all take this
    /// before that cycle rather than after just the read: a re-read narrows
    /// the window another writer can land in, but does not close it, since
    /// nothing stopped that other writer's own put from landing between this
    /// call's re-read and its own put. Shared across every clone, since every
    /// clone is a handle onto the same files - see [`crate::talk::Talks`],
    /// which this mirrors.
    lock: Arc<Mutex<()>>,
}

impl Chats {
    /// The operator's conversations, `<home>/chats`.
    pub fn open() -> Self {
        Self::at(crate::run::home().join("chats"))
    }

    /// A store at an explicit root. Tests use this, which is why none of them
    /// need the operator's real home.
    pub fn at(root: PathBuf) -> Self {
        Self {
            root,
            lock: Arc::new(Mutex::new(())),
        }
    }

    /// Claim the right to read-modify-write a chat's `status`. See
    /// [`crate::talk::Talks::guard`], which this mirrors, including recovering
    /// from poisoning rather than propagating it: one panicking caller must
    /// not wedge every chat in the store.
    ///
    /// This is an in-process `Mutex` - it serializes callers inside one
    /// `magi web`, but is invisible to `magi plan --abandon` running as its
    /// own process with its own `Arc`. [`Chats::claim`] is what closes that
    /// gap; the two are meant to be taken together, this one first.
    fn guard(&self) -> MutexGuard<'_, ()> {
        self.lock.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Path of the claim lock for a conversation. One definition, so
    /// [`Chats::claim`] cannot end up naming a different file than whatever
    /// else might go looking for it.
    fn lock_path(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.lock"))
    }

    /// Take cross-process exclusive ownership of one chat's record for the
    /// read-modify-write cycle that decides its `status`.
    ///
    /// [`Chats::guard`] only ever sees other callers inside the same process;
    /// `magi plan --abandon` is a separate CLI invocation with its own
    /// `Arc<Mutex<()>>`, wired to nothing a running `magi web` holds. This is
    /// a `create_new` file next to the record instead - atomic on every
    /// platform magi targets, and invisible to no process that asks - the
    /// same primitive [`crate::queue::Queue::claim`] uses to keep a CLI edit
    /// and a running daemon off the same task file at once. The returned
    /// guard releases on drop, including on panic.
    fn claim(&self, id: &str) -> Result<ChatClaim> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        let path = self.lock_path(id);
        match std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&path)
        {
            Ok(mut f) => {
                use std::io::Write as _;
                // Best effort: the pid is for the human looking at a stale lock.
                let _ = writeln!(f, "{}", std::process::id());
                Ok(ChatClaim { path })
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                bail!("chat {id} is claimed by another process right now")
            }
            Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
        }
    }

    /// Directory holding the conversation files.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Path for one conversation id.
    pub fn path_of(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.json"))
    }

    /// Where one conversation's prompts and CLI output are kept.
    ///
    /// Beside the record rather than inside it, with the same stem convention a
    /// run's nodes use, so a conversation that went wrong can be read back
    /// turn by turn - which is the only way to tell "the agent said nothing"
    /// apart from "magi never asked it".
    pub fn artifacts_of(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.artifacts"))
    }

    /// Where this conversation's attached images live: a subdirectory of
    /// `artifacts_of`, so deleting the conversation deletes its attachments
    /// too and nothing here needs its own cleanup path.
    pub fn attachments_dir(&self, id: &str) -> PathBuf {
        self.artifacts_of(id).join("attachments")
    }

    /// Persist one already-validated attachment and return its metadata.
    ///
    /// `web::chat_attachment_post` is the only caller: it has already
    /// checked `mime` against the whitelist and sniffed the bytes, so an
    /// unrecognised mime reaching here is a bug in that caller, not
    /// something an operator did. The id is minted here and never taken
    /// from the client; `name` is stored for display only and never used to
    /// build a path.
    pub fn put_attachment(
        &self,
        id: &str,
        mime: &str,
        name: &str,
        data: &[u8],
    ) -> Result<Attachment> {
        let dir = self.attachments_dir(id);
        std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
        let ext = attachment_ext(mime).with_context(|| format!("unsupported mime `{mime}`"))?;
        let att = Attachment {
            id: new_attachment_id(),
            name: name.to_owned(),
            mime: mime.to_owned(),
            bytes: data.len() as u64,
        };
        std::fs::write(dir.join(format!("{}.{ext}", att.id)), data)
            .with_context(|| format!("write attachment {}", att.id))?;
        std::fs::write(
            dir.join(format!("{}.json", att.id)),
            serde_json::to_string(&att).context("serialize attachment")?,
        )
        .with_context(|| format!("write attachment metadata {}", att.id))?;
        Ok(att)
    }

    /// Just the metadata, without reading the image bytes back off disk -
    /// what `web::chat_say` uses to turn an id the operator referenced into
    /// an [`Attachment`] before appending a [`Turn`], where the bytes
    /// themselves are of no interest. `None` for an id this conversation
    /// never stored - including one that merely looks plausible:
    /// [`valid_attachment_id`] is checked here too, not only by the caller,
    /// the same defence-in-depth `Questions::panel_asset` uses for its own
    /// asset ids.
    pub fn attachment_meta(&self, id: &str, att_id: &str) -> Result<Option<Attachment>> {
        if !valid_attachment_id(att_id) {
            return Ok(None);
        }
        let meta_path = self.attachments_dir(id).join(format!("{att_id}.json"));
        if !meta_path.is_file() {
            return Ok(None);
        }
        let att = serde_json::from_str(
            &std::fs::read_to_string(&meta_path)
                .with_context(|| format!("read {}", meta_path.display()))?,
        )
        .with_context(|| format!("parse {}", meta_path.display()))?;
        Ok(Some(att))
    }

    /// A stored attachment's metadata and its bytes together, for serving it
    /// back on `GET`. `None` under the same conditions as
    /// [`Chats::attachment_meta`], which this is built on.
    pub fn read_attachment(&self, id: &str, att_id: &str) -> Result<Option<(Attachment, Vec<u8>)>> {
        let Some(att) = self.attachment_meta(id, att_id)? else {
            return Ok(None);
        };
        let ext = attachment_ext(&att.mime).with_context(|| {
            format!("attachment {att_id} has an unsupported mime `{}`", att.mime)
        })?;
        let data_path = self.attachments_dir(id).join(format!("{att_id}.{ext}"));
        let data =
            std::fs::read(&data_path).with_context(|| format!("read {}", data_path.display()))?;
        Ok(Some((att, data)))
    }

    /// Absolute path of one attachment's bytes, for the prompt note [`turn`]
    /// appends and for [`Invocation::attachments`]. `None` only for a mime
    /// [`put_attachment`] could never have written, which means the
    /// attachment did not come from this store.
    ///
    /// `self.root` (and so `attachments_dir`) is not guaranteed absolute on
    /// its own - `run::home()` returns a bare relative `PathBuf` verbatim
    /// when the operator sets `MAGI_HOME` to a relative path, and nothing
    /// canonicalizes it on the way in. That is harmless for every other use
    /// of this store, since its own I/O runs in this process against this
    /// process's cwd - but this path is handed to a CLI invoked with `cwd:
    /// &chat.repo`, a different directory, so a relative path here would
    /// resolve against the wrong place once it reached the prompt.
    /// `std::path::absolute` fixes it against *this* process's cwd before
    /// that happens; see `disk::free_bytes_by_os` for the same function used
    /// the same way elsewhere in this codebase.
    fn attachment_path(&self, id: &str, att: &Attachment) -> Option<PathBuf> {
        let ext = attachment_ext(&att.mime)?;
        let path = self.attachments_dir(id).join(format!("{}.{ext}", att.id));
        std::path::absolute(&path).ok()
    }

    /// Write a conversation, atomically, so a process killed mid-write leaves
    /// the previous state readable rather than a truncated file that would lose
    /// the whole interview.
    pub fn put(&self, c: &mut Chat) -> Result<()> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        c.updated_at = Timestamp::now();
        let body = serde_json::to_string_pretty(c).context("serialize chat")?;
        let path = self.path_of(&c.id);
        let tmp = path.with_extension("json.tmp");
        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
        std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
        Ok(())
    }

    /// Load a conversation by id or unambiguous id prefix.
    pub fn get(&self, id: &str) -> Result<Chat> {
        let resolved = self.resolve_id(id)?;
        read_path(&self.path_of(&resolved))
    }

    /// Every conversation on disk: open first, then newest first.
    ///
    /// Open first because that ordering is the product - the list exists to
    /// show the operator what is still being talked through, and a filed
    /// interview is history underneath it. Unreadable files are skipped rather
    /// than fatal: one corrupt record must not take the web UI down, and must
    /// certainly not hide the open conversation the operator came back for.
    pub fn list(&self) -> Vec<Chat> {
        let mut all: Vec<Chat> = std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|x| x == "json"))
            .filter_map(|p| read_path(&p).ok())
            .collect();
        all.sort_unstable_by(|a, b| {
            let rank = |c: &Chat| u8::from(!c.status.open());
            rank(a).cmp(&rank(b)).then_with(|| b.id.cmp(&a.id))
        });
        all
    }

    /// Expand an id prefix to exactly one conversation id. The short id the
    /// phone shows is a suffix, so that is accepted too.
    pub fn resolve_id(&self, prefix: &str) -> Result<String> {
        if self.path_of(prefix).is_file() {
            return Ok(prefix.to_owned());
        }
        let hits: Vec<String> = self
            .list()
            .into_iter()
            .map(|c| c.id)
            .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
            .collect();
        match hits.len() {
            1 => Ok(hits.into_iter().next().expect("exactly one hit")),
            0 => bail!("no chat matches `{prefix}`"),
            _ => bail!(
                "`{prefix}` matches {} chats: {}",
                hits.len(),
                hits.join(", ")
            ),
        }
    }

    /// Newest modification time in the store, in milliseconds, for change
    /// detection. The web UI compares this instead of re-reading every
    /// conversation, so an idle phone on a slow link costs one `stat` per file.
    pub fn revision(&self) -> u64 {
        std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .filter_map(|e| e.metadata().ok())
            .filter_map(|m| m.modified().ok())
            .filter_map(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as u64)
            .max()
            .unwrap_or(0)
    }

    /// How many conversations are still open. The badge on the phone.
    pub fn count_open(&self) -> usize {
        self.list().iter().filter(|c| c.status.open()).count()
    }
}

/// Cross-process exclusive ownership of one chat's record, released on drop.
/// See [`Chats::claim`], which this is returned by, and
/// [`crate::queue::Claim`], which it mirrors.
#[derive(Debug)]
struct ChatClaim {
    path: PathBuf,
}

impl Drop for ChatClaim {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.path);
    }
}

/// Construct a conversation record in memory, without writing it anywhere.
///
/// Split out of [`open`] so a caller can claim [`crate::web::Ui::begin_turn`]
/// on `chat.id` *before* the record is ever written to disk - `chat_post`
/// does exactly that, because [`Chats::put`] is what makes the id visible to
/// every other request (`GET /api/chats`, `POST /api/chats/{id}/say`), and a
/// gap between "the file exists" and "the turn is claimed" is a window for
/// `chat_say` to claim and record into an interview whose first turn never
/// ran - see `chat_post`'s doc for the failure that produces.
///
/// `agent` is resolved by [`plan::pick`], the same policy `magi plan` uses: an
/// explicit id wins and is an error rather than a fallback when it is not
/// runnable, otherwise a `claude` seat, otherwise the first runnable agent in
/// roster order. Called rather than copied, because two copies of a preference
/// order drift and the copy that drifts is the one nobody reads.
///
/// `from` is the conversation this one was derived from, when the operator
/// asked to continue an existing interview in a different repository (see
/// [`derived_background`]). It is read, never written: the source chat's
/// `status`, `turns` and `draft` are left exactly as they were.
pub fn build(
    cfg: &Config,
    repo: PathBuf,
    idea: &str,
    agent: Option<&str>,
    from: Option<&Chat>,
) -> Result<Chat> {
    let idea = idea.trim();
    if idea.is_empty() {
        bail!("an interview needs something to start from: say what you want to change");
    }
    // Absolute, because the daemon that eventually runs the filed task has its
    // own working directory and a relative path would mean the wrong
    // repository.
    let repo = repo.canonicalize().unwrap_or(repo);
    // The API's `agent` beats the config, the config beats the built-in order.
    // On a phone there is no flag to pass, so `[roles] chatter` (falling back
    // to `planner`, so an operator who never set `chatter` sees no change) is
    // the only way an operator states who answers this conversation - kept
    // separate from `planner` so the resident chat does not compete with a
    // judge seat for the same account by default.
    let want = agent
        .or(cfg.roles.chatter.as_deref())
        .or(cfg.roles.planner.as_deref());
    let spec = plan::pick(&cfg.agents, want, &plan::installed)?;

    let now = Timestamp::now();
    let id = new_id();
    Ok(Chat {
        schema: SCHEMA,
        id,
        repo,
        from: from.map(|c| c.id.clone()),
        agent: spec.id.clone(),
        status: ChatStatus::Open,
        turns: vec![Turn {
            who: Who::Operator,
            body: idea.to_owned(),
            at: now,
            // The idea box that opens an interview has no attachment path of
            // its own - only the ongoing `chat-say` composer does, once a
            // conversation (and therefore an `artifacts_of` id to hold
            // uploads under) exists.
            attachments: Vec::new(),
        }],
        draft: None,
        task: None,
        created_at: now,
        updated_at: now,
        seat: SeatState::new(SEAT, &spec.id, crate::rng::entropy()),
    })
}

/// [`build`], then persist. No first agent turn is taken.
///
/// Convenient when there is nothing racing the write - [`start`] is the only
/// caller - but `POST /api/chats` cannot use it: see [`build`]'s doc for why
/// the claim has to land between construction and this function's own
/// [`Chats::put`].
pub fn open(
    store: &Chats,
    cfg: &Config,
    repo: PathBuf,
    idea: &str,
    agent: Option<&str>,
    from: Option<&Chat>,
) -> Result<Chat> {
    let mut chat = build(cfg, repo, idea, agent, from)?;
    store.put(&mut chat)?;
    Ok(chat)
}

/// Take the first agent turn of a conversation created by [`open`].
///
/// `from`, when given, must be the same source conversation `open` was called
/// with - it is read again here rather than stashed on `chat` because the
/// persisted record carries only the source's id, and the full record is
/// what [`derived_background`] needs.
pub async fn first_turn(
    chat: &mut Chat,
    store: &Chats,
    cfg: &Config,
    from: Option<&Chat>,
) -> Result<()> {
    let idea = chat
        .turns
        .first()
        .map(|t| t.body.as_str())
        .unwrap_or_default();
    let mut prompt = briefing(idea, &chat.repo);
    // The source's own attachments, resolved against *its* id - `derived_background`
    // already names their absolute paths in the prompt text below, and those
    // paths live under the source conversation's own artifacts dir, not this
    // new one's, so `turn` needs them passed in separately to widen a sandbox
    // that only ever assumes its own conversation's directory.
    let mut inherited_attachments: Vec<PathBuf> = Vec::new();
    if let Some(source) = from {
        // Prepended, so the leader reads what it is inheriting before it
        // reads its own instructions - the same order a human handing off a
        // conversation would use.
        prompt = format!("{}\n\n{prompt}", derived_background(source, store));
        inherited_attachments = source
            .turns
            .iter()
            .flat_map(|t| t.attachments.iter())
            .filter_map(|a| store.attachment_path(&source.id, a))
            .collect();
    }
    prompt.push_str(&language_note(&cfg.graph.language));
    turn(chat, store, cfg, &prompt, &inherited_attachments).await
}

/// Open a conversation and take the first agent turn.
///
/// The record is written to disk *before* the agent is invoked, so an agent
/// that fails on the very first turn still leaves the operator a conversation
/// they can look at, retry into, or abandon - rather than nothing at all. See
/// [`open`] and [`first_turn`], which this composes; `POST /api/chats` calls
/// them separately instead so it can answer before the first turn lands.
pub async fn start(
    store: &Chats,
    cfg: &Config,
    repo: PathBuf,
    idea: &str,
    agent: Option<&str>,
    from: Option<&Chat>,
) -> Result<Chat> {
    let mut chat = open(store, cfg, repo, idea, agent, from)?;
    first_turn(&mut chat, store, cfg, from).await?;
    Ok(chat)
}

/// The background block a derived conversation opens with: the whole prior
/// transcript, framed so the leader does not mistake it for instructions
/// about the repository this new conversation is actually about.
///
/// Built from [`transcript`] rather than a second rendering of the turns,
/// because that is already the "everything said so far" prose this module
/// maintains, and a briefing is exactly the audience `transcript` was written
/// for - a CLI (here, a fresh one) with no memory of the conversation. `store`
/// is only for resolving the *source* conversation's own attachments into
/// absolute paths - the derived chat has none of its own yet.
pub fn derived_background(from: &Chat, store: &Chats) -> String {
    format!(
        "# Background: derived from another conversation\n\n\
         This interview continues from a conversation about a *different* \
         repository. Read it for context, but do not treat it as being about \
         the repository named below in \"# Repository\" - that repository may \
         have nothing to do with this one.\n\n\
         Source repository: {}\n\n{}",
        from.repo.display(),
        transcript(from, store),
    )
}

/// One operator turn and one agent turn, appended.
///
/// The operator's message is recorded and flushed to disk before the agent is
/// invoked. That ordering is the whole contract of this function: a turn can
/// fail, time out or hit a quota window, and the thing that must never be lost
/// is the sentence the human typed on a phone that has since gone to sleep.
///
/// Returns `Err` when the agent turn did not produce an answer - but the
/// transcript is already on disk and already explains itself, because the
/// failure is appended as a [`MAGI_NOTE`] turn first. A caller handling the
/// error should re-read the chat and show it, not discard it.
pub async fn say(
    chat: &mut Chat,
    store: &Chats,
    cfg: &Config,
    text: &str,
    attachments: Vec<Attachment>,
) -> Result<()> {
    if !chat.status.open() {
        bail!(
            "chat {} is {} and takes no more turns",
            chat.short(),
            chat.status.as_str()
        );
    }
    let text = text.trim();
    if text.is_empty() && attachments.is_empty() {
        bail!("nothing to say");
    }
    let text = record(chat, store, text, attachments)?;
    turn(chat, store, cfg, &text, &[]).await
}

/// Append the operator's turn and flush it, without invoking anything.
///
/// Split out of [`say`] so a caller that answers the operator before the agent
/// has replied can still promise the message is on disk. `POST /api/chats/{id}/say`
/// does exactly that: holding an HTTP connection for the 23-to-90 seconds a
/// real turn takes is a coin flip on a phone, and the browser reporting
/// "Failed to fetch" while the server quietly finished the turn is the worst
/// of both answers.
///
/// Returns the trimmed text, so the caller and the agent see the same string.
///
/// Re-reads the record under [`Chats::guard`] and [`Chats::claim`] rather
/// than trusting the caller's copy of `chat`'s `status`: `web::chat_say`
/// reads the chat, then awaits config discovery before calling this - a gap
/// a concurrent `POST /api/chats/{id}/abandon`, or a `magi plan --abandon`
/// running as its own process, can land in. Re-reading without both would
/// only narrow that window, not close it - see [`crate::talk::record`],
/// which this mirrors for the in-process half.
///
/// `attachments` may be non-empty while `text` is empty - a turn that is
/// only images is a normal thing to send - but not both empty, the same rule
/// this always enforced for text alone.
pub fn record(
    chat: &mut Chat,
    store: &Chats,
    text: &str,
    attachments: Vec<Attachment>,
) -> Result<String> {
    let _guard = store.guard();
    let _claim = store.claim(&chat.id)?;
    let fresh = store
        .get(&chat.id)
        .with_context(|| format!("chat {} could not be re-read", chat.short()))?;
    chat.status = fresh.status;
    if !chat.status.open() {
        bail!(
            "chat {} is {} and takes no more turns",
            chat.short(),
            chat.status.as_str()
        );
    }
    let text = text.trim();
    if text.is_empty() && attachments.is_empty() {
        bail!("nothing to say");
    }
    chat.turns.push(Turn {
        who: Who::Operator,
        body: text.to_owned(),
        at: Timestamp::now(),
        attachments,
    });
    store.put(chat)?;
    Ok(text.to_owned())
}

/// The agent's half of a turn: invoke, append, flush.
///
/// Pairs with [`record`]. `text` is the operator's message that this reply
/// answers - the same string `record` returned, so the transcript and the
/// prompt cannot disagree.
pub async fn respond(chat: &mut Chat, store: &Chats, cfg: &Config, text: &str) -> Result<()> {
    turn(chat, store, cfg, text, &[]).await
}

/// Invoke the interviewing agent once and append what it said.
///
/// `prompt` is only the new material. Whether that is enough depends on the
/// CLI: [`agent::has_session`] answers honestly - it is `false` when sessions
/// are switched off, before the first turn, or for a CLI that never reported an
/// id back - and only then is the transcript prepended, because a model with no
/// memory of the interview would otherwise answer the last sentence in a
/// vacuum. When the CLI *can* resume, magi sends nothing extra: paying for the
/// whole conversation on every message is the cost this design exists to avoid,
/// and a magi-authored replay of history is also a second, divergent version of
/// it.
async fn turn(
    chat: &mut Chat,
    store: &Chats,
    cfg: &Config,
    prompt: &str,
    inherited_attachments: &[PathBuf],
) -> Result<()> {
    let spec = cfg
        .agents
        .iter()
        .find(|a| a.id == chat.agent)
        .with_context(|| {
            format!(
                "chat {} was interviewed by agent `{}`, which is no longer in \
                 the roster; restore it in magi.toml or start a new chat",
                chat.short(),
                chat.agent
            )
        })?;

    let resuming = agent::has_session(spec.kind, &chat.seat, cfg.graph.sessions);
    // The newest turn is always the operator message this call is answering
    // - `record` (or `start`, for the very first turn) appended it before
    // `turn` was ever called - so its own attachments are what belong at the
    // end of *this* prompt, resuming or not.
    let last_note = attachment_note(
        store,
        &chat.id,
        chat.turns
            .last()
            .map_or(&[][..], |t| t.attachments.as_slice()),
    );
    let body = if resuming {
        format!("{prompt}{last_note}")
    } else {
        format!("{}\n\n{prompt}{last_note}", transcript(chat, store))
    };

    // Every attachment this conversation has ever held, not only this
    // turn's: a resumed session gets a fresh process every turn, so a CLI
    // whose sandbox needs `--add-dir` (see `agent::build_command`) needs the
    // grant again to open an image from three turns ago, even when nothing
    // new was attached just now. Plus whatever the caller inherited from a
    // *different* conversation - `first_turn` passes the source's own
    // attachments here, since `derived_background` already named their paths
    // in the prompt and a sandbox that only widens for `chat.id`'s own
    // directory would leave those unreadable.
    let attachment_paths: Vec<PathBuf> = chat
        .turns
        .iter()
        .flat_map(|t| t.attachments.iter())
        .filter_map(|a| store.attachment_path(&chat.id, a))
        .chain(inherited_attachments.iter().cloned())
        .collect();

    let artifacts = store.artifacts_of(&chat.id);
    let stem = format!("turn-{}", chat.seat.turns + 1);
    let cache_dir = cfg.cache_dir();
    let inv = Invocation {
        cwd: &chat.repo,
        prompt: &body,
        timeout: turn_timeout(cfg),
        // The interviewer writes a task file into its reply, never into the
        // repository: the competing agents do the implementation, and a
        // repository the planner has already edited makes their diffs
        // unjudgeable.
        allow_write: false,
        sessions: cfg.graph.sessions,
        artifacts: &artifacts,
        stem: &stem,
        run: &chat.id,
        node: "chat",
        cache_dir: cache_dir.as_deref(),
        attachments: &attachment_paths,
    };

    let outcome = agent::invoke(spec, &mut chat.seat, &inv).await;
    let note = |why: String| Turn {
        who: Who::Agent,
        body: format!("{MAGI_NOTE}{why}"),
        at: Timestamp::now(),
        attachments: Vec::new(),
    };
    let (reply, failure) = match outcome {
        Err(e) => (
            note(format!("could not run agent `{}`: {e}", chat.agent)),
            Some(format!("could not run agent `{}`: {e}", chat.agent)),
        ),
        Ok(out) if out.quota_exhausted() => {
            let reset = out
                .quota
                .as_ref()
                .and_then(|q| q.reset.clone())
                .map_or_else(String::new, |r| format!(" (resets {r})"));
            let why = format!(
                "agent `{}` is out of quota{reset}; your message is saved, so \
                 say it again when the window reopens",
                chat.agent
            );
            (note(why.clone()), Some(why))
        }
        Ok(out) if out.timed_out => {
            let why = format!(
                "agent `{}` did not answer within {}s; your message is saved",
                chat.agent,
                turn_timeout(cfg).as_secs()
            );
            (note(why.clone()), Some(why))
        }
        Ok(out) if !out.usable() => {
            let why = format!(
                "agent `{}` produced no answer (exit {}); your message is saved",
                chat.agent,
                out.exit_code
                    .map_or_else(|| "unknown".to_owned(), |c| c.to_string())
            );
            (note(why.clone()), Some(why))
        }
        Ok(out) => (
            Turn {
                who: Who::Agent,
                body: out.text.trim().to_owned(),
                at: Timestamp::now(),
                attachments: Vec::new(),
            },
            None,
        ),
    };

    // A reply carrying no fenced draft leaves the existing one alone. The agent
    // asking one more follow-up question must not erase the task file it
    // already wrote, which the operator may well be reading at that moment.
    if let Some(draft) = extract_draft(&reply.body) {
        chat.draft = Some(draft);
    }

    // A concurrent `abandon` or `file_draft` can have landed on disk while
    // this turn - possibly minutes long - was in flight, from inside this
    // same `magi web` or from a separate `magi plan --abandon` process. Read
    // `status` and `task` back here, under the same guard and claim those two
    // take, rather than trust the snapshot this call started with: finishing
    // the turn on that stale snapshot would silently undo whichever of them
    // got there first. See [`crate::talk::turn`]'s tail, which this mirrors
    // for the in-process half.
    let _guard = store.guard();
    let _claim = store.claim(&chat.id)?;
    let fresh = store
        .get(&chat.id)
        .with_context(|| format!("chat {} could not be re-read", chat.short()))?;
    chat.status = fresh.status;
    chat.task = fresh.task;
    chat.turns.push(reply);
    store.put(chat)?;

    match failure {
        Some(why) => bail!("{why}"),
        None => Ok(()),
    }
}

/// Everything said so far, as prose, for a CLI that cannot resume.
///
/// Only reached when [`agent::has_session`] says the conversation cannot be
/// continued on the CLI's side. It is a fallback and not the design: it re-pays
/// for the history on every turn and it is magi's rendering of the
/// conversation rather than the model's own.
fn transcript(chat: &Chat, store: &Chats) -> String {
    let mut out = String::from(
        "You are mid-interview. This CLI cannot resume its own conversation, \
         so here is everything said so far; answer only the last message.\n",
    );
    for t in &chat.turns {
        let who = match t.who {
            Who::Operator => "operator",
            Who::Agent => "you",
        };
        out.push_str(&format!("\n## {who}\n\n{}\n", t.body.trim()));
        out.push_str(&attachment_note(store, &chat.id, &t.attachments));
    }
    out
}

/// The section named at the end of a turn's body, listing every attachment's
/// absolute path and mime so the agent knows exactly what to open - see the
/// module doc on `Invocation::attachments`. Empty when `attachments` is,
/// which is every turn but the rare one carrying an image, so a turn with
/// none changes nothing about the prompt.
fn attachment_note(store: &Chats, chat_id: &str, attachments: &[Attachment]) -> String {
    if attachments.is_empty() {
        return String::new();
    }
    let mut out = String::from(
        "\n\nThe operator attached the image(s) below to this message. Open \
         and look at each one before you answer.\n",
    );
    for att in attachments {
        if let Some(path) = store.attachment_path(chat_id, att) {
            out.push_str(&format!("\n- {} ({})", path.display(), att.mime));
        }
    }
    out.push('\n');
    out
}

/// Validate the draft with [`plan::review_draft`] and queue it.
///
/// Returns the queued task's id. The conversation is left on disk either way:
/// a refused draft is a conversation to continue, not an error to recover
/// from, and the operator's next message can ask for the missing section.
///
/// Re-reads the record under [`Chats::guard`] and [`Chats::claim`] rather
/// than trusting the caller's copy of `chat`, and writes that fresh copy back
/// rather than the one passed in - the same reason [`abandon`] does, and for
/// the same concurrent-writer risk: without the shared guard *and* claim, an
/// `abandon` landing between this call's own read and its `put` - whether
/// from this same `magi web` or from a separate `magi plan --abandon`
/// process, which the in-process guard alone cannot see - would either be
/// clobbered by this call filing over it, or - the other order - have this
/// call's queued task silently orphaned from the record when `abandon`
/// writes over it. Refuses a conversation that is no longer `Open` by the
/// time this runs, in the same style [`say`] and [`record`] already use.
pub fn file_draft(chat: &mut Chat, store: &Chats, queue: &Queue, priority: i32) -> Result<String> {
    let _guard = store.guard();
    let _claim = store.claim(&chat.id)?;
    let mut fresh = store
        .get(&chat.id)
        .with_context(|| format!("chat {} could not be re-read", chat.short()))?;
    if !fresh.status.open() {
        bail!(
            "chat {} is {} and takes no more turns",
            fresh.short(),
            fresh.status.as_str()
        );
    }
    if let Err(problems) = draft_problems(&fresh) {
        bail!(
            "this draft is not fileable yet:\n- {}",
            problems.join("\n- ")
        );
    }
    let body = fresh
        .draft
        .clone()
        .expect("draft_problems accepted a chat with a draft");

    // `title_from` rather than a title the agent was asked to supply
    // separately: the task file's first line already is the title, and asking
    // for it twice is how the two come to disagree.
    let title = queue::title_from(&body, 72);
    // `Human`, not `Agent`: the agent conducted the interview, but the change
    // being asked for is the operator's, and "who asked for this" is the
    // question `source` exists to answer.
    let mut task = Task::new(title, body, fresh.repo.clone(), Source::Human);
    task.priority = priority;
    queue.put(&mut task)?;

    fresh.task = Some(task.id.clone());
    fresh.status = ChatStatus::Filed;
    store.put(&mut fresh)?;
    *chat = fresh;
    Ok(task.id)
}

/// Give up on a conversation. Idempotent: abandoning an already-abandoned
/// conversation is not an error, since the operator's intent - "I don't want
/// this anymore" - is already satisfied. Refuses a `Filed` conversation: that
/// status means a task was already produced from this interview, and abandon
/// must not roll back the terminal state that produced it.
///
/// Re-reads the record under [`Chats::guard`] rather than trusting the
/// caller's copy of `chat`, and writes that fresh copy back rather than the
/// one passed in - the same reason [`crate::talk::close`] does: a concurrent
/// [`say`] or [`file_draft`] must not have its result overwritten by a
/// decision made against a stale snapshot's idea of what `status` was, and the
/// guard is what stops that snapshot from being made stale *again* between
/// this call's own re-read and its `put`.
///
/// Also takes [`Chats::claim`], because unlike [`say`] and [`file_draft`] this
/// function is reachable from `magi plan --abandon` - a separate CLI process,
/// with its own `Arc<Mutex<()>>` that `store.guard()` shares with nothing
/// `magi web` holds. The claim is a `create_new` file next to the record
/// instead, which every process asking for it sees, so a chat cannot be
/// abandoned here at the exact moment `magi web` is filing or answering it.
pub fn abandon(chat: &mut Chat, store: &Chats) -> Result<()> {
    let _guard = store.guard();
    let _claim = store.claim(&chat.id)?;
    let mut fresh = store
        .get(&chat.id)
        .with_context(|| format!("chat {} could not be re-read", chat.short()))?;
    match fresh.status {
        ChatStatus::Open => {
            fresh.status = ChatStatus::Abandoned;
            store.put(&mut fresh)?;
        }
        ChatStatus::Abandoned => {}
        ChatStatus::Filed => bail!(
            "chat {} is {} and takes no more turns",
            fresh.short(),
            fresh.status.as_str()
        ),
    }
    *chat = fresh;
    Ok(())
}

/// Is this conversation's draft fileable, and if not, what is wrong with it?
///
/// Every problem is returned, not the first: an operator about to ask the agent
/// for a fix wants the whole list, and a validator that reveals one defect per
/// round turns one follow-up message into three.
///
/// [`plan::SHORT_DRAFT`] alone does not refuse. Length is a smell, not a
/// defect, and a genuinely small change deserves a small task file - which is
/// exactly the judgement `magi plan` makes, so the browser path makes it too.
/// It is still reported, because a two-line draft is usually an interview that
/// ended early.
pub fn draft_problems(chat: &Chat) -> Result<(), Vec<String>> {
    let Some(body) = chat.draft.as_deref() else {
        return Err(vec![
            "this chat has no draft yet: the agent has not written a task file".to_owned(),
        ]);
    };
    match plan::review_draft(body) {
        Ok(()) => Ok(()),
        Err(problems) => {
            if problems.iter().all(|p| p == plan::SHORT_DRAFT) {
                Ok(())
            } else {
                Err(problems)
            }
        }
    }
}

/// The briefing the agent is opened with.
///
/// Pure, so the one property that matters can be asserted without an
/// interview: it carries [`plan::TASK_FILE_SPEC`] verbatim. The spec and
/// [`plan::review_draft`] are checked against each other by `plan`'s own tests,
/// so including it here is what keeps this path from asking for a shape the
/// validator will refuse - a twenty-message interview rejected for a reason the
/// operator was never told is the worst outcome this module has.
///
/// The output contract is the other half. `magi plan` tells the agent to write
/// a file, which works because that agent has a terminal and a filesystem the
/// operator is watching. Here the reply *is* the channel: the task file comes
/// back inside a fenced block tagged `task`, and [`extract_draft`] is the only
/// thing that reads it.
pub fn briefing(idea: &str, repo: &Path) -> String {
    format!(
        "You are the planning leader for magi, which runs a blind \
         multi-agent implementation competition: several agents will implement \
         the task file you write, in isolated worktrees, unaware of each other, \
         and judges will rank the results without knowing who wrote what.\n\n\
         Your job is not to implement anything. It is to interview the operator \
         until the change is pinned down, and then write one task file.\n\n\
         The operator is on a phone. Every message you send is read on a small \
         screen, so keep it short: no preamble, no restating what they just \
         said.\n\n\
         # Repository\n\n{repo}\n\n\
         Read it before you start asking. Questions the code already answers \
         spend the operator's patience for nothing. Do not modify it: the \
         competing agents do the implementation, and a repository you have \
         already edited makes their diffs unjudgeable.\n\n\
         # The idea\n\n{idea}\n\n\
         # How to run the interview\n\n\
         - Ask about what you cannot determine yourself: intent, scope, which \
         of several defensible designs the operator wants, what must not \
         change.\n\
         - Ask about ONE thing per message and wait for the answer. This is a \
         phone, not a form: a message with five questions in it gets one of \
         them answered.\n\
         - Do not produce the task file after one exchange.\n\
         - Disagree when you have grounds. A leader that agrees with everything \
         adds nothing to what the operator already typed.\n\
         - Confirm the plan in your own words and get an explicit yes before \
         writing.\n\n\
         # How to deliver the task file\n\n\
         When the operator agrees the plan is right, put the whole task file in \
         your reply inside a fenced block tagged `task`, like this:\n\n\
         ```task\n\
         # <the task file>\n\
         ```\n\n\
         Nothing else goes in that block, and there is exactly one of them per \
         message. magi extracts it and files it; a task file written to a file \
         on disk, or pasted without the fence, is one magi cannot see. You may \
         send a revised version later in the same conversation - the newest \
         `task` block wins - and while you are still asking questions, send no \
         `task` block at all.\n\n\
         magi will refuse a task file with no completion criteria, so those are \
         not optional.\n\n\
         # Task file specification\n\n{spec}",
        repo = repo.display(),
        spec = plan::TASK_FILE_SPEC,
    )
}

/// The interview is the operator talking, so their language matters more here
/// than in any prompt the graph sends: an agent that answers a Japanese
/// question in English makes the conversation slower for exactly the person
/// magi is trying to help.
fn language_note(language: &str) -> String {
    if language.trim().is_empty() || language.eq_ignore_ascii_case("en") {
        String::new()
    } else {
        format!("\n\nConduct the interview in {language}, and write the task file in {language}.")
    }
}

/// Pull the task draft out of an agent reply, if it wrote one.
///
/// The *last* fenced `task` block, not the first. A conversation revises: an
/// agent that rewrites the task file after one more answer sends both versions
/// over the course of the interview, and within one message it may quote what
/// it had before changing it. The newest block is the one the operator has been
/// reading and the one they are about to approve.
///
/// Blocks tagged anything else - ```` ```rust ````, ```` ```json ```` - are
/// ignored, so an agent illustrating its plan with code does not overwrite the
/// draft with a snippet. An unterminated block is still taken: a reply cut off
/// mid-draft is worth showing the operator, who can then just ask for it again.
pub fn extract_draft(reply: &str) -> Option<String> {
    let mut last: Option<String> = None;
    let mut open: Option<(usize, Vec<&str>)> = None;
    for line in reply.lines() {
        let trimmed = line.trim_start();
        // Backticks are one byte each, so the count is also the byte offset of
        // the info string.
        let ticks = trimmed.chars().take_while(|c| *c == '`').count();
        match &mut open {
            Some((width, body)) => {
                if ticks >= *width && trimmed[ticks..].trim().is_empty() {
                    last = Some(joined(body));
                    open = None;
                } else {
                    body.push(line);
                }
            }
            None => {
                if ticks >= 3 && trimmed[ticks..].trim().eq_ignore_ascii_case("task") {
                    open = Some((ticks, Vec::new()));
                }
            }
        }
    }
    if let Some((_, body)) = open {
        last = Some(joined(&body));
    }
    last.filter(|s| !s.trim().is_empty())
}

/// A fenced block's lines as one document, newline-terminated the way a file
/// would be, because [`plan::review_draft`] reads it as a task file.
fn joined(lines: &[&str]) -> String {
    if lines.is_empty() {
        return String::new();
    }
    let mut out = lines.join("\n");
    out.push('\n');
    out
}

fn read_path(path: &Path) -> Result<Chat> {
    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
    serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))
}

fn short(id: &str) -> &str {
    id.split('-').next_back().unwrap_or(id)
}

fn new_id() -> String {
    let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
    let seed = crate::rng::entropy();
    format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
}

/// Extension an attachment's bytes are stored under, from its (already
/// validated) mime. The one place this mapping exists on the write side;
/// `web`'s own whitelist is what actually decides which mimes are accepted
/// in the first place.
fn attachment_ext(mime: &str) -> Option<&'static str> {
    match mime {
        "image/png" => Some("png"),
        "image/jpeg" => Some("jpg"),
        "image/gif" => Some("gif"),
        "image/webp" => Some("webp"),
        _ => None,
    }
}

/// Is `id` a shape [`put_attachment`](Chats::put_attachment) could have
/// produced? 32 lowercase hex digits and nothing else, checked before an id
/// that came from the client is ever allowed to build a path - so `..` and a
/// path separator are never even possible.
pub fn valid_attachment_id(id: &str) -> bool {
    id.len() == 32
        && id
            .bytes()
            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}

/// A fresh attachment id: 128 bits of process entropy as lowercase hex - the
/// same "mint it, never take it from the client" rule [`new_id`] follows for
/// conversation ids.
fn new_attachment_id() -> String {
    let mut r = crate::rng::SplitMix64::new(crate::rng::entropy());
    format!("{:016x}{:016x}", r.next_u64(), r.next_u64())
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use crate::config::{AgentKind, AgentSpec, Graph};

    use super::*;

    /// A store of its own, with no process-global state - which is the point of
    /// [`Chats::at`], and why these can run in parallel.
    fn store() -> (tempfile::TempDir, Chats) {
        let tmp = tempfile::tempdir().expect("tempdir");
        let chats = Chats::at(tmp.path().join("chats"));
        (tmp, chats)
    }

    /// A task file of the shape [`plan::TASK_FILE_SPEC`] describes, long enough
    /// that length is not one of the problems under test.
    fn good_draft() -> String {
        "# Report per-node durations in `magi show`\n\
         \n\
         ## Context\n\
         \n\
         `magi show` prints a run's nodes but not how long any of them took, so \
         the operator cannot see which seat is expensive. The data is already \
         in `run.events`.\n\
         \n\
         ## Change\n\
         \n\
         Add a duration column to the node table in `src/report.rs`.\n\
         \n\
         ## Constraints\n\
         \n\
         Do not change the JSON shape of a run record.\n\
         \n\
         ## Completion criteria\n\
         \n\
         - [ ] `magi show <run>` prints a duration for every completed node.\n\
         - [ ] A node with no end event prints nothing rather than zero.\n\
         \n\
         ## Out of scope\n\
         \n\
         The TUI's detail pane.\n"
            .to_owned()
    }

    /// A `kind = "command"` agent whose whole behaviour is a POSIX shell
    /// script. No test in this module may spawn a real agent CLI: they are the
    /// operator's paid subscriptions, they reach the network, and they are not
    /// installed on CI.
    fn mock_agent(dir: &Path, script: &str, env: BTreeMap<String, String>) -> AgentSpec {
        let path = dir.join("mock-chat-agent.sh");
        std::fs::write(&path, script).expect("write mock");
        AgentSpec {
            id: "mock".to_owned(),
            kind: AgentKind::Command,
            model: None,
            command: vec!["sh".to_owned(), path.to_string_lossy().into_owned()],
            extra_args: Vec::new(),
            env,
            prompt_delivery: None,
        }
    }

    /// A config whose only agent is `spec`, with the graph left at its
    /// defaults except for the language, so `language_note` stays out of the
    /// prompt assertions.
    fn config(spec: AgentSpec) -> Config {
        Config {
            agents: vec![spec],
            graph: Graph {
                language: "en".to_owned(),
                ..Graph::default()
            },
            ..Config::default()
        }
    }

    /// Echo a canned reply, ignoring the prompt on stdin.
    const REPLY: &str = "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' \"$MOCK_REPLY\"\n";

    /// Say nothing and fail, the way a CLI that cannot start does.
    const BROKEN: &str = "#!/bin/sh\ncat >/dev/null\nexit 3\n";

    /// Reply with the prompt it was given, so a test can inspect exactly what
    /// the leader received on stdin.
    const ECHO: &str = "#!/bin/sh\ncat\n";

    fn env(reply: &str) -> BTreeMap<String, String> {
        BTreeMap::from([("MOCK_REPLY".to_owned(), reply.to_owned())])
    }

    #[test]
    fn the_frozen_json_field_names_round_trip_through_disk() {
        let (tmp, chats) = store();
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ab12".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "sonnet".to_owned(),
            status: ChatStatus::Open,
            turns: vec![Turn {
                who: Who::Operator,
                body: "rework the config loader".to_owned(),
                at: Timestamp::now(),
                attachments: Vec::new(),
            }],
            draft: None,
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "sonnet", 7),
        };
        chats.put(&mut chat).expect("put");

        // Asserted literally, against the text on disk. The web UI is written
        // against these names by hand, so a rename that only round-trips
        // through serde would break the phone silently.
        let raw = std::fs::read_to_string(chats.path_of(&chat.id)).expect("read back");
        let v: serde_json::Value = serde_json::from_str(&raw).expect("parse");
        for field in [
            "schema",
            "id",
            "repo",
            "from",
            "agent",
            "status",
            "turns",
            "draft",
            "task",
            "created_at",
            "updated_at",
        ] {
            assert!(v.get(field).is_some(), "missing field `{field}`");
        }
        assert_eq!(v["schema"], 1);
        assert_eq!(v["status"], "open");
        assert_eq!(v["turns"][0]["who"], "operator");
        assert_eq!(v["turns"][0]["body"], "rework the config loader");
        assert!(v["turns"][0].get("at").is_some());
        assert!(v["turns"][0].get("attachments").is_some());
        assert!(v["draft"].is_null());
        assert!(v["task"].is_null());
        assert!(v["from"].is_null());

        let back = chats.get(&chat.id).expect("get");
        assert_eq!(back.id, chat.id);
        assert_eq!(back.turns, chat.turns);
        assert_eq!(back.status, ChatStatus::Open);
        assert_eq!(back.from, None);
    }

    /// A conversation recorded before `from` existed must still read: the
    /// `#[serde(deny_unknown_fields)]` on [`Chat`] would otherwise make this
    /// field's addition a breaking change for every chat already on disk.
    #[test]
    fn a_chat_recorded_without_a_from_field_still_reads() {
        let (tmp, chats) = store();
        let path = chats.path_of("20260903-014455-ab12");
        std::fs::create_dir_all(chats.root()).expect("chats dir");
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": SCHEMA,
                "id": "20260903-014455-ab12",
                "repo": tmp.path(),
                "agent": "sonnet",
                "status": "open",
                "turns": [],
                "draft": null,
                "task": null,
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
                "seat": SeatState::new(SEAT, "sonnet", 7),
            })
            .to_string(),
        )
        .expect("write pre-`from` chat");

        let chat = chats.get("20260903-014455-ab12").expect("must still read");
        assert_eq!(chat.from, None);
    }

    /// A conversation recorded before attachments existed - schema 1, no
    /// `attachments` key on any turn - must still read, the same guarantee
    /// `a_chat_recorded_without_a_from_field_still_reads` gives `from`.
    #[test]
    fn a_chat_recorded_without_attachments_still_reads() {
        let (tmp, chats) = store();
        let path = chats.path_of("20260903-014455-ab12");
        std::fs::create_dir_all(chats.root()).expect("chats dir");
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": 1,
                "id": "20260903-014455-ab12",
                "repo": tmp.path(),
                "agent": "sonnet",
                "status": "open",
                "turns": [
                    { "who": "operator", "body": "rework the config loader",
                      "at": Timestamp::now().to_string() },
                ],
                "draft": null,
                "task": null,
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
                "seat": SeatState::new(SEAT, "sonnet", 7),
            })
            .to_string(),
        )
        .expect("write pre-attachments chat");

        let chat = chats.get("20260903-014455-ab12").expect("must still read");
        assert!(chat.turns[0].attachments.is_empty());
    }

    #[test]
    fn derived_background_names_the_source_repository_and_carries_the_transcript() {
        let (_tmp, chats) = store();
        let chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ab12".to_owned(),
            repo: PathBuf::from("/repo/other"),
            from: None,
            agent: "sonnet".to_owned(),
            status: ChatStatus::Open,
            turns: vec![
                Turn {
                    who: Who::Operator,
                    body: "rework the queue drain".to_owned(),
                    at: Timestamp::now(),
                    attachments: Vec::new(),
                },
                Turn {
                    who: Who::Agent,
                    body: "which part of the drain?".to_owned(),
                    at: Timestamp::now(),
                    attachments: Vec::new(),
                },
            ],
            draft: None,
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "sonnet", 7),
        };
        let background = derived_background(&chat, &chats);
        assert!(background.contains("/repo/other"));
        assert!(background.contains("rework the queue drain"));
        assert!(background.contains("which part of the drain?"));
        assert!(background.contains("different"));
    }

    #[tokio::test]
    async fn starting_a_derived_chat_carries_the_source_transcript_and_leaves_it_untouched() {
        let (tmp, chats) = store();
        let source_spec = mock_agent(tmp.path(), REPLY, env("which module?"));
        let source_cfg = config(source_spec);
        let source = start(
            &chats,
            &source_cfg,
            tmp.path().to_owned(),
            "rework the queue drain",
            None,
            None,
        )
        .await
        .expect("start source");
        let before = source.clone();

        let other_repo = tmp.path().join("other-repo");
        std::fs::create_dir_all(&other_repo).expect("other repo dir");
        // Overwrites the script `source_spec` pointed at: the source's own
        // turn already ran, so only the derived chat's invocation sees this.
        let echo_spec = mock_agent(tmp.path(), ECHO, BTreeMap::new());
        let derived_cfg = config(echo_spec);
        let derived = start(
            &chats,
            &derived_cfg,
            other_repo,
            "same idea, different repository",
            None,
            Some(&source),
        )
        .await
        .expect("start derived");

        assert_eq!(derived.from.as_deref(), Some(source.id.as_str()));

        let prompt = &derived.turns.last().expect("agent reply").body;
        assert!(prompt.contains("Background: derived from another conversation"));
        assert!(prompt.contains(&source.repo.display().to_string()));
        assert!(prompt.contains("rework the queue drain"));
        assert!(prompt.contains("same idea, different repository"));

        // Deriving a chat must not touch the one it came from.
        let reread = chats.get(&source.id).expect("source still on disk");
        assert_eq!(reread.status, before.status);
        assert_eq!(reread.turns, before.turns);
        assert_eq!(reread.draft, before.draft);
    }

    /// [`build`] must not write the record anywhere: `chat_post` claims
    /// [`crate::web::Ui::begin_turn`] on `chat.id` between calling this and
    /// persisting it, and that ordering only closes the race it exists for
    /// (see `chat_post`'s doc) if nothing observable exists yet for anyone
    /// else to resolve, claim or record into ahead of the claim.
    #[test]
    fn build_constructs_the_record_without_writing_it_anywhere() {
        let (tmp, chats) = store();
        let spec = mock_agent(tmp.path(), REPLY, BTreeMap::new());
        let cfg = config(spec);

        let chat = build(
            &cfg,
            tmp.path().to_owned(),
            "rework the config loader",
            None,
            None,
        )
        .expect("build");

        assert!(
            !chats.path_of(&chat.id).is_file(),
            "build must not touch the filesystem"
        );
        assert!(
            chats.list().is_empty(),
            "no record must be resolvable until something calls `Chats::put`"
        );
    }

    /// `roles.chatter`, not `roles.planner`, decides who answers this
    /// conversation - and when `chatter` is unset, `planner` still does, so
    /// an operator who only ever named a planner sees no change. This is the
    /// distinction the resident chat's timeout under a triple-booked `opus`
    /// (planner, chatter, and a judge seat all at once) turned up.
    #[tokio::test]
    async fn a_chat_prefers_the_chatter_role_over_the_planner_role() {
        let (tmp, chats) = store();
        let planner_spec = mock_agent(tmp.path(), REPLY, env("from the planner"));
        let mut chatter_spec = mock_agent(tmp.path(), REPLY, env("from the chatter"));
        chatter_spec.id = "chatter-mock".to_owned();

        let mut cfg = Config {
            agents: vec![planner_spec.clone(), chatter_spec.clone()],
            graph: Graph {
                language: "en".to_owned(),
                ..Graph::default()
            },
            ..Config::default()
        };
        cfg.roles.planner = Some(planner_spec.id.clone());
        cfg.roles.chatter = Some(chatter_spec.id.clone());

        let chat = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "rework the drain",
            None,
            None,
        )
        .await
        .expect("start with chatter set");
        assert_eq!(chat.agent, chatter_spec.id, "chatter must win over planner");

        cfg.roles.chatter = None;
        let fallback = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "rework the drain again",
            None,
            None,
        )
        .await
        .expect("start with chatter unset");
        assert_eq!(
            fallback.agent, planner_spec.id,
            "unset chatter must fall back to planner, unchanged from before this role existed"
        );
    }

    #[test]
    fn extract_draft_takes_the_last_task_block_and_ignores_other_fences() {
        let reply = "here is a sketch\n\
                     \n\
                     ```rust\n\
                     fn not_the_draft() {}\n\
                     ```\n\
                     \n\
                     ```task\n\
                     # first version\n\
                     ```\n\
                     \n\
                     ```json\n\
                     {\"also\": \"not it\"}\n\
                     ```\n\
                     \n\
                     revised:\n\
                     \n\
                     ```task\n\
                     # second version\n\
                     ## Completion criteria\n\
                     ```\n";
        assert_eq!(
            extract_draft(reply).as_deref(),
            Some("# second version\n## Completion criteria\n")
        );
    }

    #[test]
    fn extract_draft_returns_none_when_there_is_no_task_block() {
        assert_eq!(extract_draft("which storage backend do you want?"), None);
        assert_eq!(extract_draft("```rust\nfn f() {}\n```\n"), None);
        // An empty block is not a draft: filing it would produce a task with
        // nothing in it.
        assert_eq!(extract_draft("```task\n```\n"), None);
    }

    #[tokio::test]
    async fn a_reply_with_no_draft_leaves_the_existing_draft_in_place() {
        let (tmp, chats) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("one more thing: which module?"));
        let cfg = config(spec);
        let mut chat = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "add durations",
            None,
            None,
        )
        .await
        .expect("start");
        chat.draft = Some(good_draft());
        chats.put(&mut chat).expect("put");

        say(&mut chat, &chats, &cfg, "the report module", Vec::new())
            .await
            .expect("say");

        assert_eq!(chat.draft.as_deref(), Some(good_draft().as_str()));
        assert_eq!(
            chats.get(&chat.id).expect("get").draft.as_deref(),
            Some(good_draft().as_str())
        );
    }

    #[test]
    fn the_briefing_carries_the_task_file_spec_and_the_task_fence() {
        let brief = briefing("rework the config loader", Path::new("/repo"));
        // The spec verbatim, so the shape asked for cannot drift from the shape
        // `plan::review_draft` enforces.
        assert!(brief.contains(plan::TASK_FILE_SPEC));
        assert!(brief.contains("```task"));
        assert!(brief.contains("rework the config loader"));
        assert!(brief.contains("/repo"));
        assert!(brief.contains("completion criteria"));
    }

    #[test]
    fn file_draft_refuses_a_bad_draft_with_every_problem() {
        let (tmp, chats) = store();
        let queue = Queue::at(tmp.path().join("queue"));
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ab12".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Open,
            turns: Vec::new(),
            // Short *and* missing completion criteria: both must be reported,
            // or the operator asks for one fix and gets refused again.
            draft: Some("# do the thing\n\nsome context.\n".to_owned()),
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        // `file_draft` re-reads its record from disk (see its own doc), so a
        // chat that only ever exists in memory in this test would fail with
        // "no chat matches" rather than the draft error under test.
        chats.put(&mut chat).expect("put");

        let problems = draft_problems(&chat).expect_err("a draft with no criteria is not fileable");
        assert!(
            problems.len() >= 2,
            "expected every problem, got {problems:?}"
        );
        assert!(problems.iter().any(|p| p.contains("completion criteria")));
        assert!(problems.iter().any(|p| p == plan::SHORT_DRAFT));

        let err = file_draft(&mut chat, &chats, &queue, 0)
            .expect_err("file_draft must refuse it too")
            .to_string();
        for p in &problems {
            assert!(err.contains(p.as_str()), "`{p}` missing from `{err}`");
        }
        assert_eq!(chat.status, ChatStatus::Open);
        assert!(chat.task.is_none());
        assert!(queue.list().is_empty());
    }

    #[test]
    fn file_draft_queues_a_good_draft_and_records_the_task() {
        let (tmp, chats) = store();
        let queue = Queue::at(tmp.path().join("queue"));
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-cd34".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Open,
            turns: Vec::new(),
            draft: Some(good_draft()),
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        // See the sibling test above for why this has to be on disk before
        // `file_draft` (which re-reads it) is called.
        chats.put(&mut chat).expect("put");

        let id = file_draft(&mut chat, &chats, &queue, 5).expect("file");

        assert_eq!(chat.status, ChatStatus::Filed);
        assert_eq!(chat.task.as_deref(), Some(id.as_str()));
        assert_eq!(
            chats.get(&chat.id).expect("get").task.as_deref(),
            Some(id.as_str()),
            "the task id must survive on disk, or the phone shows an unfiled chat"
        );

        let task = queue.get(&id).expect("queued task");
        assert_eq!(task.title, queue::title_from(&good_draft(), 72));
        assert_eq!(task.instruction, good_draft());
        assert_eq!(task.priority, 5);
        assert_eq!(task.source, Source::Human);
    }

    #[test]
    fn abandon_moves_an_open_chat_to_abandoned() {
        let (tmp, chats) = store();
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ab12".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Open,
            turns: Vec::new(),
            draft: None,
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        chats.put(&mut chat).expect("put");

        abandon(&mut chat, &chats).expect("abandon");

        assert_eq!(chat.status, ChatStatus::Abandoned);
        assert_eq!(
            chats.get(&chat.id).expect("get").status,
            ChatStatus::Abandoned
        );
    }

    #[test]
    fn abandoning_an_already_abandoned_chat_is_not_an_error() {
        let (tmp, chats) = store();
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ab13".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Abandoned,
            turns: Vec::new(),
            draft: None,
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        chats.put(&mut chat).expect("put");

        abandon(&mut chat, &chats).expect("abandoning twice is not an error");

        assert_eq!(chat.status, ChatStatus::Abandoned);
        assert_eq!(
            chats.get(&chat.id).expect("get").status,
            ChatStatus::Abandoned
        );
    }

    #[test]
    fn abandon_refuses_a_filed_chat_and_leaves_it_filed() {
        let (tmp, chats) = store();
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ab14".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Filed,
            turns: Vec::new(),
            draft: None,
            task: Some("some-task-id".to_owned()),
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        chats.put(&mut chat).expect("put");

        let err = abandon(&mut chat, &chats).expect_err("a filed chat refuses abandon");
        assert!(err.to_string().contains("filed"), "{err}");

        assert_eq!(
            chats.get(&chat.id).expect("get").status,
            ChatStatus::Filed,
            "a refused abandon must not touch the on-disk status"
        );
    }

    #[tokio::test]
    async fn an_abandon_that_lands_while_a_turn_is_in_flight_is_not_undone_by_the_reply() {
        let (tmp, chats) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("here you go"));
        let cfg = config(spec);
        // The in-flight turn's own handle: loaded once, the way a spawned
        // background task in `web::chat_say` holds one for the whole turn.
        let mut in_flight = open(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "add durations",
            None,
            None,
        )
        .expect("open");

        // The operator abandons the conversation through a *different* handle
        // while the turn above is still running - exactly what an abandon
        // typed on the phone while an agent is mid-answer looks like.
        let mut abandoned_elsewhere = chats.get(&in_flight.id).expect("reread");
        abandon(&mut abandoned_elsewhere, &chats).expect("abandon");
        assert_eq!(
            chats.get(&in_flight.id).expect("reread").status,
            ChatStatus::Abandoned,
            "the abandon landed on disk before the turn finished"
        );

        // The turn's own handle still says `open` - it was loaded before the
        // abandon - and finishing it must not resurrect the conversation the
        // operator already ended.
        assert_eq!(in_flight.status, ChatStatus::Open);
        first_turn(&mut in_flight, &chats, &cfg, None)
            .await
            .expect("the turn itself still completes");

        let on_disk = chats.get(&in_flight.id).expect("reread");
        assert_eq!(
            on_disk.status,
            ChatStatus::Abandoned,
            "an abandon must stick even when a turn that started before it finishes after it"
        );
        // The reply is not lost either: a turn already in flight when the
        // operator abandoned still gets its answer recorded.
        assert!(
            on_disk.turns.iter().any(|t| t.body == "here you go"),
            "the in-flight turn's own reply is still recorded: {:?}",
            on_disk.turns
        );
    }

    #[test]
    fn abandon_blocks_on_the_shared_guard_rather_than_interleaving_with_a_racing_writer() {
        let (tmp, chats) = store();
        let queue = Queue::at(tmp.path().join("queue"));
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ee15".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Open,
            turns: Vec::new(),
            draft: Some(good_draft()),
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        chats.put(&mut chat).expect("put");

        // Hold the same guard `file_draft`'s read-modify-write section holds
        // for the whole of its own read-then-write, standing in for
        // `file_draft` being paused between its read and its `put`.
        let held = chats.guard();

        let chats2 = chats.clone();
        let id = chat.id.clone();
        let abandoning = std::thread::spawn(move || {
            let mut chat = chats2.get(&id).expect("get");
            abandon(&mut chat, &chats2).expect("abandon");
        });

        std::thread::sleep(Duration::from_millis(50));
        assert!(
            !abandoning.is_finished(),
            "abandon must wait for the guard, not read and write while it is held - \
             a re-read alone narrows this window without closing it"
        );

        drop(held);
        abandoning.join().expect("abandon thread panicked");

        assert_eq!(
            chats.get(&chat.id).expect("reread").status,
            ChatStatus::Abandoned,
            "once the guard is free, abandon still lands"
        );
        assert!(queue.list().is_empty(), "file_draft never ran in this test");
    }

    /// The gap the in-process guard cannot see: `magi plan --abandon` opens
    /// its own `Chats`, with its own `Arc<Mutex<()>>` wired to nothing this
    /// process holds. Holding a claim directly - rather than the guard - is
    /// what stands in for that separate process here, since `Chats::claim` is
    /// a `create_new` file on disk, indistinguishable to `abandon` from a
    /// second `magi web` or a second `magi plan --abandon` already inside its
    /// own read-modify-write section.
    #[test]
    fn abandon_is_refused_while_another_process_holds_the_chats_claim() {
        let (tmp, chats) = store();
        let mut chat = Chat {
            schema: SCHEMA,
            id: "20260903-014455-ee16".to_owned(),
            repo: tmp.path().to_owned(),
            from: None,
            agent: "mock".to_owned(),
            status: ChatStatus::Open,
            turns: Vec::new(),
            draft: None,
            task: None,
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "mock", 7),
        };
        chats.put(&mut chat).expect("put");

        let held = chats.claim(&chat.id).expect("claim");
        let err = abandon(&mut chat, &chats).expect_err("a claimed chat refuses abandon");
        assert!(err.to_string().contains("claimed"), "{err}");
        assert_eq!(
            chats.get(&chat.id).expect("reread").status,
            ChatStatus::Open,
            "a refused abandon must not touch the on-disk status"
        );

        drop(held);
        abandon(&mut chat, &chats).expect("abandon succeeds once the claim is released");
        assert_eq!(chat.status, ChatStatus::Abandoned);
    }

    #[tokio::test]
    async fn say_appends_the_operator_turn_then_the_agent_turn() {
        let (tmp, chats) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("which module?"));
        let cfg = config(spec);
        let mut chat = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "add durations",
            None,
            None,
        )
        .await
        .expect("start");
        // start is one operator turn (the idea) plus one agent turn.
        assert_eq!(chat.turns.len(), 2);
        assert_eq!(chat.turns[0].who, Who::Operator);
        assert_eq!(chat.turns[1].who, Who::Agent);

        say(&mut chat, &chats, &cfg, "the report module", Vec::new())
            .await
            .expect("say");

        assert_eq!(chat.turns.len(), 4);
        assert_eq!(chat.turns[2].who, Who::Operator);
        assert_eq!(chat.turns[2].body, "the report module");
        assert_eq!(chat.turns[3].who, Who::Agent);
        assert_eq!(chat.turns[3].body, "which module?");
        assert_eq!(chats.get(&chat.id).expect("get").turns, chat.turns);
    }

    #[tokio::test]
    async fn a_failed_turn_keeps_the_operator_message_and_says_what_happened() {
        let (tmp, chats) = store();
        let good = mock_agent(tmp.path(), REPLY, env("which module?"));
        let cfg = config(good);
        let mut chat = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "add durations",
            None,
            None,
        )
        .await
        .expect("start");

        // The chat is bound to roster agent `mock`, so break what `mock`
        // actually runs: `mock_agent` rewrites the same script path, which is
        // what it looks like when that CLI stops working mid-interview.
        mock_agent(tmp.path(), BROKEN, BTreeMap::new());
        let err = say(&mut chat, &chats, &cfg, "the report module", Vec::new())
            .await
            .expect_err("a turn with no answer is an error");
        assert!(err.to_string().contains("no answer"), "{err}");

        let on_disk = chats.get(&chat.id).expect("get");
        assert_eq!(on_disk.turns.len(), 4);
        assert_eq!(
            on_disk.turns[2].body, "the report module",
            "the operator's message must survive the failure"
        );
        let note = &on_disk.turns[3];
        assert_eq!(note.who, Who::Agent);
        assert!(
            note.body.starts_with(MAGI_NOTE),
            "the failure must be visible in the transcript: {}",
            note.body
        );
        assert!(note.body.contains("your message is saved"));
    }

    /// An attachment lets the operator send an otherwise-empty message, and
    /// its absolute path (never the id or the operator's own filename alone)
    /// is what actually reaches the agent's prompt - the whole point of
    /// `Invocation::attachments` and the note `turn` appends.
    #[tokio::test]
    async fn attachments_reach_the_prompt_and_an_empty_body_is_still_a_turn() {
        let (tmp, chats) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("first reply"));
        let cfg = config(spec);
        let mut chat = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "add durations",
            None,
            None,
        )
        .await
        .expect("start");

        let att = chats
            .put_attachment(
                &chat.id,
                "image/png",
                "screenshot.png",
                b"pretend-png-bytes",
            )
            .expect("put attachment");

        // Overwrites the script `mock_agent` above pointed at, the same trick
        // `starting_a_derived_chat_carries_the_source_transcript_and_leaves_it_untouched`
        // uses: the reply becomes whatever the agent received on stdin.
        mock_agent(tmp.path(), ECHO, BTreeMap::new());
        say(&mut chat, &chats, &cfg, "", vec![att.clone()])
            .await
            .expect("an empty body with an attachment is still a turn");

        let operator_turn = &chat.turns[chat.turns.len() - 2];
        assert_eq!(operator_turn.who, Who::Operator);
        assert_eq!(operator_turn.body, "");
        assert_eq!(operator_turn.attachments, vec![att.clone()]);

        let prompt = &chat.turns.last().expect("agent reply").body;
        let expected_path = chats
            .attachments_dir(&chat.id)
            .join(format!("{}.png", att.id));
        assert!(
            prompt.contains(&expected_path.display().to_string()),
            "the agent must be told the attachment's absolute path: {prompt}"
        );
        assert!(prompt.contains("image/png"), "and its mime: {prompt}");
    }

    /// `run::home()` returns a bare relative `PathBuf` verbatim when the
    /// operator sets `MAGI_HOME` to a relative path - nothing canonicalizes
    /// it - so a `Chats` store built on top of it has a relative `root` too.
    /// That is harmless for this store's own reads and writes, which run in
    /// this same process against this process's cwd, but `attachment_path`
    /// hands its result to a *different* process invoked with `cwd:
    /// &chat.repo`: an uncorrected relative path would resolve against the
    /// repository instead of wherever the attachment actually landed, and
    /// the CLI would find nothing there.
    #[test]
    fn attachment_path_is_absolute_even_when_the_store_root_is_relative() {
        let chats = Chats::at(PathBuf::from("relative-chats-root-for-this-test"));
        let att = Attachment {
            id: "0".repeat(32),
            name: "shot.png".to_owned(),
            mime: "image/png".to_owned(),
            bytes: 3,
        };
        let path = chats
            .attachment_path("some-chat-id", &att)
            .expect("a supported mime always yields a path");
        assert!(
            path.is_absolute(),
            "must be absolute even off a relative store root: {}",
            path.display()
        );
    }

    #[tokio::test]
    async fn a_turn_past_the_configured_chat_timeout_is_reported_with_that_timeout() {
        // `[graph] timeout_chat` must be the number this module actually
        // waits, not a leftover hardcoded five minutes - so the mock sleeps
        // past a deliberately tiny override and the failure note is checked
        // against that same override, not the old default.
        let (tmp, chats) = store();
        let slow = mock_agent(
            tmp.path(),
            "#!/bin/sh\ncat >/dev/null\nsleep 2\n",
            BTreeMap::new(),
        );
        let mut cfg = config(slow);
        cfg.graph.timeout_chat = 1;

        let err = start(
            &chats,
            &cfg,
            tmp.path().to_owned(),
            "add durations",
            None,
            None,
        )
        .await
        .expect_err("a turn that never answers is an error");
        assert!(
            err.to_string().contains("did not answer within 1s"),
            "{err}"
        );

        let on_disk = chats.list();
        let chat = &on_disk[0];
        let note = chat.turns.last().expect("a note turn was recorded");
        assert!(
            note.body.contains("did not answer within 1s"),
            "the transcript must show the configured timeout: {}",
            note.body
        );
    }

    #[test]
    fn list_puts_open_chats_before_filed_ones() {
        let (tmp, chats) = store();
        let make = |id: &str, status: ChatStatus| {
            let mut c = Chat {
                schema: SCHEMA,
                id: id.to_owned(),
                repo: tmp.path().to_owned(),
                from: None,
                agent: "mock".to_owned(),
                status,
                turns: Vec::new(),
                draft: None,
                task: None,
                created_at: Timestamp::now(),
                updated_at: Timestamp::now(),
                seat: SeatState::new(SEAT, "mock", 7),
            };
            chats.put(&mut c).expect("put");
        };
        // The filed one is newest, so ordering by id alone would put it first.
        make("20260901-000000-0001", ChatStatus::Open);
        make("20260902-000000-0002", ChatStatus::Open);
        make("20260903-000000-0003", ChatStatus::Filed);

        let ids: Vec<String> = chats.list().into_iter().map(|c| c.id).collect();
        assert_eq!(
            ids,
            [
                "20260902-000000-0002",
                "20260901-000000-0001",
                "20260903-000000-0003"
            ]
        );
        assert_eq!(chats.count_open(), 2);
    }
}