//! The Claude Code driver: `claude -p` speaking stream-json on stdio,
//! translated into the common event model.
//!
//! This half owns the process -- starting it, *adopting one this server
//! left running*, writing lines to it, and ending it. Where it runs is
//! `session::transport`'s business, not this file's: this one emits a
//! `Launch` and never learns whether it became a local child or an ssh
//! invocation.
//!
//! The process is meant to outlive the server, so that restarting the
//! backend does not end a turn: its stdio lives in the session directory
//! (a fifo it holds open itself, plus logs read from a byte offset) and
//! `session::process` records what it is. Everything comes through
//! [`ClaudeDriver::launch`], which adopts if it can and starts if it
//! cannot -- `--resume` is reachable only on the second path, because two
//! CLIs on one session file duplicate the conversation into it.
//! Turning a line into [`Event`]s is [`translate`], which changes when the
//! CLI's wire format does rather than when any of the above does.
//!
//! The probing record below stays here, since it is the provenance for
//! both halves: the flags are this file's, the message catalogue is what
//! `translate` implements.
//!
//! Wire format pinned against CLI 2.1.237 by probing (2026-08-24; scripts
//! summarized here since they live outside the repo):
//!
//! - Outbound: `system/init` (carries `session_id`, the `--resume` token),
//!   `stream_event` (raw API deltas; `text_delta` is the streaming text),
//!   consolidated `assistant` messages (their `tool_use` blocks have the
//!   complete input), `user` messages with `tool_result` blocks, a `result`
//!   per turn (usage + cost), `control_request` for anything needing a
//!   human, `control_response` answering ours.
//! - Permission prompts require the hidden `--permission-prompt-tool stdio`
//!   flag; they arrive as `control_request{subtype:can_use_tool}` and are
//!   answered with `{behavior:"allow",updatedInput}` or
//!   `{behavior:"deny",message}`. `AskUserQuestion` uses the same shape,
//!   with the chosen labels added to `updatedInput` as
//!   `answers:{<question text>:<label>}`.
//! - Inbound `user` messages sent mid-turn are queued and injected at the
//!   next tool boundary (verified live: the model acknowledged a steer
//!   between two Bash calls) -- the behavior this app exists for.
//! - `control_request{subtype:set_model}` answers success;
//!   `{subtype:interrupt}` stops the turn.
//! - `control_request{subtype:set_permission_mode}` answers success and
//!   echoes the mode back (`{"response":{"mode":"acceptEdits"}}`), so the
//!   mode is changeable mid-session rather than only at spawn. Probed the
//!   same way as the rest, against 2.1.237 on 2026-08-29.

use std::collections::VecDeque;
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use anyhow::{Context, Result};
use serde_json::{Value, json};
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;

use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus};
use super::process;
use super::transport::{Launch, Streams, Transport};
use crate::config::{ProviderConfig, SessionConfig};
use translate::{AnswerOutcome, Setting, Translator, starts_a_model_call};

/// How much of a failing process's stderr the exit report carries.
///
/// Enough for a shell's complaint plus the context it prints around it --
/// fish's `cd` failure is seven lines including a caret pointing at the
/// offending line -- and bounded because this is held per session for the
/// life of the process and a chatty program would otherwise grow without
/// limit.
const STDERR_LINES_KEPT: usize = 50;

/// The kept stderr as one block, with blank lines trimmed off both ends.
///
/// The trailing trim is the point: a shell's error ends with a blank line,
/// so the last line of stderr is routinely empty and anything that reports
/// "the last line" reports nothing at all.
fn tail_of(kept: &VecDeque<String>) -> String {
    let lines: Vec<&str> = kept.iter().map(String::as_str).collect();
    let start = lines
        .iter()
        .position(|line| !line.trim().is_empty())
        .unwrap_or(lines.len());
    let end = lines
        .iter()
        .rposition(|line| !line.trim().is_empty())
        .map(|last| last + 1)
        .unwrap_or(start);
    lines[start..end].join("\n")
}

/// Where the driver remembers its CLI session id between backend runs --
/// the whole crash-recovery story: respawning with `--resume <id>` picks
/// the conversation back up from Claude's own session files. Kept in the
/// session directory rather than config.ron so the shared schema stays
/// free of per-driver state.
pub(super) mod translate;

const RESUME_FILE: &str = "claude-session.json";

/// Messages handed to the CLI that it has not visibly acted on yet.
///
/// The CLI *does* take a message written mid-turn: it goes into the next
/// model call, which is the next tool boundary, and the whole point of
/// this app is steering a turn that is already running. An earlier version
/// of this file claimed the opposite and held every mid-turn message until
/// the turn ended -- so a steer sent after the second tool call sat unread
/// until all the work it was meant to redirect had finished. Measured
/// rather than argued: a line written between two Bash calls was answered
/// inside the same turn, with one `result` for the whole thing.
///
/// What the CLI does not do is say on stdout that it has read one. So the
/// line goes out immediately and the *announcement* waits here instead,
/// until the CLI opens the next model call -- see
/// [`translate::starts_a_model_call`]. That keeps a phone's held bubble
/// where it belongs -- below the working indicator until the session has
/// actually taken it -- without delaying the message itself to get it.
///
/// The proof has to be the model call and not the output, which is what
/// an earlier version took it to be. Assistant text and a tool call both
/// keep arriving from a message that was *already in flight* when the
/// steer was written, and that message saw none of it: a steer sent
/// while an answer was streaming was recorded in the middle of it, above
/// tool calls the model had already committed to. On screen the answer
/// split into two bubbles around a message it had not read, and the tool
/// results that followed read as things the steer had asked for.
#[derive(Default)]
struct Queue {
    /// A turn is in flight, so a message sent now is a steer into it.
    running: bool,
    /// Written, not yet announced, oldest first, each with the id of the
    /// `MessageQueued` that told the phone it was waiting -- so the
    /// announcement can name which bubble it resolves.
    awaiting: VecDeque<(String, String, Vec<ImageRef>)>,
    /// The process is gone, so nothing can be taken up any more.
    ///
    /// Needed because every other way out of a turn is an `Idle` this
    /// driver sees, and an exit is the one that is not. Without it a
    /// process that died mid-turn left `running` true for good, and since
    /// a message is only recorded when it is *announced*, each later one
    /// vanished with nothing on screen to say it had not been delivered.
    closed: bool,
}

impl Queue {
    /// Gives up on everything held, because the process is gone.
    ///
    /// Reported rather than dropped. These are messages somebody typed
    /// that never reached the session and never reached the transcript, so
    /// this is the only place they can be mentioned at all.
    fn close(&mut self, sink: &EventSink, why: &str) {
        self.closed = true;
        self.running = false;
        let lost: Vec<String> = self.awaiting.drain(..).map(|(_, text, _)| text).collect();
        if lost.is_empty() {
            return;
        }
        let _ = sink.send(Event::Error {
            message: format!(
                "{why} before it read {}: {}",
                if lost.len() == 1 {
                    "this message".to_string()
                } else {
                    format!("{} queued messages", lost.len())
                },
                lost.join(" / ")
            ),
        });
    }
}

/// The session directory's copies of the process's standard streams.
///
/// Named once rather than built at each use, because the spawn path and
/// the attach path must agree about which file is which; if they drift,
/// a reattached session reads a file nothing is writing and simply looks
/// idle forever.
const STDIN_FIFO: &str = "stdin.fifo";
const STDOUT_LOG: &str = "stdout.log";
const STDERR_LOG: &str = "stderr.log";

/// How often a reader with nothing to read looks again.
///
/// A poll rather than a watch: the alternative is an inotify dependency
/// for one file per session, and at this interval the streaming text is
/// already arriving faster than a phone renders it.
const POLL: std::time::Duration = std::time::Duration::from_millis(50);

pub struct ClaudeDriver {
    sink: EventSink,
    queue: Arc<Mutex<Queue>>,
    /// Lines for the process's stdin.
    ///
    /// Not closeable, unlike the pipe this used to be: stdin is a fifo the
    /// process holds open itself, so closing this end says nothing to it.
    /// Ending the process is [`Driver::stop`]'s job and it uses a signal.
    to_child: mpsc::UnboundedSender<String>,
    state: Arc<Mutex<Translator>>,
    session_dir: PathBuf,
    /// Cleared to stop the reader without touching the process -- which is
    /// exactly what detaching is.
    reading: Arc<AtomicBool>,
}

impl ClaudeDriver {
    /// Takes charge of this session's process: the one already running if
    /// there is one, otherwise a new one.
    ///
    /// One entry point rather than two, because the choice is not the
    /// caller's to make and getting it wrong is the expensive bug. A
    /// second `--resume` against a session file that is already open
    /// duplicates the whole conversation into it and bills the reattached
    /// copy for re-reading it -- measured at 65 MB and 154 screenshots on
    /// 2026-08-29, when an import of a *live* session did exactly this.
    /// So `--resume` is reachable only through the spawn half below, under
    /// a check that nothing is running.
    pub fn launch(
        meta: &SessionConfig,
        provider: &ProviderConfig,
        transport: &Transport,
        session_dir: &Path,
        sink: EventSink,
    ) -> Result<Self> {
        let state = Arc::new(Mutex::new(Translator::new(session_dir.to_path_buf())));
        let queue = Arc::new(Mutex::new(Queue::default()));
        let reading = Arc::new(AtomicBool::new(true));

        // Adopting is only possible for a process this server left behind
        // on this machine: an ssh session's child is at the far end of a
        // connection that died with the server, so there is nothing there
        // to find. Nothing was ever recorded for one, so this answers "no"
        // without needing to know that, which is why the remote case is
        // not a branch here.
        // Whether this launch *started* a process or picked up one that was
        // already there. The two owe the session different things -- see the
        // `Status` below.
        let started_here;
        let record = match process::recorded(session_dir) {
            // Still running, and ours. Pick it up where it was left --
            // the one path that must not pass `--resume`.
            Some((record, process::Liveness::Alive)) => {
                tracing::info!(
                    "session {} reattaching to the {} it left running (pid {})",
                    meta.id,
                    provider.name,
                    record.pid
                );
                started_here = false;
                record
            }
            // Recorded, and the machine will not say whether it is still
            // there. Starting one anyway is the mistake this module is
            // for, so nothing is started; `follow` keeps asking and
            // reports the state as unknown until it gets an answer.
            Some((record, process::Liveness::Unknown)) => {
                tracing::warn!(
                    "session {} recorded pid {} but this machine won't say whether it is running; \
                     not starting a second one",
                    meta.id,
                    record.pid
                );
                started_here = false;
                record
            }
            Some((_, process::Liveness::Dead)) | None => {
                started_here = true;
                Self::start(meta, provider, transport, session_dir)?
            }
        };

        // A process this driver has just started has been asked for nothing,
        // which is what idle means. Said here because nothing else will say
        // it: the CLI writes not one line until it is given work, so a
        // session whose transcript last recorded `Exited` -- one whose
        // process died while this server was down, or one somebody stopped
        // from the phone -- would keep that word. `Exited` refuses every
        // command sent to the session, and it offers a phone the chance to
        // start a second CLI against a conversation that already has one.
        //
        // From the driver rather than from the manager, and before `follow`
        // is spawned, so it cannot overtake the exit `follow` reports for a
        // process that dies immediately: both come from here, in this order.
        // Adopting says nothing, because a process that was already running
        // may be mid-turn, and the transcript's last word is the better
        // answer until its output says otherwise.
        //
        // The llama driver has always done this (see `LlamaDriver::attached`,
        // which reports `Running` while the model loads and `Idle` when it
        // answers); this side was the one silent about it.
        if started_here {
            let _ = sink.send(Event::Status {
                state: SessionStatus::Idle,
            });
        }
        // Where reading of its output had reached. A process just started
        // has said nothing, so its record says zero and this is the same
        // question with the same answer.
        let resuming_from = match record.detail {
            process::Detail::Stdio { stdout_read } => stdout_read,
            // A record of the wrong shape belongs to a different driver;
            // read its output from the start rather than trusting an
            // offset into a file that means something else.
            _ => 0,
        };

        // The writer end of the fifo. Opened write-only here: the process
        // holds its own read-write handle, so this side coming and going
        // across a restart is invisible to it.
        let stdin = std::fs::OpenOptions::new()
            .write(true)
            .open(session_dir.join(STDIN_FIFO))
            .with_context(|| format!("opening {STDIN_FIFO} for session {}", meta.id))?;
        let (to_child, mut from_driver) = mpsc::unbounded_channel::<String>();
        tokio::spawn(async move {
            let mut stdin = tokio::fs::File::from_std(stdin);
            while let Some(line) = from_driver.recv().await {
                if stdin.write_all(line.as_bytes()).await.is_err()
                    || stdin.write_all(b"\n").await.is_err()
                    || stdin.flush().await.is_err()
                {
                    break;
                }
            }
        });

        tokio::spawn(follow(
            session_dir.to_path_buf(),
            record,
            resuming_from,
            Arc::clone(&state),
            sink.clone(),
            Arc::clone(&queue),
            Arc::clone(&reading),
            format!("{} {}", provider.name, transport.describe()),
        ));

        Ok(Self {
            sink,
            queue,
            to_child,
            state,
            session_dir: session_dir.to_path_buf(),
            reading,
        })
    }

    /// Starts a new CLI for this session, with its streams in the session
    /// directory so the next run of this server can find them.
    ///
    /// The only path that passes `--resume`, and it is reached only when
    /// nothing is running -- see [`ClaudeDriver::launch`].
    fn start(
        meta: &SessionConfig,
        provider: &ProviderConfig,
        transport: &Transport,
        session_dir: &Path,
    ) -> Result<process::Record> {
        let mut args: Vec<String> = ["-p", "--verbose"].iter().map(|a| a.to_string()).collect();
        let mut push = |flag: &str, value: &str| {
            args.push(flag.to_string());
            args.push(value.to_string());
        };
        push("--input-format", "stream-json");
        push("--output-format", "stream-json");
        // Hidden but load-bearing: without it the CLI resolves permissions
        // itself and nothing ever reaches the phone.
        push("--permission-prompt-tool", "stdio");
        if let Some(model) = &meta.model {
            push("--model", model);
        }
        if let Some(mode) = &meta.permission_mode {
            push("--permission-mode", mode);
        }
        // Named at birth, so this session is the same session in the CLI's
        // own picker and in what other agents see when they list it.
        //
        // Only when we are the ones creating it. A resume is a session
        // that already existed -- an import, or this server starting again
        // -- and it already has whatever name it was given, quite possibly
        // by the person who was typing in it. Renaming that from a title
        // we derived from its first message would be taking something the
        // app was only ever shown. `Driver::set_title` is how it changes
        // after this point, and that one is asked for.
        match read_resume_token(session_dir) {
            Some(resume) => push("--resume", &resume),
            None => push("--name", &meta.title),
        }
        args.push("--include-partial-messages".to_string());
        // Makes `bypassPermissions` *reachable*, without selecting it: the
        // session still starts in whatever mode was asked for above, and
        // only moves if somebody moves it.
        //
        // Here because the CLI is asymmetric about that mode, which is not
        // obvious and cost a confused bug report. It will *launch* in
        // `bypassPermissions` on the strength of `--permission-mode` alone
        // -- so spawning straight into it from the phone has always worked
        // -- but it refuses to *switch* into it later:
        //
        //     Cannot set permission mode to bypassPermissions because the
        //     session was not launched with --dangerously-skip-permissions
        //
        // So the phone's own mode picker offered a mode that could not be
        // picked, on every session it had not been given at birth. Since
        // the mode is already reachable at spawn, this grants nothing that
        // was being withheld; it makes the two routes to it agree.
        //
        // Measured against 2.1.237, both ways round: without this flag the
        // control request comes back `subtype: error` with the message
        // above, and with it `subtype: success, mode: bypassPermissions`.
        // Note it is the `--allow-` form -- `--dangerously-skip-permissions`
        // is the one that turns it on for everything, and that would take
        // the choice away from whoever is holding the phone.
        args.push("--allow-dangerously-skip-permissions".to_string());

        // Fresh logs, because the offsets that index them start at zero
        // and everything the previous process said is already in the
        // transcript.
        let stdin = make_fifo(&session_dir.join(STDIN_FIFO))?;
        let stdout = create_log(&session_dir.join(STDOUT_LOG))?;
        let stderr = create_log(&session_dir.join(STDERR_LOG))?;

        let program = provider.command.as_deref().unwrap_or("claude");
        let launch = Launch::new(program, args, meta.cwd.as_deref());
        let child = transport.spawn(
            &launch,
            Streams::Detached {
                stdin: stdin.into(),
                stdout: stdout.into(),
                stderr: stderr.into(),
            },
        )?;
        let pid = child
            .id()
            .context("the process exited before it could be recorded")?;
        tracing::info!(
            "session {} running {program} {} as pid {pid}",
            meta.id,
            transport.describe()
        );

        // Reaped rather than waited on. This server is the parent, so
        // something has to collect the exit status or the process becomes
        // a zombie -- but it is `follow` that decides what the session is
        // doing, because after a restart there is no `Child` to wait on
        // and the answer has to come from the same place either way.
        tokio::spawn(async move {
            let mut child = child;
            let _ = child.wait().await;
        });

        let record = process::Record::of(pid, process::Detail::Stdio { stdout_read: 0 })
            .context("the process was gone before its start time could be read")?;
        process::write(session_dir, &record);
        Ok(record)
    }

    fn send_line(&self, line: String) {
        let _ = self.to_child.send(line);
    }

    /// Writes one of the CLI's own commands into the session.
    ///
    /// Slash commands ride the normal user-message channel -- there is no
    /// control request for them; `set_session_name` is not a subtype the
    /// CLI knows, measured by asking. The turn they start is marked here
    /// because they produce a `result` like any other, so a message sent
    /// meanwhile belongs in the queue's "written, announce when read"
    /// path rather than being reported as read the moment it is typed.
    ///
    /// Nothing is emitted about the command itself: the manager has
    /// already said it was sent, and the CLI announces what it does --
    /// saying so here would be this side's guess standing in for its
    /// measurement.
    fn local_command(&self, text: String) {
        let mut queue = self.queue.lock().unwrap();
        // The same check `send_user_message` makes, for the same reason: a
        // line written into a fifo nothing is reading goes nowhere and looks
        // exactly like one that arrived. `Commands::submit` refuses a
        // session already known to have exited, so what this catches is the
        // process going away between that check and this write.
        if queue.closed {
            drop(queue);
            let _ = self.sink.send(Event::Error {
                message: format!("this session's process has exited, so it can't run {text}"),
            });
            return;
        }
        queue.running = true;
        drop(queue);
        // The session is working from this moment, and until now nothing
        // said so: a command's reply carries no assistant text, so
        // `proves_a_turn` never saw it and the recorded status stayed idle
        // for the whole round trip -- which meant the *next* idle was not a
        // change, so nothing was ever released behind it.
        let _ = self.sink.send(Event::Status {
            state: SessionStatus::Running,
        });
        self.send_line(
            json!({"type": "user", "message": {"role": "user", "content": [
                {"type": "text", "text": text}
            ]}})
            .to_string(),
        );
    }

    /// Sends a control request, remembering what it asked for.
    ///
    /// `confirms` is the setting this request will have made if the CLI
    /// answers success -- see [`Translator::expect_setting`], and
    /// `Driver::set_model` for why a request is not a confirmation.
    /// `None` for the ones that change no setting, like an interrupt.
    ///
    /// The id is random rather than the clock it used to be: two requests
    /// in the same second shared an id, which was harmless while nothing
    /// looked one up and is not any more.
    fn send_control(&self, request: Value, confirms: Option<Setting>) {
        let id = format!("req-{}", super::random_hex());
        if let Some(setting) = confirms {
            // Before the line goes out: the reader thread is already
            // running, and a fast answer to a slow lock arrives first.
            self.state
                .lock()
                .unwrap()
                .expect_setting(id.clone(), setting);
        }
        self.send_line(
            json!({"type": "control_request", "request_id": id, "request": request}).to_string(),
        );
    }
}

impl Driver for ClaudeDriver {
    fn send_user_message(&self, text: String, images: Vec<ImageRef>) {
        let mut content = Vec::new();
        for id in &images {
            match attachment_block(&self.session_dir, id) {
                Ok(block) => content.push(block),
                Err(err) => {
                    let _ = self.sink.send(Event::Error {
                        message: format!("attachment {id} couldn't be sent: {err:#}"),
                    });
                }
            }
        }
        if !text.is_empty() {
            content.push(json!({"type": "text", "text": text}));
        }
        let line =
            json!({"type": "user", "message": {"role": "user", "content": content}}).to_string();
        let mut queue = self.queue.lock().unwrap();
        // Saying so beats writing into a fifo that nothing is reading,
        // which is what this used to do -- the message went nowhere and
        // looked exactly like one that had been delivered.
        if queue.closed {
            drop(queue);
            let _ = self.sink.send(Event::Error {
                message: "this session's process has exited, so it can't be sent anything"
                    .to_string(),
            });
            return;
        }
        if queue.running {
            // Into the running turn, now. Announced when the CLI shows it
            // has been round the model again -- see `Queue`.
            //
            // The *waiting* is recorded here, though, which is the one
            // thing that must not be left to the phone to remember: it put
            // the bubble on screen from its own state, so leaving the
            // session or restarting the app drew nothing pending while a
            // message was still in the queue.
            let id = super::random_hex();
            queue
                .awaiting
                .push_back((id.clone(), text.clone(), images.clone()));
            drop(queue);
            let _ = self.sink.send(Event::MessageQueued { id, text, images });
            self.send_line(line);
            return;
        }
        queue.running = true;
        drop(queue);
        // Nothing is in flight, so there is nothing to wait for: this
        // message *is* the turn about to start, and it never had a
        // `MessageQueued` to resolve.
        let _ = self.sink.send(Event::MessageTaken {
            id: None,
            text,
            images,
        });
        let _ = self.sink.send(Event::Status {
            state: SessionStatus::Running,
        });
        self.send_line(line);
    }

    fn answer_question(&self, id: &str, answers: &[String]) {
        let response = {
            let mut state = self.state.lock().unwrap();
            state.answer(id, answers)
        };
        match response {
            AnswerOutcome::Respond(control_response) => {
                let _ = self.sink.send(Event::Status {
                    state: SessionStatus::Running,
                });
                self.send_line(control_response.to_string());
            }
            // A multi-question AskUserQuestion still waiting on the rest.
            AnswerOutcome::Pending => {}
            AnswerOutcome::Unknown => {
                let _ = self.sink.send(Event::Error {
                    message: format!("no question {id} is awaiting an answer"),
                });
            }
        }
    }

    /// Anything queued behind the interrupted turn still goes: it was
    /// typed deliberately, and dropping it would lose a message that never
    /// reached the transcript, with nothing on screen to say so.
    fn interrupt(&self) {
        // Recorded before the request goes out, so the result it produces is
        // read as the stop somebody asked for rather than as a failure --
        // see `Translator::interrupting`.
        self.state.lock().unwrap().expect_interrupt();
        self.send_control(json!({"subtype": "interrupt"}), None);
    }

    fn set_permission_mode(&self, mode: &str) {
        self.send_control(
            json!({"subtype": "set_permission_mode", "mode": mode}),
            Some(Setting::PermissionMode(mode.to_string())),
        );
    }

    fn set_model(&self, model: &str) {
        self.send_control(
            json!({"subtype": "set_model", "model": model}),
            Some(Setting::Model(model.to_string())),
        );
    }

    fn run_command(&self, text: &str) {
        // Whatever the CLI's own vocabulary holds -- `/context`, `/usage`,
        // a command added after this was written. It rides the same
        // channel as `/compact` and `/rename` and starts a turn the same
        // way, so the same bookkeeping applies; what it means is the
        // CLI's business, not this file's.
        self.local_command(text.to_string());
    }

    fn set_title(&self, title: &str) {
        // The CLI's own mechanism, and a local command rather than a
        // control request -- `set_session_name` is not a subtype it
        // knows, measured by asking. It answers this the way it answers
        // `/compact`: a fresh `init`, then a `result` for a turn with no
        // model call in it, so the same "a turn is in flight" bookkeeping
        // applies. A name with a newline in it would be two lines and the
        // second would be a message, so it is refused rather than sent.
        if title.contains('\n') {
            let _ = self.sink.send(Event::Error {
                message: "a session name cannot contain a line break".to_string(),
            });
            return;
        }
        self.local_command(format!("/rename {title}"));
    }

    fn compact(&self) {
        self.local_command("/compact".to_string());
    }

    fn clear(&self) {
        // Nothing is emitted here on purpose: the transcript should
        // record a clear that happened, not one that was asked for. The
        // CLI announces it with a `conversation_reset` line, which
        // `translate.rs` turns into `Event::Cleared`, and follows it with
        // a fresh `init` whose new `session_id` the reader persists as
        // the resume token -- so the next launch resumes the cleared
        // conversation with nothing here to keep in step.
        self.local_command("/clear".to_string());
    }

    fn between_turns(&self) -> bool {
        let queue = self.queue.lock().unwrap();
        !queue.running && !queue.closed
    }

    fn detach(&self) {
        // Stop reading and leave everything else exactly as it is. The
        // process keeps its fifo (which it holds open itself), keeps
        // writing its log, and keeps its record -- which is how the next
        // run of this server finds it. See `Driver::detach`.
        self.reading.store(false, Ordering::SeqCst);
    }

    fn stop(&self) {
        self.reading.store(false, Ordering::SeqCst);
        if let Some(record) = process::live(&self.session_dir) {
            process::stop(&record, process::STOP_GRACE);
        }
        process::clear(&self.session_dir);
    }
}

/// Follows the process's stdout log, turning it into events, and is also
/// what decides whether the session is still running.
///
/// One loop rather than a reader plus a monitor. After a restart there is
/// no `Child` to wait on -- the process was reparented away from this
/// server -- so liveness has to be a question asked of the record either
/// way, and asking it in two places is how the two answers come to
/// disagree.
///
/// Reading is resumable because the position is written down with the
/// process (see [`process::Record`]): everything before it is already in
/// the transcript, so a server coming back picks up exactly where the last
/// one stopped and the conversation has no hole in it.
#[allow(clippy::too_many_arguments)]
async fn follow(
    session_dir: PathBuf,
    mut record: process::Record,
    mut offset: u64,
    state: Arc<Mutex<Translator>>,
    sink: EventSink,
    queue: Arc<Mutex<Queue>>,
    reading: Arc<AtomicBool>,
    label: String,
) {
    let stdout_path = session_dir.join(STDOUT_LOG);
    let stderr_path = session_dir.join(STDERR_LOG);
    // Whatever is already in the stderr log has been logged by whichever
    // run of this server was watching when it was written, so a reattach
    // starts at the end of it rather than repeating it. The tail is still
    // read from the file if the process dies, which is when it matters.
    let mut stderr_at = process::size_of(&stderr_path);
    let mut said_unknown = false;

    while reading.load(Ordering::SeqCst) {
        let (bytes, _) = match process::read_from(&stdout_path, offset) {
            Ok(found) => found,
            Err(err) => {
                // Reported, not only logged. This is the end of the
                // session's output as far as anyone watching is
                // concerned, and a phone told nothing shows a session
                // that is merely quiet -- indistinguishable from one
                // thinking. The status is `Unknown` rather than `Exited`
                // because the process may well still be running; what
                // has failed is this server's ability to hear it.
                tracing::error!("couldn't read {}: {err:#}", stdout_path.display());
                let _ = sink.send(Event::Error {
                    message: format!(
                        "lost track of {label}: its output can't be read ({err:#}). The process \
                         may still be running; restarting the backend will try to pick it up \
                         again."
                    ),
                });
                queue
                    .lock()
                    .unwrap()
                    .close(&sink, "this server lost track of the session");
                let _ = sink.send(Event::Status {
                    state: SessionStatus::Unknown,
                });
                return;
            }
        };
        // Only whole lines, and the offset stops at the last newline -- so a
        // line the process is halfway through writing is simply read again
        // next pass. Deliberately *not* held in memory between passes: the
        // offset would then have to point behind the bytes being held, and
        // the next read would return them a second time to be prepended to
        // the copy already there. It is also what makes the position
        // crash-safe, since it never claims a partial line was handled.
        //
        // Counted in bytes rather than on a decoded string: a read can cut
        // a multi-byte character in half, and the replacement character
        // that decoding puts there is a different length from what it
        // replaced -- which would slide the offset out of step with the
        // file for the rest of the session.
        let complete = complete_lines(&bytes);

        for line in String::from_utf8_lossy(&bytes[..complete]).lines() {
            if line.trim().is_empty() {
                continue;
            }
            if !translate_line(line, &session_dir, &state, &sink, &queue) {
                return; // session torn down
            }
        }
        if complete > 0 {
            offset += complete as u64;
            record.detail = process::Detail::Stdio {
                stdout_read: offset,
            };
            process::write(&session_dir, &record);
        }

        // Diagnostics only, and the tail of it is what an exit report
        // carries -- so it is read from the file rather than kept in
        // memory, which also means a reattached session can still explain
        // a failure it did not witness.
        if let Ok((bytes, at)) = process::read_from(&stderr_path, stderr_at)
            && at != stderr_at
        {
            stderr_at = at;
            for line in String::from_utf8_lossy(&bytes).lines() {
                if !line.trim().is_empty() {
                    tracing::warn!("{label} stderr: {line}");
                }
            }
        }

        match record.liveness() {
            process::Liveness::Alive => said_unknown = false,
            // Drain whatever it wrote on the way out before saying so.
            //
            // Progress, not "there were bytes": a process that died
            // mid-line leaves a partial one that is re-read every pass and
            // never completes, so waiting on a non-empty read would wait
            // for ever and the exit would never be reported.
            process::Liveness::Dead if complete > 0 => {}
            process::Liveness::Dead => {
                queue.lock().unwrap().close(&sink, "the session ended");
                let detail = stderr_tail(&stderr_path);
                if !detail.is_empty() {
                    let _ = sink.send(Event::Error {
                        message: format!("{label} exited:\n{detail}"),
                    });
                }
                let _ = sink.send(Event::Status {
                    state: SessionStatus::Exited,
                });
                process::clear(&session_dir);
                return;
            }
            // The record is there and the machine will not say whether the
            // process behind it is. Reported rather than guessed: calling
            // it exited would invite starting a second one against the
            // same conversation, which is the expensive mistake here.
            // Kept polling, so it resolves itself if the answer comes back.
            process::Liveness::Unknown => {
                if !said_unknown {
                    said_unknown = true;
                    let _ = sink.send(Event::Status {
                        state: SessionStatus::Unknown,
                    });
                }
            }
        }
        tokio::time::sleep(POLL).await;
    }
}

/// How many leading bytes of `bytes` form complete lines.
///
/// The offset only ever advances by this, which is what lets a read land
/// anywhere -- mid-line, mid-character -- without the reader losing its
/// place. See the call site for why the remainder is not kept.
///
/// A line ends at `\n` and at nothing else, deliberately. This stream is
/// JSONL: a record is a line, and something terminated by a bare `\r` is
/// not a record, so treating one as a line would hand `serde_json` a
/// fragment. The accepted consequence is that such a line is held here
/// forever rather than being reported -- and it is worth knowing what
/// that would look like, because it looks like nothing: the session goes
/// quiet with the process healthy, no error anywhere, and the cause is a
/// line splitter, which is not where anybody would look. A progress
/// indicator is the usual reason a program writes one (`\r` is how it
/// redraws in place), and the CLI has never written one here.
fn complete_lines(bytes: &[u8]) -> usize {
    bytes
        .iter()
        .rposition(|byte| *byte == b'\n')
        .map(|at| at + 1)
        .unwrap_or(0)
}

/// One line of the CLI's output as events on the sink. `false` when the
/// session has been torn down and there is nothing left to send to.
fn translate_line(
    line: &str,
    session_dir: &Path,
    state: &Arc<Mutex<Translator>>,
    sink: &EventSink,
    queue: &Arc<Mutex<Queue>>,
) -> bool {
    let Ok(message) = serde_json::from_str::<Value>(line) else {
        // By characters, not bytes: the CLI emits plenty of non-ASCII, and
        // a byte slice that lands mid-character panics -- inside `follow`,
        // which is the task reading this session's output, so the session
        // would go permanently deaf with nothing on screen to say so. The
        // other three truncations in this codebase (`setups.rs`,
        // `translate.rs`, `import.rs`) already do it this way.
        let shown: String = line.chars().take(200).collect();
        tracing::warn!("unparseable claude output line: {shown}");
        return true;
    };
    let opens_a_model_call = starts_a_model_call(&message);
    let (events, new_session_id, before) = {
        let mut state = state.lock().unwrap();
        let before = state.session_id.clone();
        let events = state.translate(&message);
        let after = state.session_id.clone();
        (events, if before != after { after } else { None }, before)
    };
    if let Some(session_id) = new_session_id {
        write_resume_token(session_dir, &session_id);
    }
    // A turn the CLI began by itself, said one line earlier than anything
    // else could say it.
    //
    // The CLI picks the conversation back up with nothing written to it --
    // measured: a backgrounded `sleep` finished nine seconds after the
    // turn's result and it started again unprompted. It announces that with
    // an `init`, and the first assistant text follows about a second and a
    // half later; until this, that second and a half read as idle, which is
    // long enough to send a command into and have it read as text.
    //
    // `before.is_some()` is what separates this from the `init` at startup,
    // which announces a session that is *waiting*. Our own `/clear` also
    // produces one, and is excluded by `running` already being true --
    // `local_command` set it before the line went out.
    if opens_a_turn_by_itself(&message, before.is_some()) {
        let started = {
            let mut queue = queue.lock().unwrap();
            let started = !queue.running && !queue.closed;
            if started {
                queue.running = true;
            }
            started
        };
        if started
            && sink
                .send(Event::Status {
                    state: SessionStatus::Running,
                })
                .is_err()
        {
            return false;
        }
    }
    // The steer is announced where the CLI opens the model call that read
    // it, and the announcement goes out *before* that call's output, so
    // the message sits above what it produced and below what it did not.
    //
    // This line carries no events of its own, which is what makes it the
    // right place: everything the previous call produced -- its text, its
    // tool calls, their results -- is already recorded above.
    if opens_a_model_call && !announce_steers(queue, sink) {
        return false;
    }
    for event in events {
        // A turn nobody here started -- see `proves_a_turn`. Said before
        // the event that proves it, for the same reason a steer is: the
        // session was already working when it produced this.
        let started = {
            let mut queue = queue.lock().unwrap();
            let started = proves_a_turn(&event) && !queue.running && !queue.closed;
            if started {
                queue.running = true;
            }
            started
        };
        if started
            && sink
                .send(Event::Status {
                    state: SessionStatus::Running,
                })
                .is_err()
        {
            return false;
        }
        if matches!(
            event,
            Event::Status {
                state: SessionStatus::Idle
            }
        ) {
            // The case that must not be missed: a message written after
            // the final model call of a turn has no later `message_start`
            // to prove anything, so without this it would never be
            // announced at all. The end of the turn is where it belongs
            // anyway -- nothing above it came after the message.
            if !announce_steers(queue, sink) {
                return false;
            }
            queue.lock().unwrap().running = false;
        }
        if sink.send(event).is_err() {
            return false; // session torn down
        }
    }
    true
}

/// Whether this line is the CLI announcing work it started on its own.
///
/// `system/init` is how it says a conversation is beginning, and it sends
/// one in three cases: at startup, after a `/clear`, and when it picks the
/// conversation back up by itself. Only the third is a turn nobody here
/// asked for. `already_started` -- whether the translator had a session id
/// before this line -- rules out the first, and the caller's `running`
/// check rules out the second.
fn opens_a_turn_by_itself(message: &Value, already_started: bool) -> bool {
    already_started
        && message.get("type").and_then(Value::as_str) == Some("system")
        && message.get("subtype").and_then(Value::as_str) == Some("init")
}

/// Whether this event could only have come from a turn in flight.
///
/// The turn this side starts is announced where it is started, and that
/// covers the common case and nothing else. Everything below happens
/// without a phone asking for it: a compaction the CLI decided on by
/// itself, a session adopted while it was already mid-turn, a message
/// that reached the conversation by some route other than this server --
/// another agent writing to it, or somebody at the terminal. In all of
/// them the CLI is plainly working and the only thing that would ever
/// have said so is a `Running` nobody sent, so the session sits there
/// reading as idle until the turn ends.
///
/// So the driver says it from what it observes rather than from what it
/// was asked to do. Deliberately a wider set than what announces a steer
/// (see [`announce_steers`]): any sign of work proves a turn is running,
/// while only a `message_start` proves a line written a moment ago has
/// been read. `Idle` is the pair to this -- it is where `running` goes
/// back to false, a few lines above where it is set here.
fn proves_a_turn(event: &Event) -> bool {
    matches!(
        event,
        Event::AssistantText { .. }
            | Event::ToolStart { .. }
            | Event::ToolUpdate { .. }
            | Event::ToolEnd { .. }
            | Event::Question { .. }
            | Event::Compacted { .. }
            | Event::Status {
                state: SessionStatus::Compacting | SessionStatus::AwaitingInput
            }
    )
}

/// Records every message written since the last announcement, in the
/// order it was written. False means the session has been torn down.
///
/// Called from the two places that prove the CLI has consumed them: the
/// start of a new model call, and the end of the turn. Both are in
/// [`translate_line`], and the pair is the whole of the rule -- a steer
/// announced anywhere else lands above output that predates it.
fn announce_steers(queue: &Arc<Mutex<Queue>>, sink: &EventSink) -> bool {
    let taken: Vec<(String, String, Vec<ImageRef>)> = {
        let mut queue = queue.lock().unwrap();
        queue.awaiting.drain(..).collect()
    };
    for (id, text, images) in taken {
        if sink
            .send(Event::MessageTaken {
                id: Some(id),
                text,
                images,
            })
            .is_err()
        {
            return false;
        }
    }
    true
}

/// The end of the stderr log, for an exit report a person reads.
///
/// Bounded because this is held in a message; trimmed of blank lines at
/// both ends because a shell's error ends with one, so anything reporting
/// "the last line" reports nothing at all. A failing `cd` cost an evening
/// to exactly that.
fn stderr_tail(path: &Path) -> String {
    let Ok(text) = std::fs::read_to_string(path) else {
        return String::new();
    };
    let kept: VecDeque<String> = text
        .lines()
        .rev()
        .take(STDERR_LINES_KEPT)
        .map(str::to_string)
        .collect::<Vec<_>>()
        .into_iter()
        .rev()
        .collect();
    tail_of(&kept)
}

/// Creates the stdin fifo if it is not already there, and opens it
/// read-write for the process to inherit.
///
/// Read-write is the whole trick, and it is not an accident of
/// convenience: a fifo opened read-only delivers EOF as soon as the last
/// writer closes, so the process would exit the moment this server did --
/// which is exactly what leaving it running has to prevent. Holding it
/// open for writing as well means the process is its own last writer and
/// never sees the end of its input.
fn make_fifo(path: &Path) -> Result<std::fs::File> {
    if !path.exists() {
        let c_path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes())
            .with_context(|| format!("{} is not a usable path", path.display()))?;
        // SAFETY: a nul-terminated path this call only reads, and a mode
        // with no bits the kernel can object to. Owner-only, like
        // everything else in a session directory: this carries what the
        // person typed.
        let made = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) };
        if made != 0 {
            return Err(std::io::Error::last_os_error())
                .with_context(|| format!("creating the fifo {}", path.display()));
        }
    }
    std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(path)
        .with_context(|| format!("opening the fifo {}", path.display()))
}

/// A fresh, empty, owner-only log for one of the process's output streams.
fn create_log(path: &Path) -> Result<std::fs::File> {
    std::fs::OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true)
        .mode(0o600)
        .open(path)
        .with_context(|| format!("creating {}", path.display()))
}

pub(super) fn read_resume_token(session_dir: &Path) -> Option<String> {
    let text = std::fs::read_to_string(session_dir.join(RESUME_FILE)).ok()?;
    serde_json::from_str::<Value>(&text)
        .ok()?
        .get("sessionId")?
        .as_str()
        .map(String::from)
}

pub(super) fn write_resume_token(session_dir: &Path, session_id: &str) {
    let path = session_dir.join(RESUME_FILE);
    if let Err(err) = std::fs::write(&path, json!({"sessionId": session_id}).to_string()) {
        tracing::error!("couldn't persist resume token to {}: {err}", path.display());
    }
}

/// Reads an uploaded attachment into an API image content block.
fn attachment_block(session_dir: &Path, id: &str) -> Result<Value> {
    // Ids are server-generated hex (see routes::upload_attachment); the
    // check keeps a crafted "id" from naming an arbitrary file.
    if !id
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
    {
        anyhow::bail!("invalid attachment id");
    }
    let path = session_dir.join("attachments").join(id);
    let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
    use base64::Engine;
    Ok(json!({
        "type": "image",
        "source": {
            "type": "base64",
            // Ids carry the extension the upload was stored under, so an
            // unrecognized one means a name this server didn't write.
            "media_type": crate::media::media_type_for(id).unwrap_or("image/jpeg"),
            "data": base64::engine::general_purpose::STANDARD.encode(bytes),
        }
    }))
}

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

    /// Drives real CLI output lines through the reader and collects what
    /// came out, which is the only way to check the wiring between "the
    /// CLI said this" and "the transcript records that".
    fn events_from_lines(lines: &[&str]) -> Vec<Event> {
        let dir = tempfile::tempdir().expect("temp dir");
        let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
        let queue = Arc::new(Mutex::new(Queue::default()));
        let (sink, mut out) = mpsc::unbounded_channel::<Event>();
        for line in lines {
            assert!(translate_line(line, dir.path(), &state, &sink, &queue));
        }
        drop(sink);
        let mut events = Vec::new();
        while let Ok(event) = out.try_recv() {
            events.push(event);
        }
        events
    }

    /// Feeds lines through the reader, running `interject` between two of
    /// them, and returns what came out.
    ///
    /// The hook is what makes a steer testable at all: what matters is
    /// not which events a line produces but *where* a message written
    /// part-way through the stream ends up among them.
    fn events_with_interjection(
        lines: &[&str],
        after: usize,
        interject: impl FnOnce(&Arc<Mutex<Queue>>),
    ) -> Vec<Event> {
        let dir = tempfile::tempdir().expect("temp dir");
        let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
        let queue = Arc::new(Mutex::new(Queue::default()));
        let (sink, mut out) = mpsc::unbounded_channel::<Event>();
        let mut interject = Some(interject);
        for (i, line) in lines.iter().enumerate() {
            assert!(translate_line(line, dir.path(), &state, &sink, &queue));
            if i == after {
                interject.take().expect("one interjection")(&queue);
            }
        }
        drop(sink);
        let mut events = Vec::new();
        while let Ok(event) = out.try_recv() {
            events.push(event);
        }
        events
    }

    /// One assistant message, streamed: two text deltas, then the
    /// `tool_use` it ends with, then that call's result.
    ///
    /// Written out rather than shortened because the point of both tests
    /// below is the *order*, and the shape of a real turn is what makes
    /// the order mean anything. Recorded from 2.1.237.
    const STREAMED_CALL: &[&str] = &[
        r#"{"type":"stream_event","event":{"type":"message_start"},"session_id":"s","parent_tool_use_id":null}"#,
        r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me "}},"session_id":"s","parent_tool_use_id":null}"#,
        r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"check that."}},"session_id":"s","parent_tool_use_id":null}"#,
        r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo one"}}]},"parent_tool_use_id":null}"#,
        r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"one","is_error":false}]},"parent_tool_use_id":null}"#,
    ];

    /// A steer typed while an answer is streaming is recorded below that
    /// answer's tool call and its result, not among them.
    ///
    /// The message reaches the CLI immediately; what waits is saying so.
    /// Everything the CLI emits after it was typed still belongs to a
    /// model call that had not read it -- the rest of the text, the
    /// `tool_use` the model had already committed to, the result that
    /// came back. `message_start` is the first line that proves the next
    /// call has it, so that is where the announcement goes.
    #[test]
    fn a_steer_is_recorded_below_the_call_that_had_not_read_it() {
        let mut lines = STREAMED_CALL.to_vec();
        lines.push(
            r#"{"type":"stream_event","event":{"type":"message_start"},"session_id":"s","parent_tool_use_id":null}"#,
        );
        lines.push(
            r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Doing that instead."}},"session_id":"s","parent_tool_use_id":null}"#,
        );
        // Typed after the first delta, with the answer still arriving.
        let events = events_with_interjection(&lines, 1, |queue| {
            queue.lock().unwrap().awaiting.push_back((
                "q1".into(),
                "do the other one instead".into(),
                Vec::new(),
            ))
        });

        let at = |find: fn(&Event) -> bool| {
            events
                .iter()
                .position(find)
                .unwrap_or_else(|| panic!("nothing matched in {events:?}"))
        };
        let taken = at(|e| matches!(e, Event::MessageTaken { .. }));
        // Named, not just announced: the phone has a waiting bubble on screen for this message
        // and clears the one with this id. Matching on the text instead would clear the wrong
        // bubble whenever the same thing was sent twice.
        assert!(
            matches!(
                &events[taken],
                Event::MessageTaken { id: Some(id), .. } if id == "q1"
            ),
            "an announcement must name the queue entry it resolves: {events:?}"
        );
        assert!(
            taken > at(|e| matches!(e, Event::ToolStart { .. })),
            "a steer must not sit above a call the model had already made: {events:?}"
        );
        assert!(
            taken > at(|e| matches!(e, Event::ToolEnd { .. })),
            "a steer must not sit above the result of that call: {events:?}"
        );
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, Event::AssistantText { .. }))
                .count(),
            3,
            "the streamed answer must stay whole: {events:?}"
        );
    }

    /// A steer written after the turn's last model call is still recorded.
    ///
    /// Nothing further is coming, so no `message_start` will ever prove
    /// it was read -- and a message that is only recorded when announced
    /// would otherwise vanish, leaving a phone drawing it as still
    /// waiting forever. The end of the turn is also where it belongs:
    /// nothing above it happened after it was typed.
    #[test]
    fn a_steer_with_no_model_call_left_is_recorded_at_the_end_of_the_turn() {
        let mut lines = STREAMED_CALL.to_vec();
        lines.push(
            r#"{"type":"result","subtype":"success","usage":{"input_tokens":1,"output_tokens":1}}"#,
        );
        // Typed after the tool result, with only the turn's end to come.
        let events = events_with_interjection(&lines, 4, |queue| {
            queue
                .lock()
                .unwrap()
                .awaiting
                .push_back(("q2".into(), "never mind".into(), Vec::new()))
        });

        let taken = events
            .iter()
            .position(|e| matches!(e, Event::MessageTaken { .. }))
            .unwrap_or_else(|| panic!("a steer must never be dropped: {events:?}"));
        let idle = events
            .iter()
            .position(|e| {
                matches!(
                    e,
                    Event::Status {
                        state: SessionStatus::Idle
                    }
                )
            })
            .unwrap_or_else(|| panic!("expected the turn to end: {events:?}"));
        assert!(
            taken < idle,
            "the steer belongs inside the turn it was typed into: {events:?}"
        );
    }

    /// The divider comes from the CLI announcing the reset, not from an
    /// `init` arriving.
    ///
    /// Measured against 2.1.237: `/clear` emits `conversation_reset`,
    /// then a fresh `init` carrying a new session id. Watching the id be
    /// replaced would work, but it reads the event through one of its
    /// side effects; the announcement says so directly and arrives first,
    /// so the divider lands above the new conversation rather than below
    /// its opening line.
    #[test]
    fn a_conversation_reset_is_what_records_a_clear() {
        let events = events_from_lines(&[
            r#"{"type":"system","subtype":"init","session_id":"first","tools":[],"model":"claude-haiku-4-5-20251001"}"#,
            r#"{"type":"conversation_reset","session_id":"first"}"#,
            r#"{"type":"system","subtype":"init","session_id":"second","tools":[],"model":"claude-haiku-4-5-20251001"}"#,
        ]);
        assert_eq!(
            events.iter().filter(|e| **e == Event::Cleared).count(),
            1,
            "got {events:?}"
        );
    }

    /// An `init` on its own never records a clear, whatever id it carries.
    ///
    /// Three ways one arrives and none of them is a cleared conversation:
    /// the first init of a session, the one a compaction re-announces
    /// carrying the *same* id, and the one that follows a resume. Reading
    /// any of them as a clear would open sessions with a divider
    /// announcing something that never happened, or draw one on top of a
    /// compaction's own mark and tell the reader the conversation had been
    /// dropped when it had been summarised.
    #[test]
    fn an_init_alone_is_never_a_clear() {
        for ids in [["first", "first"], ["first", "second"]] {
            let events = events_from_lines(&[
                &format!(
                    r#"{{"type":"system","subtype":"init","session_id":"{}","tools":[],"model":"claude-haiku-4-5-20251001"}}"#,
                    ids[0]
                ),
                &format!(
                    r#"{{"type":"system","subtype":"init","session_id":"{}","tools":[],"model":"claude-haiku-4-5-20251001"}}"#,
                    ids[1]
                ),
            ]);
            assert!(!events.contains(&Event::Cleared), "{ids:?} gave {events:?}");
        }
    }

    /// The failure this exists for: a shell's complaint ends with a blank
    /// line, so reporting "the last line of stderr" reported nothing, and
    /// the phone showed a bare exit status while the reason sat in the
    /// server's log.
    #[test]
    fn the_report_keeps_the_message_and_not_the_blank_line_after_it() {
        let fish_cd_failure = [
            "cd: The directory '~/repos/ai-app' does not exist",
            "",
            "embedded:functions/cd.fish (line 26): ",
            "    builtin cd $argv",
            "    ^",
            "in function 'cd' with arguments '~/repos/ai-app'",
            "",
        ];
        let kept: VecDeque<String> = fish_cd_failure.iter().map(|l| l.to_string()).collect();

        let report = tail_of(&kept);
        assert!(
            report.starts_with("cd: The directory"),
            "the complaint leads: {report}",
        );
        assert!(
            report.ends_with("'~/repos/ai-app'"),
            "the trailing blank is trimmed: {report:?}",
        );
        // The blank *between* lines is part of the message and stays.
        assert!(report.contains("does not exist\n\nembedded:"), "{report:?}");
    }

    /// Nothing to say is said as nothing, so the caller can tell the two
    /// apart and print just the exit status.
    #[test]
    fn stderr_that_is_only_blank_lines_reports_as_empty() {
        let kept: VecDeque<String> = ["", "  ", ""].iter().map(|l| l.to_string()).collect();
        assert_eq!(tail_of(&kept), "");
        assert_eq!(tail_of(&VecDeque::new()), "");
    }

    #[test]
    fn a_stream_read_in_arbitrary_chunks_yields_each_line_once() {
        // The property the reading position has to have: however a write
        // is split -- mid-line, or mid-character -- every line comes out
        // exactly once and in order. Chunked at every prime-ish size so
        // the cuts land in different places, including inside the
        // multi-byte character.
        let stream = "{\"a\":1}\n{\"b\":\"caf\u{e9}\"}\n{\"c\":3}\n";
        for chunk in [1usize, 2, 3, 5, 7, 11, 1000] {
            let mut offset = 0usize;
            let mut lines: Vec<String> = Vec::new();
            let bytes = stream.as_bytes();
            let mut available = 0usize;
            while available < bytes.len() {
                available = (available + chunk).min(bytes.len());
                // What a read from the recorded offset returns: the file
                // as far as it has been written, from where we left off.
                let unread = &bytes[offset..available];
                let complete = complete_lines(unread);
                for line in String::from_utf8_lossy(&unread[..complete]).lines() {
                    lines.push(line.to_string());
                }
                offset += complete;
            }
            assert_eq!(offset, bytes.len(), "chunk {chunk} left bytes unread");
            assert_eq!(
                lines,
                vec!["{\"a\":1}", "{\"b\":\"caf\u{e9}\"}", "{\"c\":3}"],
                "chunk {chunk}"
            );
        }
    }

    #[test]
    fn an_incomplete_line_advances_nothing() {
        // Nothing to do yet, and crucially the position does not move --
        // so a crash here re-reads the line rather than skipping it.
        assert_eq!(complete_lines(b"{\"partial\": tru"), 0);
        assert_eq!(complete_lines(b""), 0);
        // And a complete line followed by a partial one advances only past
        // the complete one.
        assert_eq!(complete_lines(b"done\nhalf"), 5);
    }

    #[test]
    fn closing_the_queue_reports_what_was_never_read() {
        let (sink, mut received) = mpsc::unbounded_channel();
        let mut queue = Queue {
            running: true,
            ..Queue::default()
        };
        queue
            .awaiting
            .push_back(("q1".into(), "first".into(), Vec::new()));
        queue
            .awaiting
            .push_back(("q2".into(), "second".into(), Vec::new()));
        queue.close(&sink, "the session ended");

        // Named rather than counted, because these never reached the
        // transcript: this message is the only record they existed.
        let Some(Event::Error { message }) = received.try_recv().ok() else {
            panic!("closing a queue holding messages must report them");
        };
        assert!(message.contains("2 queued messages"), "{message}");
        assert!(message.starts_with("the session ended"), "{message}");
        assert!(
            message.contains("first") && message.contains("second"),
            "{message}"
        );

        // And the flag is cleared, so a later message is refused with a
        // reason rather than queued behind a turn that will never end.
        assert!(!queue.running);
        assert!(queue.closed);
    }

    #[test]
    fn a_turn_this_side_did_not_start_still_reports_as_running() {
        // The case: a session picked up while it was already working, or
        // one another agent wrote to. Nothing called `send_user_message`,
        // so the only thing that can say the session is busy is what it
        // is observed doing.
        let dir = tempfile::tempdir().expect("tempdir");
        let (sink, mut received) = mpsc::unbounded_channel();
        let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
        let queue = Arc::new(Mutex::new(Queue::default()));

        let text = r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"working"}},"parent_tool_use_id":null}"#;
        assert!(translate_line(text, dir.path(), &state, &sink, &queue));
        assert_eq!(
            received.try_recv().ok(),
            Some(Event::Status {
                state: SessionStatus::Running
            }),
            "a turn in flight has to be reported before the output proving it"
        );
        assert!(matches!(
            received.try_recv().ok(),
            Some(Event::AssistantText { .. })
        ));

        // Once only: the turn is known to be running now, and a status per
        // delta would be a status per word.
        assert!(translate_line(text, dir.path(), &state, &sink, &queue));
        assert!(matches!(
            received.try_recv().ok(),
            Some(Event::AssistantText { .. })
        ));

        // And the end of the turn puts it back, so the next one is
        // reported the same way.
        let done = r#"{"type":"result","subtype":"success","is_error":false,"usage":{}}"#;
        assert!(translate_line(done, dir.path(), &state, &sink, &queue));
        assert_eq!(
            received.try_recv().ok(),
            Some(Event::Status {
                state: SessionStatus::Idle
            })
        );
        assert!(!queue.lock().unwrap().running);
    }

    #[test]
    fn output_from_a_process_that_has_gone_does_not_revive_the_turn() {
        // `close` is what says the process is gone and reports the
        // messages that died with it. Anything still in the pipe after
        // that must not put the session back to work, because there is
        // nothing left to do the work.
        let dir = tempfile::tempdir().expect("tempdir");
        let (sink, mut received) = mpsc::unbounded_channel();
        let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
        let queue = Arc::new(Mutex::new(Queue::default()));
        queue.lock().unwrap().close(&sink, "the session ended");

        let text = r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"late"}},"parent_tool_use_id":null}"#;
        assert!(translate_line(text, dir.path(), &state, &sink, &queue));
        assert!(matches!(
            received.try_recv().ok(),
            Some(Event::AssistantText { .. })
        ));
        assert!(!queue.lock().unwrap().running);
    }

    #[test]
    fn closing_an_empty_queue_says_nothing() {
        let (sink, mut received) = mpsc::unbounded_channel();
        let mut queue = Queue::default();
        queue.close(&sink, "the session ended");
        // A session that exits with nothing held has lost nothing, and an
        // error saying so would be noise on every ordinary exit.
        assert!(received.try_recv().is_err());
        assert!(queue.closed);
    }
}