//! The HTTP surface -- REST for actions, one SSE stream per open session
//! screen for events, all behind the bearer-token middleware `main.rs`
//! wraps the whole router in.
//!
//! ```text
//! GET    /setups                       machines, each with what it can run
//! POST   /setups                       add {name, ssh?} -- providers are discovered
//! POST   /setups/probe                 dry run {ssh?}: what would be found there
//! GET    /setups/{id}                  one machine, for refetching after a change
//! PUT    /setups/{id}                  rename {name?} and/or re-probe {rediscover?}
//! DELETE /setups/{id}                  remove, refused while sessions use it
//! GET    /sessions                     list (id, provider, title, model, status, last activity)
//! GET    /sessions/{id}                one session, for refetching after a change
//! POST   /sessions                     spawn {setup, provider, title?, model?, cwd?, params?}
//! GET    /sessions/{id}/events?after=N SSE: backlog after N, then live
//!                                     (a backlog past CATCH_UP_LIMIT arrives as a
//!                                      `reset` frame plus the newest window)
//! POST   /sessions/{id}/message        {text, attachmentIds?}
//!                                     (starts the process first if it has exited)
//! POST   /sessions/{id}/answer         {questionId, answers}  (questions and permissions)
//! POST   /sessions/{id}/interrupt      stop the running turn; the process stays
//! POST   /sessions/{id}/stop           end the process; the session and transcript stay
//! POST   /sessions/{id}/start          run the process again, continuing the conversation
//! POST   /sessions/{id}/title          {title}
//! POST   /sessions/{id}/model          {model}
//! POST   /sessions/{id}/command        {text}  -- /compact, /clear, /rename x, or the dialect's own
//!                                     (starts the process first if it has exited)
//! POST   /sessions/{id}/compact
//! POST   /sessions/{id}/attachments    multipart image upload -> {id}, referenced by /message
//! GET    /sessions/{id}/files/{name}   images the session produced or was sent
//! DELETE /sessions/{id}                kill process, delete transcript + files
//! POST   /sessions/{id}/notify         {notify}  -- announce this one or not
//! GET    /notifications                SSE: every session's attention-wanting
//!                                      moments, live only (see `notifications`)
//! GET    /usage                        cached usage windows per provider
//! ```
//!
//! Later phases add: `GET|PUT /hosts` and `/models` -- see PLAN.md's table.
//!
//! Everything here works purely in the common event model; nothing may
//! branch on the session kind (that's what drivers are for).
//!
//! **Every request body in this module refuses fields it does not know**
//! (`serde(deny_unknown_fields)`), and a new one is expected to do the
//! same. Silently ignoring a field is the worst available answer: a caller
//! that misspells `permissionMode` got a 200 and a session running in the
//! default permission mode, which is indistinguishable from success at the
//! place they are looking. It cost an hour here, chasing a "startup race"
//! that was a snake_case key serde had dropped on the floor. Query strings
//! are deliberately left permissive -- a stale link carrying an extra
//! parameter is not a mistake worth failing a request over.

use std::convert::Infallible;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::Context;

use axum::Router;
use axum::extract::{Path as UrlPath, Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
use serde::Deserialize;
use tokio::sync::{broadcast, mpsc};
use tokio_stream::StreamExt;
use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};

use crate::session::driver::SessionCommand;
use crate::session::transcript::{CATCH_UP_LIMIT, CatchUp, SeqEvent, catch_up};
use crate::session::{LiveSession, SessionInfo, SessionManager, SpawnSpec};

pub fn router(manager: Arc<SessionManager>) -> Router {
    Router::new()
        .route("/setups", get(list_setups).post(add_setup))
        .route("/setups/probe", post(probe_setup))
        .route("/setups/{id}/importable", get(list_importable))
        .route(
            "/setups/{id}/importable/{session}",
            delete(delete_importable),
        )
        .route(
            "/setups/{id}",
            get(read_setup).put(update_setup).delete(delete_setup),
        )
        .route("/sessions", get(list_sessions).post(spawn_session))
        .route("/sessions/{id}", get(read_session).delete(delete_session))
        .route("/sessions/{id}/events", get(events))
        .route("/sessions/{id}/transcript", get(transcript))
        .route("/sessions/{id}/message", post(message))
        .route("/sessions/{id}/answer", post(answer))
        .route("/sessions/{id}/interrupt", post(interrupt))
        .route("/sessions/{id}/stop", post(stop))
        .route("/sessions/{id}/start", post(start))
        .route("/sessions/{id}/title", post(rename))
        .route("/sessions/{id}/model", post(set_model))
        .route("/sessions/{id}/permission-mode", post(set_permission_mode))
        .route("/sessions/{id}/notify", post(set_notify))
        .route("/notifications", get(notifications))
        .route("/sessions/{id}/compact", post(compact))
        .route("/sessions/{id}/command", post(command))
        .route("/sessions/{id}/attachments", post(upload_attachment))
        .route("/sessions/{id}/files/{name}", get(serve_file))
        // Phone photos overflow axum's 2 MB default body cap.
        .layer(axum::extract::DefaultBodyLimit::max(32 * 1024 * 1024))
        // An explicit fallback so the auth middleware (layered around the
        // whole router in main.rs) also covers unknown paths -- a scanner
        // gets the same 401 everywhere, never a route map.
        .fallback(|| async { ApiError::UnknownRoute })
        .with_state(manager)
}

#[derive(Debug, thiserror::Error)]
enum ApiError {
    #[error("{0}")]
    NotFound(String),
    #[error("no such route")]
    UnknownRoute,
    #[error("{0}")]
    BadRequest(String),
    #[error(transparent)]
    Internal(#[from] anyhow::Error),
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let status = match self {
            Self::NotFound(_) | Self::UnknownRoute => StatusCode::NOT_FOUND,
            Self::BadRequest(_) => StatusCode::BAD_REQUEST,
            Self::Internal(err) => {
                // The only variant whose real cause isn't safe to hand
                // back verbatim, and the only one worth a log line.
                tracing::error!("{err:#}");
                return StatusCode::INTERNAL_SERVER_ERROR.into_response();
            }
        };
        (status, self.to_string()).into_response()
    }
}

/// An `anyhow` error from a session mutation is a message written *for*
/// the phone ("no session abc123") -- not an internal fault, so it comes
/// back as a 400 with that message rather than a 500 and a log line.
fn bad_request(err: anyhow::Error) -> ApiError {
    ApiError::BadRequest(format!("{err:#}"))
}

fn lookup(manager: &SessionManager, id: &str) -> Result<Arc<LiveSession>, ApiError> {
    manager
        .session(id)
        .ok_or_else(|| ApiError::NotFound(format!("no session {id}")))
}

async fn list_sessions(State(manager): State<Arc<SessionManager>>) -> axum::Json<Vec<SessionInfo>> {
    axum::Json(manager.sessions())
}

/// One session's row, for a screen that has to show what is true now.
///
/// The list is a snapshot taken when somebody last looked at it, and a
/// screen opened from a row carries that snapshot with it. That is fine for
/// what a row *says* and wrong for what a control is *set to*: a switch
/// drawn from a stale row shows the position it had when the list was
/// fetched, which may be minutes and another device ago, and the person
/// reading it cannot tell. Same reason `GET /setups/{id}` exists.
async fn read_session(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<axum::Json<SessionInfo>, ApiError> {
    manager
        .sessions()
        .into_iter()
        .find(|session| session.id == id)
        .map(axum::Json)
        .ok_or_else(|| ApiError::NotFound(format!("no session {id}")))
}

/// What the spawn screen needs to render itself, so the phone holds no
/// hardcoded list: a setup added to `config.ron` shows up with no app
/// rebuild.
///
/// One list rather than two, because the choice is a pair and the halves
/// are not independent. A provider only exists on a machine that has it
/// installed, so listing providers and machines separately offered their
/// whole cross-product -- including "the Claude CLI on the box that hasn't
/// got it".
#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct SetupInfo {
    /// Stable; what a session stores and what these routes address.
    id: String,
    /// The editable label.
    name: String,
    /// Where it runs, for telling two setups apart. Absent for the one
    /// that is this machine.
    #[serde(skip_serializing_if = "Option::is_none")]
    address: Option<String>,
    providers: Vec<ProviderInfo>,
}

#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct ProviderInfo {
    name: String,
    kind: crate::config::DriverKind,
    models: Vec<String>,
}

async fn list_setups(State(manager): State<Arc<SessionManager>>) -> axum::Json<Vec<SetupInfo>> {
    axum::Json(manager.setups().into_iter().map(info_for).collect())
}

fn info_for(setup: crate::config::SetupConfig) -> SetupInfo {
    SetupInfo {
        id: setup.id,
        name: setup.name,
        address: setup.ssh.map(|ssh| ssh.address),
        providers: setup
            .providers
            .into_iter()
            .map(|provider| ProviderInfo {
                name: provider.name,
                kind: provider.kind,
                models: provider.models,
            })
            .collect(),
    }
}

/// How to reach a machine, as the phone describes it.
///
/// Note what is absent: nothing here names a program. Providers are found
/// by asking the machine (`crate::setups`), never sent, so the enrolled
/// token cannot introduce something to run.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct SshRequest {
    address: String,
    #[serde(default)]
    port: Option<u16>,
    /// A path on the *backend*, not a key itself: private keys do not
    /// travel, so this names one that must already be there.
    #[serde(default)]
    identity_file: Option<String>,
    #[serde(default)]
    options: Vec<String>,
}

impl SshRequest {
    /// Tidied at the boundary rather than stored as typed -- this came
    /// from a phone keyboard, so it may have a stray space or a `~`.
    fn into_config(self) -> Result<crate::config::SshConfig, ApiError> {
        let address = crate::setups::tidy(&self.address)
            .ok_or_else(|| ApiError::BadRequest("a machine needs an address".to_string()))?;
        Ok(crate::config::SshConfig {
            address,
            port: self.port,
            identity_file: self
                .identity_file
                .as_deref()
                .and_then(crate::setups::tidy)
                .map(std::path::PathBuf::from),
            options: self
                .options
                .iter()
                .filter_map(|o| crate::setups::tidy(o))
                .collect(),
        })
    }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct AddSetupRequest {
    name: String,
    /// Absent means this machine.
    #[serde(default)]
    ssh: Option<SshRequest>,
}

/// What a machine turned out to have, without saving anything.
///
/// The point of trying before committing: a wrong address or an
/// unauthorised key is caught while the person is still looking at the
/// form that caused it, rather than at the first spawn.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct ProbeRequest {
    #[serde(default)]
    ssh: Option<SshRequest>,
}

async fn probe_setup(
    axum::Json(body): axum::Json<ProbeRequest>,
) -> Result<axum::Json<Vec<ProviderInfo>>, ApiError> {
    let ssh = body.ssh.map(SshRequest::into_config).transpose()?;
    let providers = probe(ssh, "this setup").await?;
    Ok(axum::Json(
        providers
            .into_iter()
            .map(|provider| ProviderInfo {
                name: provider.name,
                kind: provider.kind,
                models: provider.models,
            })
            .collect(),
    ))
}

/// Asks the machine an `ssh` block describes -- or this one -- what it has.
///
/// `label` only ever appears in a failure message, so a probe of an
/// unsaved form can still say which machine would not answer.
async fn probe(
    ssh: Option<crate::config::SshConfig>,
    label: &str,
) -> Result<Vec<crate::config::ProviderConfig>, ApiError> {
    let transport = match ssh {
        Some(ssh) => crate::session::transport::Transport::Ssh {
            name: label.to_string(),
            ssh,
        },
        None => crate::session::transport::Transport::Here,
    };
    crate::setups::discover(&transport)
        .await
        .map_err(bad_request)
}

async fn add_setup(
    State(manager): State<Arc<SessionManager>>,
    axum::Json(body): axum::Json<AddSetupRequest>,
) -> Result<axum::Json<SetupInfo>, ApiError> {
    let ssh = body.ssh.map(SshRequest::into_config).transpose()?;
    // Ask the machine being added what it has, before writing anything --
    // so a bad address fails here rather than leaving a setup that can
    // never spawn.
    let providers = probe(ssh.clone(), &body.name).await?;
    let setup = manager
        .add_setup(&body.name, ssh, providers)
        .map_err(bad_request)?;
    Ok(axum::Json(info_for(setup)))
}

/// One setup by id, or the 404 that says so.
///
/// Three handlers ask this same question; the answer, and the wording of
/// the refusal, belong in one place.
fn setup_by_id(
    manager: &Arc<SessionManager>,
    id: &str,
) -> Result<crate::config::SetupConfig, ApiError> {
    manager
        .setups()
        .into_iter()
        .find(|setup| setup.id == id)
        .ok_or_else(|| ApiError::NotFound(format!("no setup {id}")))
}

async fn read_setup(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<axum::Json<SetupInfo>, ApiError> {
    setup_by_id(&manager, &id).map(|setup| axum::Json(info_for(setup)))
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct UpdateSetupRequest {
    #[serde(default)]
    name: Option<String>,
    /// Ask the machine again what it has -- after installing something
    /// there, or when a binary moved.
    #[serde(default)]
    rediscover: bool,
}

async fn update_setup(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<UpdateSetupRequest>,
) -> Result<axum::Json<SetupInfo>, ApiError> {
    let providers = if body.rediscover {
        let existing = manager
            .setups()
            .into_iter()
            .find(|setup| setup.id == id)
            .ok_or_else(|| ApiError::NotFound(format!("no setup {id}")))?;
        let transport = crate::session::transport::Transport::for_setup(&existing);
        Some(
            crate::setups::discover(&transport)
                .await
                .map_err(bad_request)?,
        )
    } else {
        None
    };
    let setup = manager
        .update_setup(&id, body.name.as_deref(), providers)
        .map_err(bad_request)?;
    Ok(axum::Json(info_for(setup)))
}

async fn delete_setup(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<StatusCode, ApiError> {
    manager.delete_setup(&id).map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct SpawnRequest {
    /// Which machine, and which of the things it offers.
    setup: String,
    provider: String,
    #[serde(default)]
    title: Option<String>,
    #[serde(default)]
    model: Option<String>,
    #[serde(default)]
    cwd: Option<PathBuf>,
    #[serde(default)]
    permission_mode: Option<String>,
    /// Whatever the chosen driver understands -- llama.cpp's context size
    /// and sampling, for instance. Opaque here on purpose: see
    /// `SessionConfig::params`.
    #[serde(default)]
    params: std::collections::BTreeMap<String, String>,
    /// Continue a Claude Code session the machine already has, named by
    /// the id `GET /setups/{id}/importable` reported.
    ///
    /// An id and not a path, deliberately. The server looks the path up
    /// again among the sessions it enumerated, so an enrolled token cannot
    /// turn this field into "read me an arbitrary file" -- the same rule
    /// that keeps a provider's command out of `POST /setups`.
    #[serde(default)]
    import: Option<String>,
}

/// What a machine already has that could be continued.
async fn list_importable(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<axum::Json<Vec<crate::session::import::Importable>>, ApiError> {
    let setup = setup_by_id(&manager, &id)?;
    let transport = crate::session::transport::Transport::for_setup(&setup);
    let mut found = crate::session::import::list(&transport)
        .await
        .map_err(bad_request)?;
    // Anything this app is already continuing is not offered again. Left
    // out rather than shown-and-disabled, because it has not disappeared:
    // it is in the session list, which is where it now belongs. Absence
    // here means "already somewhere you can reach it", not "gone".
    //
    // Joined here because the importer knows about files and the manager
    // knows about sessions, and putting the two together is the route's
    // job rather than either one's.
    found.retain(|candidate| manager.session_driving(&candidate.id).is_none());
    Ok(axum::Json(found))
}

/// Removes a Claude Code session from a machine.
///
/// The transcript *is* the session, so this ends any chance of resuming
/// that conversation -- including from an ai-app session already importing
/// it. The phone confirms before calling this; the server does not
/// second-guess a decision somebody was shown the cost of.
async fn delete_importable(
    State(manager): State<Arc<SessionManager>>,
    UrlPath((id, session)): UrlPath<(String, String)>,
) -> Result<StatusCode, ApiError> {
    let setup = setup_by_id(&manager, &id)?;
    let transport = crate::session::transport::Transport::for_setup(&setup);
    crate::session::import::delete(&transport, &session)
        .await
        .map_err(bad_request)?;
    tracing::info!("deleted Claude Code session {session} from setup {id}");
    Ok(StatusCode::NO_CONTENT)
}

async fn spawn_session(
    State(manager): State<Arc<SessionManager>>,
    axum::Json(body): axum::Json<SpawnRequest>,
) -> Result<axum::Json<SessionInfo>, ApiError> {
    // Resolved before the spawn because both halves of it are the
    // machine's answer, not the phone's: which file that id names, and
    // what is in it.
    let seed = match &body.import {
        Some(want) => {
            let setup = setup_by_id(&manager, &body.setup)?;
            let transport = crate::session::transport::Transport::for_setup(&setup);
            let found = crate::session::import::list(&transport)
                .await
                .map_err(bad_request)?;
            let chosen = found
                .into_iter()
                .find(|candidate| &candidate.id == want)
                .ok_or_else(|| {
                    ApiError::NotFound(format!(
                        "setup \"{}\" has no Claude Code session {want} to import",
                        body.setup
                    ))
                })?;
            if let Some(existing) = manager.session_driving(want) {
                return Err(ApiError::BadRequest(format!(
                    "session {existing} is already continuing that one -- delete it first if you \
                     want a fresh copy. Deleting it here does not touch the conversation itself, \
                     only this app's view of it."
                )));
            }
            // Refused rather than warned about, because there is nothing
            // useful on the other side of it. Importing an open session
            // puts a second `--resume` on a file the first one is still
            // writing: the conversation gets duplicated into it, each copy
            // replays the other's writes as work done elsewhere, and the
            // adopted one is billed for re-reading the whole thing. On
            // 2026-08-29 that was 65 MB and 154 screenshots.
            if chosen.in_use == crate::session::import::InUse::Yes {
                return Err(ApiError::BadRequest(format!(
                    "{want} is open in a terminal right now. Importing it would put a second \
                     Claude Code on the same conversation, which duplicates it and re-reads the \
                     whole thing. Close it there first, then import it here."
                )));
            }
            let records = crate::session::import::read_tail(&transport, &chosen.path)
                .await
                .map_err(bad_request)?;
            // The recorded directory can outlive itself; resuming into one
            // that is gone fails at `cd` before the CLI starts. Starting
            // somewhere real keeps the conversation, which is the point of
            // importing, and the log says which one was dropped.
            let mut chosen = chosen;
            if !crate::session::import::directory_exists(&transport, &chosen.cwd).await {
                tracing::warn!(
                    "imported session {} recorded {} as its directory, which is not there any \
                     more -- starting in the default one instead",
                    chosen.id,
                    chosen.cwd,
                );
                chosen.cwd = String::new();
            }
            Some((chosen, records))
        }
        None => None,
    };

    let spec = SpawnSpec {
        setup: body.setup,
        provider: body.provider,
        // An imported session is recognised by what it was about, so its
        // opening message is the title unless one was typed.
        // Blank normalised to absent here rather than trusted as a
        // choice. A client with nothing to say sends `""`, which is
        // `Some` and so satisfied `or_else` -- the imported session's real
        // title was computed, then discarded in favour of the "<provider>
        // session" fallback, so every import arrived called "claude-cli
        // session". Absent and empty mean the same thing to a person and
        // have to mean the same thing here.
        title: body
            .title
            .filter(|title| !title.trim().is_empty())
            .or_else(|| {
                seed.as_ref()
                    .map(|(chosen, _)| chosen.title.clone())
                    .filter(|title| !title.trim().is_empty())
            }),
        model: body.model,
        // Resumed where it was working, so the CLI picks up the same tree.
        cwd: body.cwd.or_else(|| {
            seed.as_ref()
                .map(|(chosen, _)| PathBuf::from(&chosen.cwd))
                .filter(|cwd| cwd.as_os_str() != "")
        }),
        permission_mode: body.permission_mode,
        params: body.params,
    };

    let info = match seed {
        Some((chosen, records)) => {
            tracing::info!(
                "importing Claude Code session {} ({} lines replayed)",
                chosen.id,
                records.lines().count()
            );
            manager.spawn_imported(
                spec,
                crate::session::Seed {
                    resume: chosen.id.clone(),
                    // Where it came from and how far it has been shown, so
                    // the session keeps itself level with the file a
                    // terminal is also writing to.
                    cursor: crate::session::import::Cursor {
                        path: chosen.path,
                        lines: chosen.lines,
                    },
                    records,
                },
            )
        }
        None => manager.spawn_session(spec),
    }
    .map_err(bad_request)?;
    tracing::info!(
        "spawned {} session {} ({})",
        info.provider,
        info.id,
        info.title
    );
    Ok(axum::Json(info))
}

async fn delete_session(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<StatusCode, ApiError> {
    manager.delete_session(&id).map_err(bad_request)?;
    tracing::info!("deleted session {id}");
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct MessageRequest {
    text: String,
    /// Ids from `POST /attachments`, uploaded before the message that
    /// references them.
    #[serde(default)]
    attachment_ids: Vec<String>,
}

async fn message(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<MessageRequest>,
) -> Result<StatusCode, ApiError> {
    // For the 404 a session that is not here has always answered with; the
    // send itself goes through the manager, which may have to start a
    // process before there is anything to send to.
    lookup(&manager, &id)?;
    if body.text.trim().is_empty() && body.attachment_ids.is_empty() {
        return Err(ApiError::BadRequest("message is empty".to_string()));
    }
    manager
        .send_message(&id, body.text, body.attachment_ids)
        .map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct AnswerRequest {
    question_id: String,
    /// Everything chosen, in the order it was offered. A question that
    /// takes one answer sends a list of one, so there is one shape here
    /// rather than a single-answer route and a multi-answer route beside
    /// it.
    answers: Vec<String>,
}

async fn answer(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<AnswerRequest>,
) -> Result<StatusCode, ApiError> {
    if body.answers.is_empty() {
        return Err(bad_request(anyhow::anyhow!(
            "an answer needs at least one choice"
        )));
    }
    lookup(&manager, &id)?.answer_question(&body.question_id, &body.answers);
    Ok(StatusCode::NO_CONTENT)
}

async fn interrupt(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<StatusCode, ApiError> {
    lookup(&manager, &id)?.interrupt();
    Ok(StatusCode::NO_CONTENT)
}

/// Ends the session's process. The session stays, and `start` brings it
/// back -- see [`SessionManager::stop_session`].
///
/// Not `lookup`ed: a session that failed to relaunch has no live entry and
/// may still have a process running, which is exactly one worth being able
/// to stop.
async fn stop(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<StatusCode, ApiError> {
    manager.stop_session(&id).map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

/// Starts a process for a session that has none, continuing the same
/// conversation -- see [`SessionManager::start_session`], which refuses
/// unless the session is known to have exited.
async fn start(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<StatusCode, ApiError> {
    manager.start_session(&id).map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

/// The usage screen needs two things that live in different places: the
/// cache, and the current list of machines to ask. Carried together rather
/// than the monitor holding the manager, which would point the dependency
/// upward -- `usage` sits below the session layer and should not reach
/// into it.
#[derive(Clone)]
pub struct UsageState {
    monitor: Arc<crate::usage::UsageMonitor>,
    manager: Arc<SessionManager>,
}

/// Separate router because its state is the usage monitor, not the
/// session manager; merged (and auth-wrapped) with the rest in `main`.
pub fn usage_router(
    monitor: Arc<crate::usage::UsageMonitor>,
    manager: Arc<SessionManager>,
) -> Router {
    Router::new()
        .route("/usage", get(usage))
        .with_state(UsageState { monitor, manager })
}

async fn usage(
    State(state): State<UsageState>,
) -> Result<axum::Json<Vec<crate::usage::UsageSnapshot>>, ApiError> {
    // Read here rather than inside the fetch, so the list of machines is
    // the one that existed when the request arrived and cannot change
    // under a fetch that takes an ssh round trip per machine.
    let setups = state.manager.setups();
    // The fetch is blocking by design (see `usage`); off the workers.
    let snapshots = tokio::task::spawn_blocking(move || state.monitor.snapshots(&setups))
        .await
        .context("usage fetch panicked")?;
    Ok(axum::Json(snapshots))
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TitleRequest {
    title: String,
}

async fn rename(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<TitleRequest>,
) -> Result<StatusCode, ApiError> {
    manager
        .rename_session(&id, &body.title)
        .map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ModelRequest {
    model: String,
}

async fn set_model(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<ModelRequest>,
) -> Result<StatusCode, ApiError> {
    manager
        .set_session_model(&id, &body.model)
        .map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(deny_unknown_fields)]
struct PermissionModeRequest {
    mode: String,
}

async fn set_permission_mode(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<PermissionModeRequest>,
) -> Result<StatusCode, ApiError> {
    manager
        .set_session_permission_mode(&id, &body.mode)
        .map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct NotifyRequest {
    notify: bool,
}

async fn set_notify(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<NotifyRequest>,
) -> Result<StatusCode, ApiError> {
    manager
        .set_session_notify(&id, body.notify)
        .map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CommandRequest {
    text: String,
}

/// Runs one of the session's own commands, now or at the next boundary.
///
/// The two this server understands are turned into the operations it has
/// -- a compaction, a rename, which is also how the settings screen asks
/// -- and everything else is passed to the session verbatim, because a
/// dialect's vocabulary is its own and grows without this file.
async fn command(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    axum::Json(body): axum::Json<CommandRequest>,
) -> Result<StatusCode, ApiError> {
    let text = body.text.trim();
    let (name, rest) = match text.split_once(char::is_whitespace) {
        Some((name, rest)) => (name, rest.trim()),
        None => (text, ""),
    };
    // All of these start the session's process first if it has exited: a
    // command is something somebody asked the session to do, and answering
    // that its process is gone hands back the work of starting one.
    //
    // A rename still goes through `rename_session` rather than being a
    // command like the rest, because the name is persisted and listed as
    // well as forwarded, and that is one operation. It starts a process
    // too, and for a sharper reason than the others -- see there.
    let command = match (name, rest) {
        ("/compact", _) => SessionCommand::Compact,
        ("/clear", _) => SessionCommand::Clear,
        ("/rename", "") => return Err(bad_request(anyhow::anyhow!("a session needs a name"))),
        ("/rename", title) => {
            manager.rename_session(&id, title).map_err(bad_request)?;
            return Ok(StatusCode::NO_CONTENT);
        }
        _ => SessionCommand::Raw(text.to_string()),
    };
    lookup(&manager, &id)?;
    manager.run_command(&id, command).map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

async fn compact(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
) -> Result<StatusCode, ApiError> {
    lookup(&manager, &id)?;
    manager
        .run_command(&id, SessionCommand::Compact)
        .map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}

/// Accepts one image (any multipart field) and stores it under the
/// session; the returned id goes into a later `/message`'s attachmentIds.
async fn upload_attachment(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    mut multipart: axum::extract::Multipart,
) -> Result<axum::Json<serde_json::Value>, ApiError> {
    let session = lookup(&manager, &id)?;
    let field = multipart
        .next_field()
        .await
        .map_err(|err| ApiError::BadRequest(format!("bad upload: {err}")))?
        .ok_or_else(|| ApiError::BadRequest("no file in the upload".to_string()))?;
    let content_type = field.content_type().unwrap_or("image/jpeg").to_string();
    let bytes = field
        .bytes()
        .await
        .map_err(|err| ApiError::BadRequest(format!("upload read failed: {err}")))?;
    let name = session
        .save_attachment(&bytes, &content_type)
        .map_err(bad_request)?;
    Ok(axum::Json(serde_json::json!({ "id": name })))
}

/// Serves a session's stored images -- both `files/` (produced by tools)
/// and `attachments/` (uploaded from the phone), by the id events and
/// uploads reference.
async fn serve_file(
    State(manager): State<Arc<SessionManager>>,
    UrlPath((id, name)): UrlPath<(String, String)>,
) -> Result<Response, ApiError> {
    // Ids are server-generated hex + extension; anything else (and any
    // path separator in particular) is refused, not resolved.
    if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '.') || name.contains("..") {
        return Err(ApiError::BadRequest("invalid file id".to_string()));
    }
    let session = lookup(&manager, &id)?;
    let candidates = [
        session.dir().join("files").join(&name),
        session.dir().join("attachments").join(&name),
    ];
    let Some(path) = candidates.iter().find(|path| path.is_file()) else {
        return Err(ApiError::NotFound(format!(
            "no file {name} in session {id}"
        )));
    };
    // A file that is there but unreadable is this server's fault, not the
    // request's -- Internal logs it and says nothing more to the caller.
    let bytes = std::fs::read(path)
        .with_context(|| format!("read {}", path.display()))
        .map_err(ApiError::Internal)?;
    // Names are server-generated, so an unrecognized extension can only
    // mean a file this server didn't write.
    let content_type = crate::media::media_type_for(&name).unwrap_or("image/jpeg");
    Ok(([(axum::http::header::CONTENT_TYPE, content_type)], bytes).into_response())
}

#[derive(Deserialize)]
struct EventsQuery {
    #[serde(default)]
    after: u64,
}

/// The session screen's one data source: replay everything after the
/// cursor from the transcript, then live events as they happen. An SSE
/// auto-reconnect sends the last event id it saw as `Last-Event-ID`, which
/// takes precedence over `after` -- same cursor, native mechanism.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct TranscriptQuery {
    /// Page backwards from this sequence number; absent means the newest.
    #[serde(default)]
    before: Option<u64>,
    #[serde(default = "default_window")]
    limit: usize,
}

fn default_window() -> usize {
    80
}

/// A page of a session's transcript, newest first to open with.
///
/// One request rather than one stream frame per event. The SSE stream
/// stays as it is and remains the right shape for *live* events, which
/// arrive one at a time by nature; it is only the backlog that has to
/// stop pretending to be live.
async fn transcript(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    Query(query): Query<TranscriptQuery>,
) -> Result<axum::Json<Vec<crate::session::transcript::SeqEvent>>, ApiError> {
    let session = lookup(&manager, &id)?;
    let events = crate::session::transcript::read_window(
        session.transcript_path(),
        query.before,
        query.limit,
    )
    .map_err(bad_request)?;
    Ok(axum::Json(events))
}

async fn events(
    State(manager): State<Arc<SessionManager>>,
    UrlPath(id): UrlPath<String>,
    Query(query): Query<EventsQuery>,
    headers: HeaderMap,
) -> Result<Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
    let session = lookup(&manager, &id)?;
    let cursor = headers
        .get("last-event-id")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse().ok())
        .unwrap_or(query.after);

    // Subscribe before reading the file so nothing can land in the gap
    // between replay and live; overlap is deduplicated by seq.
    let live = session.subscribe();
    let (tx, stream) = mpsc::channel(64);
    tokio::spawn(stream_session(
        session.transcript_path().to_path_buf(),
        cursor,
        live,
        tx,
    ));
    Ok(Sse::new(ReceiverStream::new(stream).map(Ok)).keep_alive(KeepAlive::default()))
}

/// Every session's attention-wanting moments, on one stream.
///
/// **Live only, with no cursor**, which is the one place this server does not
/// offer to catch a client up. A notification is a claim about now: replaying
/// "your turn" from an hour ago tells somebody to go and look at a session
/// that may have been answered from another device since, and a notification
/// that is wrong is worse than one that never came -- it costs the reader the
/// trip *and* teaches them to distrust the next one. What was missed while
/// disconnected is still on the session list, which is the surface that
/// answers "what is waiting" without claiming to be news.
async fn notifications(
    State(manager): State<Arc<SessionManager>>,
) -> Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>> {
    let live = manager.subscribe_notifications();
    let stream = BroadcastStream::new(live).filter_map(|item| {
        // A lagged subscriber has lost the oldest notifications, and there is
        // nothing useful to say about that: the ones it still gets are the
        // recent ones, which are the ones worth acting on.
        let notification = item.ok()?;
        Some(Ok(SseEvent::default().json_data(&notification).ok()?))
    });
    Sse::new(stream).keep_alive(KeepAlive::default())
}

/// Feeds one SSE subscriber: transcript replay after the cursor, then live
/// events, catching back up from the file whenever the broadcast channel
/// laps us. Ends when the client disconnects (send fails) or the session
/// is deleted (channel closed).
async fn stream_session(
    transcript: PathBuf,
    mut last: u64,
    mut live: broadcast::Receiver<SeqEvent>,
    tx: mpsc::Sender<SseEvent>,
) {
    if !send_backlog(&transcript, &mut last, &tx).await {
        return;
    }
    loop {
        match live.recv().await {
            Ok(entry) => {
                if entry.seq <= last {
                    continue;
                }
                last = entry.seq;
                if send_event(&tx, &entry).await.is_err() {
                    return;
                }
            }
            Err(broadcast::error::RecvError::Lagged(_)) => {
                if !send_backlog(&transcript, &mut last, &tx).await {
                    return;
                }
            }
            Err(broadcast::error::RecvError::Closed) => return,
        }
    }
}

/// Sends everything after `last`, advancing it, and answers whether the
/// subscriber is still there.
///
/// A [`CatchUp::Restart`] is preceded by the `reset` frame that tells the
/// client to drop what it holds. Without it the window would be spliced
/// onto rows that are no longer adjacent to it, which reads as ordinary
/// output rather than as a gap -- which is why a bounded backlog cannot
/// simply be "the newest events".
///
/// Both ways into a backlog come through here -- the first replay and the
/// recovery from a lapped broadcast -- because either can be arbitrarily
/// far behind and owes the client the same answer.
///
/// Synchronous file reads from an async task: transcript lines are small
/// and local; revisit if daily use produces transcripts where this shows
/// (phase 6 territory).
async fn send_backlog(transcript: &Path, last: &mut u64, tx: &mpsc::Sender<SseEvent>) -> bool {
    let entries = match catch_up(transcript, *last, CATCH_UP_LIMIT) {
        Ok(CatchUp::Continue(entries)) => entries,
        Ok(CatchUp::Restart(entries)) => {
            if tx.send(SseEvent::default().event("reset")).await.is_err() {
                return false;
            }
            entries
        }
        Err(err) => {
            tracing::error!("transcript replay failed: {err:#}");
            return false;
        }
    };
    for entry in entries {
        *last = entry.seq;
        if send_event(tx, &entry).await.is_err() {
            return false;
        }
    }
    true
}

async fn send_event(
    tx: &mpsc::Sender<SseEvent>,
    entry: &SeqEvent,
) -> Result<(), mpsc::error::SendError<SseEvent>> {
    let data = serde_json::to_string(entry).expect("events always serialize");
    tx.send(SseEvent::default().id(entry.seq.to_string()).data(data))
        .await
}

/// Separate router because its state is the model store, like `usage`'s.
///
/// Keys are `owner/repo/file.gguf` and so contain slashes, which is why
/// nothing here puts one in the path: a key travels in the body or a query
/// string, and the routes stay addressable without escaping rules nobody
/// would get right from a phone.
pub fn models_router(store: Arc<crate::models::ModelStore>) -> Router {
    Router::new()
        .route("/models", get(list_models))
        .route("/models/search", get(search_models))
        .route("/models/files", get(repo_files))
        .route("/models/download", post(start_download))
        .route("/models/cancel", post(cancel_download))
        .route("/models/delete", post(delete_model))
        .with_state(store)
}

/// What this machine has and what it is fetching, in one answer.
///
/// Both together deliberately: a phone showing the model list needs both
/// to draw one screen, and two routes would let it render a model as
/// absent while its download sits at 99%.
#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct ModelsResponse {
    local: Vec<crate::models::LocalModel>,
    downloads: Vec<crate::models::DownloadStatus>,
}

async fn list_models(
    State(store): State<Arc<crate::models::ModelStore>>,
) -> Result<axum::Json<ModelsResponse>, ApiError> {
    let listing = tokio::task::spawn_blocking(move || ModelsResponse {
        local: store.list(),
        downloads: store.downloads(),
    })
    .await
    .context("listing models panicked")?;
    Ok(axum::Json(listing))
}

#[derive(Deserialize)]
struct SearchQuery {
    q: String,
}

async fn search_models(
    Query(query): Query<SearchQuery>,
) -> Result<axum::Json<Vec<crate::models::RemoteRepo>>, ApiError> {
    // Blocking HTTP, like the usage fetch: off the request workers.
    let found = tokio::task::spawn_blocking(move || crate::models::search(&query.q))
        .await
        .context("model search panicked")?
        .map_err(bad_request)?;
    Ok(axum::Json(found))
}

#[derive(Deserialize)]
struct RepoQuery {
    repo: String,
}

async fn repo_files(
    State(store): State<Arc<crate::models::ModelStore>>,
    Query(query): Query<RepoQuery>,
) -> Result<axum::Json<Vec<crate::models::RemoteFile>>, ApiError> {
    let files = tokio::task::spawn_blocking(move || crate::models::files(&query.repo, &store))
        .await
        .context("listing repository files panicked")?
        .map_err(bad_request)?;
    Ok(axum::Json(files))
}

#[derive(Deserialize)]
struct DownloadRequest {
    repo: String,
    file: String,
}

/// Starts a download, or rejoins the one already running for that model.
async fn start_download(
    State(store): State<Arc<crate::models::ModelStore>>,
    axum::Json(body): axum::Json<DownloadRequest>,
) -> Result<axum::Json<crate::models::DownloadStatus>, ApiError> {
    let status = store.start(&body.repo, &body.file).map_err(bad_request)?;
    Ok(axum::Json(status))
}

#[derive(Deserialize)]
struct KeyRequest {
    key: String,
}

async fn cancel_download(
    State(store): State<Arc<crate::models::ModelStore>>,
    axum::Json(body): axum::Json<KeyRequest>,
) -> Result<axum::Json<crate::models::DownloadStatus>, ApiError> {
    let status = store.cancel(&body.key).map_err(bad_request)?;
    Ok(axum::Json(status))
}

async fn delete_model(
    State(store): State<Arc<crate::models::ModelStore>>,
    axum::Json(body): axum::Json<KeyRequest>,
) -> Result<StatusCode, ApiError> {
    store.delete(&body.key).map_err(bad_request)?;
    Ok(StatusCode::NO_CONTENT)
}