//! The phase-1 fake driver: no child process, just events. It exists to
//! prove the whole pipe -- spawn, transcript, SSE cursors, questions,
//! interrupts, compaction -- before any AI is involved, and stays useful afterwards as
//! a connectivity check that costs no tokens.
//!
//! Behavior: every message is echoed back as a few streamed text deltas.
//! A leading word asks for something more specific:
//!
//! - `/tool [input]` -- a full tool run, start through end.
//! - `/tools [n]` -- n calls back to back, for what a run of them looks
//!   like when a screen groups them.
//! - `/question [text]` -- a question, exercising the answer path.
//! - `/ask` -- an AskUserQuestion call: two questions on one tool call,
//!   with descriptions, a preview and a multi-select, which is the shape
//!   that is awkward to get a real model to produce on demand.
//! - `/slow [seconds]` -- a turn that stays running (default 30), so states that only
//!   exist *while* something is happening can be looked at.
//! - `/error [text]` -- a failure, which is otherwise awkward to cause.
//! - `/peer [text]` -- a message from another agent, which otherwise takes
//!   two live sessions and one of them deciding to write.
//! - `/compact` -- a compaction, start to finish. Typed rather than
//!   pressed, because the real dialects take it as a typed command too and
//!   the phone no longer has a button for it.
//!
//! This is exactly the event vocabulary the real drivers produce, so a UI
//! that renders echo sessions correctly renders the real thing.
//!
//! - `/stream N` -- one long answer in N small pieces, 50ms apart: the
//!   shape a real model's reply arrives in, and the one where the row a
//!   reader is anchored to is the row that keeps changing height.
//! - `/mixed N` -- N beats of an interleaved transcript: paragraphs of
//!   different lengths, single tool calls, runs of adjacent ones, images
//!   and a peer message. Rows of every shape and height the app draws, in
//!   one session, which is what a scrolling problem needs in order to be
//!   reproduced twice the same way.
//!
//! `/slow` earns its place: a queued message, a Stop button, a spinner
//! where the answer will go are all states that only exist mid-turn, and
//! the obvious way to get one -- ask a real model to sleep -- does not
//! work. It declines, reasonably, and answers instantly instead, so the
//! state never arrives and the attempt still costs a turn on somebody's
//! account. A driver that can be *told* to take its time costs nothing and
//! is the same every run.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use super::driver::{Driver, Event, EventSink, ImageRef, QuestionOption, SessionStatus};

/// Delay between streamed deltas -- long enough that streaming is visibly
/// streaming in the UI, short enough that tests waiting on a full turn
/// stay fast.
const DELTA_DELAY: Duration = Duration::from_millis(50);

/// How long a fake compaction takes.
///
/// A measured one, near enough: driving a real session through `/compact`
/// on 2026-08-29 took 13 seconds for a small conversation, and a large one
/// takes minutes. Three seconds -- what this was -- is too short to look
/// at the row that only exists while a compaction is running, and too
/// short to watch its elapsed count reach two digits.
const COMPACT_TIME: Duration = Duration::from_secs(13);

/// A question echo is waiting on, and the tool call it belongs to.
///
/// `call` is `None` for `/question`, which asks on its own the way a
/// permission does; `Some` for `/ask`, where several questions share one
/// call and the call ends when the last of them is answered.
struct PendingQuestion {
    id: String,
    call: Option<String>,
}

pub struct EchoDriver {
    sink: EventSink,
    /// Whether a turn is in flight, and what arrived during it.
    ///
    /// A real CLI holds a message sent mid-turn and injects it at the next
    /// tool boundary; echo used to answer it on the spot, which made it
    /// the wrong shape for testing anything about queueing -- the status
    /// dropped to idle immediately, so a phone had nothing to show as
    /// pending. Holding it here is what makes echo able to stand in.
    busy: Arc<AtomicBool>,
    /// Held messages with the id of the `MessageQueued` each one announced,
    /// so the announcement can say which waiting bubble it resolves.
    queued: Arc<Mutex<Vec<Held>>>,
    /// Where `/mixed` writes the images it references, which is the same
    /// directory the files route serves them from.
    session_dir: PathBuf,
    /// Ids of the questions awaiting an answer, in the order they were
    /// asked. A list because `/ask` puts up to four on one tool call, the
    /// way AskUserQuestion does, and the turn resumes when the last of
    /// them is answered rather than the first.
    pending_questions: Mutex<Vec<PendingQuestion>>,
}

impl EchoDriver {
    /// An AskUserQuestion call, in the shape the CLI sends one.
    ///
    /// Two questions on one call, because that is where the display is
    /// hardest and where it was wrong: one question with four options
    /// reads fine even when the options are laid out badly. Written out
    /// in full rather than generated so it carries the parts that are
    /// easy to leave out of a fixture -- a header, an option with a
    /// description, an option with a preview block, and a multi-select.
    fn ask_user_question(&self) {
        // Written once, in the shape the events carry, and turned into
        // the tool call's own input below -- the CLI sends both, and two
        // hand-written copies of one question would drift.
        let asked = [
            (
                "Theme",
                "Which colour scheme should the transcript use?",
                false,
                vec![
                    QuestionOption {
                        label: "Catppuccin Mocha (Recommended)".to_string(),
                        description: Some(
                            "What the app uses now: a dark base with muted accents.".to_string(),
                        ),
                        preview: None,
                    },
                    QuestionOption {
                        label: "Solarized Dark".to_string(),
                        description: Some(
                            "Lower contrast, warmer. Easier at night, harder in sun.".to_string(),
                        ),
                        preview: None,
                    },
                    QuestionOption {
                        label: "High contrast".to_string(),
                        description: Some(
                            "Pure black behind white text, for reading outdoors.".to_string(),
                        ),
                        preview: Some(
                            "background: #000000\nforeground: #ffffff\naccent:     #ffd700"
                                .to_string(),
                        ),
                    },
                ],
            ),
            (
                "Collapsed",
                "Which of these should be shown collapsed by default?",
                true,
                vec![
                    QuestionOption {
                        label: "Tool calls".to_string(),
                        description: Some("A run of them becomes one card.".to_string()),
                        preview: None,
                    },
                    QuestionOption {
                        label: "Peer messages".to_string(),
                        description: Some("Messages from other agents.".to_string()),
                        preview: None,
                    },
                    QuestionOption {
                        label: "Compaction notes".to_string(),
                        description: Some("What a compaction recovered.".to_string()),
                        preview: None,
                    },
                ],
            ),
        ];

        let call = format!("echo-ask-{}", super::random_hex());
        self.emit(Event::Status {
            state: SessionStatus::Running,
        });
        self.emit(Event::ToolStart {
            id: call.clone(),
            tool: "AskUserQuestion".to_string(),
            input: serde_json::json!({"questions": asked
                .iter()
                .map(|(header, question, multi, options)| serde_json::json!({
                    "question": question,
                    "header": header,
                    "multiSelect": multi,
                    "options": options,
                }))
                .collect::<Vec<_>>()}),
        });
        for (index, (header, question, multi, options)) in asked.into_iter().enumerate() {
            let id = format!("{call}#{index}");
            self.pending_questions
                .lock()
                .unwrap()
                .push(PendingQuestion {
                    id: id.clone(),
                    call: Some(call.clone()),
                });
            self.emit(Event::Question {
                id,
                prompt: question.to_string(),
                header: Some(header.to_string()),
                options,
                multi_select: multi,
                // The call that asked, so all of it draws as one thing --
                // which is the whole point of the fixture.
                about: Some(call.clone()),
            });
        }
        self.emit(Event::Status {
            state: SessionStatus::AwaitingInput,
        });
    }

    /// One typed line, whether it arrived as a message or as a command.
    ///
    /// `announce` is the difference and it is the whole of it: a message
    /// is announced with `MessageTaken`, which is what puts it in the
    /// transcript, and a command is not -- the manager has already
    /// recorded that one was sent, and saying so twice drew the same
    /// line in both colours.
    fn handle(&self, text: String, images: Vec<ImageRef>, announce: bool) {
        let sink = self.sink.clone();

        // Mid-turn messages are held rather than answered, the way a real
        // CLI holds them until the next tool boundary. Without this the
        // session went idle the instant one arrived, and every state that
        // only exists while something is queued was untestable.
        if self.busy.load(Ordering::SeqCst) {
            // The waiting is recorded, exactly as the real driver records
            // it: the phone draws its pending bubbles from the server, so
            // an echo session has to produce the same events or the states
            // it exists to exercise are not the app's real ones.
            let id = super::random_hex();
            self.queued
                .lock()
                .unwrap()
                .push((id.clone(), text.clone(), images.clone()));
            if announce {
                self.emit(Event::MessageQueued { id, text, images });
            }
            return;
        }

        // Answered on the spot rather than in the turn below, because a
        // peer message is not a turn: it is something that arrives, and
        // what is being exercised is the row it becomes. The message that
        // asked for it is still announced -- every driver owes exactly one
        // `MessageTaken` per message, and a command that quietly vanishes
        // from the transcript is the one thing echo must not model.
        if let Some(rest) = text.strip_prefix("/peer") {
            if announce {
                self.emit(Event::MessageTaken {
                    id: None,
                    text: text.clone(),
                    images: images.clone(),
                });
            }
            self.emit(Event::PeerMessage {
                from: "dev-updater-f5".to_string(),
                text: if rest.trim().is_empty() {
                    "Pull before you touch AGENTS.md -- I pushed three commits to it \
                     in the last hour, and origin/main has moved since you last looked.\n\n\
                     The tree is clean as of now, but it was not for most of that time."
                        .to_string()
                } else {
                    rest.trim().to_string()
                },
            });
            return;
        }

        // The same word the real CLI takes, so a phone drives both the same
        // way. `Driver::compact` is what the manager's own route calls;
        // this is the typed path onto it.
        if text.trim() == "/compact" {
            if announce {
                self.emit(Event::MessageTaken {
                    id: None,
                    text,
                    images,
                });
            }
            self.compact();
            return;
        }

        if text.trim() == "/ask" {
            if announce {
                self.emit(Event::MessageTaken {
                    id: None,
                    text,
                    images,
                });
            }
            self.ask_user_question();
            return;
        }

        if let Some(rest) = text.strip_prefix("/question") {
            let id = format!("q-{}", super::random_hex());
            let prompt = if rest.trim().is_empty() {
                "Echo asks: proceed?".to_string()
            } else {
                format!("Echo asks: {}", rest.trim())
            };
            self.pending_questions
                .lock()
                .unwrap()
                .push(PendingQuestion {
                    id: id.clone(),
                    call: None,
                });
            self.emit(Event::Status {
                state: SessionStatus::Running,
            });
            self.emit(Event::Question {
                id,
                prompt,
                header: None,
                options: vec![QuestionOption::plain("Yes"), QuestionOption::plain("No")],
                multi_select: false,
                about: None,
            });
            self.emit(Event::Status {
                state: SessionStatus::AwaitingInput,
            });
            return;
        }

        // Checked before `/tool`, which is a prefix of it: matching the
        // shorter one first would read "/tools 4" as a single tool whose
        // input is "s 4".
        let many_tools = text.strip_prefix("/tools").map(|rest| {
            // At least two, because one call is not a run of them and this
            // exists to produce a run.
            rest.trim().parse::<usize>().unwrap_or(3).clamp(2, 12)
        });
        let run_tool = if many_tools.is_some() {
            None
        } else {
            text.strip_prefix("/tool")
                .map(|rest| rest.trim().to_string())
        };
        // Seconds to stay running before answering, default 30. Clamped
        // rather than trusted: this is a test affordance, and a session
        // pinned running for an hour by a typo is a worse outcome than a
        // short wait.
        let stream = text
            .strip_prefix("/stream")
            .map(|rest| rest.trim().parse::<usize>().unwrap_or(400).clamp(1, 4000));
        let mixed = text
            .strip_prefix("/mixed")
            .map(|rest| rest.trim().parse::<usize>().unwrap_or(12).clamp(1, 400));
        let linger = text.strip_prefix("/slow").map(|rest| {
            Duration::from_secs(rest.trim().parse::<u64>().unwrap_or(30).clamp(1, 600))
        });
        let fail = text
            .strip_prefix("/error")
            .map(|rest| rest.trim().to_string());
        let busy = Arc::clone(&self.busy);
        let queued = Arc::clone(&self.queued);
        let dir = self.session_dir.clone();
        busy.store(true, Ordering::SeqCst);
        tokio::spawn(async move {
            let send = |event: Event| {
                let _ = sink.send(event);
            };
            let finish = || finish_turn(&sink, &queued, &busy);
            // Echo takes a message the instant it gets one, but it says so
            // anyway: a driver that skips this leaves the phone holding a
            // message it thinks is still queued, and the point of an echo
            // provider is that it behaves like the real ones.
            if announce {
                send(Event::MessageTaken {
                    id: None,
                    text: text.clone(),
                    images: images.clone(),
                });
            }
            send(Event::Status {
                state: SessionStatus::Running,
            });

            if let Some(linger) = linger {
                // A delta a second: visibly alive rather than merely slow,
                // which is what the states being looked at accompany.
                let seconds = linger.as_secs();
                for remaining in (1..=seconds).rev() {
                    send(Event::AssistantText {
                        delta: format!("still working, {remaining}s\n"),
                    });
                    tokio::time::sleep(Duration::from_secs(1)).await;
                }
                send(Event::AssistantText {
                    delta: "done.".to_string(),
                });
                finish();
                return;
            }

            if let Some(message) = fail {
                send(Event::Error {
                    message: if message.is_empty() {
                        "echo was asked to fail".to_string()
                    } else {
                        message
                    },
                });
                finish();
                return;
            }

            if let Some(count) = many_tools {
                for i in 1..=count {
                    let id = format!("t-{}", super::random_hex());
                    send(Event::ToolStart {
                        id: id.clone(),
                        tool: if i % 2 == 0 { "Read" } else { "Bash" }.to_string(),
                        input: serde_json::json!({
                            "command": format!("grep -rn 'call {i}' /tmp | head -3"),
                            "file_path": format!("/tmp/call-{i}.txt"),
                            "description": format!("The {i} of {count} calls in this run"),
                            "timeout": 5000,
                        }),
                    });
                    tokio::time::sleep(DELTA_DELAY).await;
                    send(Event::ToolEnd {
                        id,
                        output: format!("call {i} finished"),
                    });
                }
                finish();
                return;
            }

            // One long answer arriving in small pieces, which is what a real
            // model does and what `/slow` does not: `/slow` emits a line a
            // second, so its message grows in steps a reader can watch one
            // at a time. A jump caused by the *anchor row itself* changing
            // height needs growth that is continuous.
            if let Some(pieces) = stream {
                for i in 0..pieces {
                    let len = 3 + (i * 7) % 14;
                    send(Event::AssistantText {
                        delta: format!("{i}{} ", "x".repeat(len)),
                    });
                    tokio::time::sleep(Duration::from_millis(50)).await;
                }
                finish();
                return;
            }

            if let Some(beats) = mixed {
                for beat in 1..=beats {
                    write_beat(&sink, &dir, beat).await;
                }
                finish();
                return;
            }

            if let Some(input) = run_tool {
                let id = format!("t-{}", super::random_hex());
                send(Event::ToolStart {
                    id: id.clone(),
                    tool: "echo-tool".to_string(),
                    input: serde_json::json!({ "input": input }),
                });
                tokio::time::sleep(DELTA_DELAY).await;
                send(Event::ToolUpdate {
                    id: id.clone(),
                    output: "working...".to_string(),
                });
                tokio::time::sleep(DELTA_DELAY).await;
                send(Event::ToolEnd {
                    id,
                    output: format!("echoed: {input}"),
                });
            }

            // Word-at-a-time so streaming is visibly streaming.
            for word in format!("You said: {text}").split_inclusive(' ') {
                send(Event::AssistantText {
                    delta: word.to_string(),
                });
                tokio::time::sleep(DELTA_DELAY).await;
            }
            send(Event::UsageDelta {
                tokens: text.split_whitespace().count() as u64,
                total: 0,
            });
            finish();
        });
    }

    pub fn new(sink: EventSink, session_dir: PathBuf) -> Self {
        let driver = Self {
            sink,
            pending_questions: Mutex::new(Vec::new()),
            busy: Arc::new(AtomicBool::new(false)),
            queued: Arc::new(Mutex::new(Vec::new())),
            session_dir,
        };
        driver.emit(Event::Status {
            state: SessionStatus::Idle,
        });
        driver
    }

    /// Sends are infallible from the driver's point of view: a closed sink
    /// means the session is being torn down, and there is nobody left to
    /// report to.
    fn emit(&self, event: Event) {
        let _ = self.sink.send(event);
    }
}

/// A 16x10 checkerboard, the smallest thing that is recognisably an image
/// rather than a blank rectangle.
///
/// Embedded rather than generated because the alternative is a PNG encoder
/// in a test rig, and drawn at the transcript's fixed thumbnail height
/// anyway -- what a scroll test needs from an image is that it occupies an
/// image's worth of space, not that it is pretty.
const SAMPLE_PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAABAAAAAKCAIAAAAy3EnLAAAAIklEQVR42mPo3PILiOTk9ICIGDYDyRqIVwphk65h1A9EsAGCYdJRj+JH4wAAAABJRU5ErkJggg==";

/// One beat of `/mixed`: a row shape chosen by position, so the same N
/// always produces the same transcript.
///
/// Repeatable on purpose. A scrolling fault is judged by watching the same
/// content behave differently, and a rig that produced a different
/// transcript each run would make every comparison an argument about
/// whether the content changed.
async fn write_beat(sink: &EventSink, session_dir: &Path, beat: usize) {
    let send = |event: Event| {
        let _ = sink.send(event);
    };
    match beat % 5 {
        // A paragraph, of three lengths, because a list of uniform rows
        // hides exactly the faults that uneven ones expose.
        1 => {
            let words = match beat % 3 {
                0 => 12,
                1 => 60,
                _ => 220,
            };
            // Deliberately ragged: each word's length is a function of its
            // position, so no two lines wrap the same way. A paragraph of
            // uniform tokens is a wall that looks identical at every
            // offset, which makes it impossible to tell a scroll of one
            // line from a scroll of ten -- by eye or by comparing frames.
            let body: String = (0..words)
                .map(|w| {
                    let len = 3 + (w * 7 + beat * 3) % 14;
                    format!("{beat}.{w}{} ", "x".repeat(len))
                })
                .collect();
            send(Event::AssistantText {
                delta: format!("\n\nParagraph at beat {beat}:\n{body}"),
            });
        }
        // One call on its own -- drawn as a card rather than a group.
        2 => {
            let id = format!("t-{}", super::random_hex());
            send(Event::ToolStart {
                id: id.clone(),
                tool: "Read".to_string(),
                input: serde_json::json!({ "file_path": format!("/tmp/beat-{beat}.txt") }),
            });
            send(Event::ToolEnd {
                id,
                output: format!("beat {beat}: forty-two lines of nothing in particular"),
            });
        }
        // A run of three, which the app folds into one collapsed group --
        // the row whose identity depends on what is next to it.
        3 => {
            for i in 1..=3 {
                let id = format!("t-{}", super::random_hex());
                send(Event::ToolStart {
                    id: id.clone(),
                    tool: if i % 2 == 0 { "Bash" } else { "Grep" }.to_string(),
                    input: serde_json::json!({ "command": format!("grep -rn 'beat {beat}' /tmp") }),
                });
                send(Event::ToolEnd {
                    id,
                    output: format!("beat {beat}, call {i} of 3"),
                });
            }
        }
        // An image, under the call that produced it, which is where a real
        // screenshot lands.
        4 => {
            let id = format!("t-{}", super::random_hex());
            send(Event::ToolStart {
                id: id.clone(),
                tool: "Screenshot".to_string(),
                input: serde_json::json!({ "description": format!("beat {beat}") }),
            });
            let part = serde_json::json!({
                "source": {"media_type": "image/png", "data": SAMPLE_PNG}
            });
            if let Some(name) = super::claude::translate::save_image(session_dir, &part) {
                send(Event::Image {
                    image: name,
                    about: Some(id.clone()),
                });
            }
            send(Event::ToolEnd {
                id,
                output: format!("beat {beat}: captured"),
            });
        }
        // Somebody else's voice, which is its own row shape.
        _ => {
            send(Event::PeerMessage {
                from: format!("beat-{beat}-peer"),
                text: format!("Message {beat} from another session, for the row it becomes."),
            });
        }
    }
    // Slow enough that the phone renders each beat as it arrives rather
    // than composing the whole run in one frame -- which is the condition
    // a scrolling fault actually happens under.
    tokio::time::sleep(Duration::from_millis(120)).await;
}

/// A message written during a turn and waiting for it to end: the id of the
/// `MessageQueued` that announced it, what it said, and what was attached to
/// it. All three, because all three are what the `MessageTaken` at the other
/// end owes -- named rather than written out at each of the four places that
/// mention it.
type Held = (String, String, Vec<ImageRef>);

/// Ending a turn is also when anything held during it is taken up -- the
/// moment a real CLI would have injected it. One place, because a turn has
/// several ways to end (a reply, an interrupt, a compaction) and every one
/// of them owes the same answer.
fn finish_turn(sink: &EventSink, queued: &Mutex<Vec<Held>>, busy: &AtomicBool) {
    let held = std::mem::take(&mut *queued.lock().unwrap());
    for (id, text, images) in held {
        // Announced before it is answered, in that order: a phone showing
        // the message as pending needs the signal that it has been read,
        // and the answer is meaningless above a message still drawn as
        // waiting.
        let _ = sink.send(Event::MessageTaken {
            id: Some(id),
            text: text.clone(),
            images,
        });
        let _ = sink.send(Event::AssistantText {
            delta: format!("\n(taken from the queue) You said: {text}"),
        });
    }
    busy.store(false, Ordering::SeqCst);
    let _ = sink.send(Event::Status {
        state: SessionStatus::Idle,
    });
}

impl Driver for EchoDriver {
    fn between_turns(&self) -> bool {
        !self.busy.load(Ordering::SeqCst)
    }

    fn send_user_message(&self, text: String, images: Vec<ImageRef>) {
        // Announced, because this is a message: every driver owes exactly
        // one `MessageTaken` per message, and one that quietly vanishes
        // from the transcript is the thing echo must not model. A command
        // owes none -- the manager has already recorded that it was sent,
        // and announcing it again drew the same line twice, once in each
        // colour.
        self.handle(text, images, true);
    }

    /// Echo's commands *are* its messages -- `/tool`, `/slow`, `/ask` --
    /// so this is the same path with the same parsing, and the fixture
    /// behaves like a real session driven the same way.
    fn run_command(&self, text: &str) {
        self.handle(text.to_string(), Vec::new(), false);
    }

    fn answer_question(&self, id: &str, answers: &[String]) {
        let answer = answers.join(", ");
        let (answered, waiting) = {
            let mut pending = self.pending_questions.lock().unwrap();
            let Some(at) = pending.iter().position(|question| question.id == id) else {
                self.emit(Event::Error {
                    message: format!("no question {id} is awaiting an answer"),
                });
                return;
            };
            let answered = pending.remove(at);
            // Whether anything on the same call is still unanswered: a
            // tool that asked four questions ends once, not four times.
            let waiting = answered
                .call
                .as_ref()
                .is_some_and(|call| pending.iter().any(|q| q.call.as_ref() == Some(call)));
            (answered, waiting)
        };
        if waiting {
            return;
        }
        if let Some(call) = answered.call {
            self.emit(Event::ToolEnd {
                id: call,
                output: format!("answered: {answer}"),
            });
        } else {
            self.emit(Event::AssistantText {
                delta: format!("You answered: {answer}"),
            });
        }
        self.emit(Event::Status {
            state: SessionStatus::Idle,
        });
    }

    fn interrupt(&self) {
        // Nothing real to stop; a pending question is abandoned so the
        // session isn't stuck awaiting input forever.
        self.pending_questions.lock().unwrap().clear();
        self.emit(Event::Status {
            state: SessionStatus::Idle,
        });
    }

    // Nothing to forward: this process has no notion of what the
    // conversation is called, and the rename it belongs to has already
    // happened where the name lives. See `Driver::set_title`.
    fn set_title(&self, _title: &str) {}

    fn set_permission_mode(&self, mode: &str) {
        self.emit(Event::Error {
            message: format!("an echo session asks for nothing, so {mode} changes nothing"),
        });
    }

    fn set_model(&self, model: &str) {
        self.emit(Event::Error {
            message: format!("echo sessions have no model to change to {model}"),
        });
    }

    /// A compaction with nothing to compact.
    ///
    /// The counts are invented, like everything else this driver says --
    /// what is real is the shape and the order: busy, a pause long enough
    /// to see, then the result. `Compacting` and `Compacted` are states a
    /// screen has to draw, and the only other way to reach them is to fill
    /// a real session's context and spend two minutes of somebody's
    /// account getting it back.
    fn compact(&self) {
        let sink = self.sink.clone();
        let queued = Arc::clone(&self.queued);
        let busy = Arc::clone(&self.busy);
        busy.store(true, Ordering::SeqCst);
        tokio::spawn(async move {
            let _ = sink.send(Event::Status {
                state: SessionStatus::Compacting,
            });
            tokio::time::sleep(COMPACT_TIME).await;
            let _ = sink.send(Event::Compacted {
                pre_tokens: Some(128_402),
                post_tokens: Some(9_617),
                trigger: Some("manual".to_string()),
            });
            finish_turn(&sink, &queued, &busy);
        });
    }

    /// The same marker a real driver leaves, and nothing else -- there is
    /// no context here to drop. It exists so the phone's divider, its
    /// scroll behaviour and the transcript's shape can be exercised
    /// without spending a real session's context to produce one.
    fn clear(&self) {
        let _ = self.sink.send(Event::Cleared);
    }

    /// Nothing to detach from and nothing to stop: the echo driver has no
    /// process, so both halves of the way out are already done.
    fn detach(&self) {}

    fn stop(&self) {}
}