//! 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 //! GET /setups/{id}/models GGUFs on that machine, for a llama session //! GET /setups/{id}/dir?path=P entries of directory P, and P resolved //! GET /setups/{id}/file?path=P content of file P, or why not //! PUT /setups/{id}/file {path, content, ifSha256} -> new size/mtime/sha256 //! (409 when the file no longer matches ifSha256) //! POST /setups/{id}/file {path} create empty; refused if it exists //! POST /setups/{id}/dir {path} create; refused if it exists //! GET /setups/{id}/importable Claude Code sessions on it that could be continued //! POST /setups/{id}/importable/import {sessions} -> 202; runs on the server //! POST /setups/{id}/importable/delete {sessions} -> 202; removes the machine's transcripts //! GET /setups/{id}/importable/events SSE: what is in flight against them //! 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) //! GET /sessions/{id}/transcript a page of history: ?before=N (newest when absent), //! ?limit=N, ?coalesce=true to count rows not deltas, //! ?after=N to floor it at what the caller already holds //! POST /sessions/{id}/message {text, attachmentIds?} //! (starts the process first if it has exited) //! POST /sessions/{id}/unqueue {messageId} -- take back one not read yet //! (409 when the session already has it) //! 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}/cwd {cwd} -- move it; stops the process, //! which starts again in the new one //! POST /sessions/{id}/model {model} //! POST /sessions/{id}/permission-mode {permissionMode} //! POST /sessions/{id}/effort {effort} -- null for the CLI's default; //! settled at launch, so this stops the process //! 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 upload, image or any file -> {id}, referenced by /message //! GET /sessions/{id}/files/{name} images the session produced, and what it was sent //! DELETE /sessions/{id} kill process, delete transcript + files //! (?deleteForeign=true removes the machine's own copy too) //! POST /sessions/{id}/notify {notify} -- announce this one or not //! POST /sessions/{id}/auto-resume {autoResume, message?} -- carry on by itself //! once the account's usage limit lifts //! GET /notifications SSE: every session's attention-wanting //! moments, live only (see `notifications`) //! GET /defaults {effort} -- what a new session starts at //! POST /defaults {effort} -- null for the CLI's own default //! GET /usage cached usage windows per provider //! GET /models downloaded GGUFs, and what is being fetched //! GET /models/search?q=Q HuggingFace repositories matching Q //! GET /models/files?repo=R the GGUFs in one repository //! POST /models/download {repo, file}; rejoins the run already going //! POST /models/cancel {key}; the partial stays, so starting again resumes //! POST /models/delete {key} //! ``` //! //! 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::{get, post}; use serde::{Deserialize, Serialize}; use tokio::sync::{broadcast, mpsc}; use tokio_stream::StreamExt; use tokio_stream::wrappers::{BroadcastStream, ReceiverStream}; use crate::session::driver::{SessionCommand, Unqueued}; use crate::session::pending::Operation; use crate::session::transcript::{CATCH_UP_LIMIT, CatchUp, SeqEvent, catch_up}; use crate::session::{LiveSession, SessionInfo, SessionManager, SpawnSpec}; pub fn router(manager: Arc) -> Router { Router::new() .route("/setups", get(list_setups).post(add_setup)) .route("/setups/probe", post(probe_setup)) .route("/setups/{id}/importable", get(list_importable)) // A batch at a time, never a session at a time -- see // [`delete_importable`]. .route("/setups/{id}/importable/delete", post(delete_importable)) .route("/setups/{id}/importable/import", post(start_import)) .route("/setups/{id}/importable/events", get(importable_events)) .route( "/setups/{id}", get(read_setup).put(update_setup).delete(delete_setup), ) // The models on the machine a setup names, for a llama session there. .route("/setups/{id}/models", get(setup_models)) // The filesystem of the machine a setup names. Under the setup // rather than under a session because a filesystem is a property of // a machine; a session only says where to start looking. .route("/setups/{id}/dir", get(list_dir).post(create_dir)) .route( "/setups/{id}/file", get(read_file).put(write_file).post(create_file), ) .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}/unqueue", post(unqueue)) .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}/cwd", post(set_cwd)) .route("/sessions/{id}/model", post(set_model)) .route("/sessions/{id}/permission-mode", post(set_permission_mode)) .route("/sessions/{id}/effort", post(set_effort)) .route("/defaults", get(defaults).post(set_defaults)) .route("/sessions/{id}/notify", post(set_notify)) .route("/sessions/{id}/auto-resume", post(set_auto_resume)) .route("/notifications", get(notifications)) .route("/sessions/{id}/compact", post(compact)) .route("/sessions/{id}/command", post(command)) .route( "/sessions/{id}/attachments", // A trace or a log is bigger than a photo; the cap below is for // everything else, and the innermost limit is the one axum // applies. post(upload_attachment).layer(axum::extract::DefaultBodyLimit::max(ATTACHMENT_LIMIT)), ) .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 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), /// The request was understood and the state it names has moved on -- /// distinct from `BadRequest`, which is a caller that got it wrong. #[error("{0}")] Conflict(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::Conflict(_) => StatusCode::CONFLICT, 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"), 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, ApiError> { manager .session(id) .ok_or_else(|| ApiError::NotFound(format!("no session {id}"))) } async fn list_sessions(State(manager): State>) -> axum::Json> { 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. 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, and the /// person reading it cannot tell. Same reason `GET /setups/{id}` exists. async fn read_session( State(manager): State>, UrlPath(id): UrlPath, ) -> Result, 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 halves are not independent. A /// provider only exists on a machine that has it installed, so listing them /// separately offered the 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 this machine. #[serde(skip_serializing_if = "Option::is_none")] address: Option, providers: Vec, } #[derive(serde::Serialize)] #[serde(rename_all = "camelCase")] struct ProviderInfo { name: String, kind: crate::config::DriverKind, models: Vec, } async fn list_setups(State(manager): State>) -> axum::Json> { 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, /// 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, #[serde(default)] options: Vec, /// Where attached files land on that machine; see `SshConfig`. #[serde(default)] attachments_dir: Option, /// Where that machine keeps its GGUF models; see `SshConfig`. #[serde(default)] models_dir: Option, } 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 { 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(), // Not `tidy`: that expands `~` to *this* machine's home, and this // path is on the other one. The remote shell expands it there. attachments_dir: self .attachments_dir .as_deref() .map(str::trim) .filter(|dir| !dir.is_empty()) .map(std::path::PathBuf::from), // The same rule, and for the same reason: this directory is // on the other machine, so a `~` in it is that machine's home. models_dir: self .models_dir .as_deref() .map(str::trim) .filter(|dir| !dir.is_empty()) .map(std::path::PathBuf::from), }) } } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct AddSetupRequest { name: String, /// Absent means this machine. #[serde(default)] ssh: Option, } /// 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, } async fn probe_setup( axum::Json(body): axum::Json, ) -> Result>, 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, label: &str, ) -> Result, 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>, axum::Json(body): axum::Json, ) -> Result, 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, id: &str, ) -> Result { 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>, UrlPath(id): UrlPath, ) -> Result, 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, /// 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>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result, 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>, UrlPath(id): UrlPath, ) -> Result { manager.delete_setup(&id).map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) } /// The five explorer routes below all begin the same way: find the machine, /// and check that what the phone named is a path this will act on. The check /// is `files::check_path`, shared with [`set_cwd`] -- one rule about what an /// acceptable path is, and one wording for refusing it. fn files_on( manager: &Arc, id: &str, path: &str, ) -> Result<(crate::session::transport::Transport, String), ApiError> { let setup = setup_by_id(manager, id)?; let path = crate::files::check_path(path).map_err(bad_request)?; Ok(( crate::session::transport::Transport::for_setup(&setup), path, )) } /// A failure from one of the scripts is the *machine's* message, written to /// be read where it happened, which is the phone. So it comes back as a 400 /// with those words rather than a 500 and a log line only the backend sees. fn from_machine(err: anyhow::Error) -> ApiError { ApiError::BadRequest(format!("{err:#}")) } /// Where a path is named for these routes. Query rather than a path segment: /// a path contains slashes, and a segment that had to be escaped and /// unescaped would be a second encoding to keep in step with the phone's. #[derive(Deserialize)] struct PathQuery { path: String, } /// The models **that machine** has, which is the list a llama.cpp session /// on it can choose from. /// /// Not `GET /models`, which is this backend's own downloads: those are on /// the machine a session runs on only when they are the same machine. A /// spawn screen offering this backend's list for a remote setup would be /// naming files that are not there, and the session would fail at the /// point of loading rather than at the point of choosing. async fn setup_models( State(manager): State>, UrlPath(id): UrlPath, ) -> Result>, ApiError> { let setup = setup_by_id(&manager, &id)?; let transport = crate::session::transport::Transport::for_setup(&setup); let dir = crate::models::dir_on(&transport, manager.models_dir()); crate::models::on_machine(&transport, &dir) .await .map(axum::Json) .map_err(from_machine) } /// What is in a directory, and what that directory resolved to. async fn list_dir( State(manager): State>, UrlPath(id): UrlPath, Query(query): Query, ) -> Result, ApiError> { let (transport, path) = files_on(&manager, &id, &query.path)?; crate::files::list(&transport, &path) .await .map(axum::Json) .map_err(from_machine) } /// One file's content, or which of the three reasons there is none. The path /// it was asked for rides along, so a phone that has moved on since can tell /// which answer this is. #[derive(Serialize)] struct FileResponse { path: String, #[serde(flatten)] read: crate::files::FileRead, } async fn read_file( State(manager): State>, UrlPath(id): UrlPath, Query(query): Query, ) -> Result, ApiError> { let (transport, path) = files_on(&manager, &id, &query.path)?; let read = crate::files::read(&transport, &path) .await .map_err(from_machine)?; Ok(axum::Json(FileResponse { path, read })) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct WriteFileRequest { path: String, content: String, /// The digest the read reported. Not optional: an editor that could omit /// it would be one overwrite away from losing an agent's edit, and "I did /// not check" is not something a caller should be able to say by leaving a /// field out. if_sha256: String, } async fn write_file( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result, ApiError> { let (transport, path) = files_on(&manager, &id, &body.path)?; crate::files::write( &transport, &path, &body.if_sha256, body.content.into_bytes(), ) .await .map_err(from_machine)? .map(axum::Json) .map_err(|crate::files::Stale| { ApiError::Conflict("this file changed on the machine since you opened it".to_string()) }) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct CreateRequest { path: String, } async fn create_file( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { let (transport, path) = files_on(&manager, &id, &body.path)?; crate::files::create_file(&transport, &path) .await .map_err(from_machine)?; Ok(StatusCode::NO_CONTENT) } async fn create_dir( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { let (transport, path) = files_on(&manager, &id, &body.path)?; crate::files::create_dir(&transport, &path) .await .map_err(from_machine)?; 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, #[serde(default)] model: Option, #[serde(default)] cwd: Option, #[serde(default)] permission_mode: Option, #[serde(default)] effort: Option, /// Whatever the chosen driver understands -- llama.cpp's context size and /// sampling. Opaque here on purpose: see `SessionConfig::params`. #[serde(default)] params: std::collections::BTreeMap, /// 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". #[serde(default)] import: Option, } /// What a machine already has that could be continued. async fn list_importable( State(manager): State>, UrlPath(id): UrlPath, ) -> Result>, 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. // // Except while this server is in the middle of importing it. A spawn // creates the session partway through, so the row would vanish the instant // the work started and reappear as a session only once it finished -- and // in between, the screen that asked for it would show nothing at all where // the thing it is waiting for used to be. found.retain(|candidate| { manager.pending().running(&id, &candidate.id).is_some() || manager.session_driving(&candidate.id).is_none() }); // What the server is doing to each of them, joined on here because a phone // that was asleep or freshly opened never heard the events -- see // `pending`. A row being imported has to stay visible, marked, or the list // would say the work never started. let present: Vec = found.iter().map(|row| row.id.clone()).collect(); manager.pending().prune(&id, &present); let rows: Vec = found .into_iter() .map(|importable| ImportableRow { pending: manager .pending() .running(&id, &importable.id) .map(|operation| operation.label()), error: manager.pending().failure(&id, &importable.id), importable, }) .collect(); Ok(axum::Json(rows)) } /// A row of the import list: what the machine has, plus what this server is /// doing to it. Flattened, so the two halves arrive as one object -- the phone /// is drawing one row and has no use for the seam. #[derive(Serialize)] #[serde(rename_all = "camelCase")] struct ImportableRow { #[serde(flatten)] importable: crate::session::import::Importable, /// The word the row shows while something is running: "importing" or /// "deleting". Absent when nothing is. #[serde(skip_serializing_if = "Option::is_none")] pending: Option<&'static str>, /// How the last attempt on this row failed, if it did. Kept until /// something replaces it, because the phone that needs to see it may not /// have been connected when it happened. #[serde(skip_serializing_if = "Option::is_none")] error: Option, } /// Removes Claude Code sessions from a machine. /// /// The transcript *is* the session, so this ends any chance of resuming those /// conversations. The phone confirms before calling this; the server does not /// second-guess a decision somebody was shown the cost of. /// /// A batch and never a single session, which is why this is a POST with a body /// rather than a `DELETE` on each id. One request per row made a handover only /// as atomic as the network: leave the screen, lose signal, or have the fourth /// of six requests fail, and some rows are being deleted while the rest are /// untouched, with nothing anywhere that knows the difference. Here every id /// is registered as in flight before the 202 goes back. /// /// Registering is what has to be atomic; the work is not. Each row settles on /// its own event from its own outcome, because six deletes that must all /// succeed or all roll back is not something a filesystem offers. async fn delete_importable( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { let setup = setup_by_id(&manager, &id)?; let transport = crate::session::transport::Transport::for_setup(&setup); // Registered before anything is spawned, so the 202 is only sent once // every row is already showing "deleting" -- a phone that refetches the // instant it gets the reply cannot catch a row that has not started. let running: Vec<(String, crate::session::pending::InFlight)> = body .sessions .iter() .map(|session| { ( session.clone(), manager.pending().begin(&id, session, Operation::Deleting), ) }) .collect(); let sessions = body.sessions; tokio::spawn(async move { // One failure here is the machine being unreachable, which is true of // every row rather than of any one of them. let outcomes = match crate::session::import::delete(&transport, &sessions).await { Ok(outcomes) => outcomes, Err(err) => { let message = format!("{err:#}"); tracing::warn!( "deleting {} sessions on {id} failed: {message}", running.len() ); for (_, flight) in running { flight.failed(message.clone()); } return; } }; for (session, flight) in running { match outcomes.get(&session) { Some(Ok(())) => { tracing::info!("deleting {session} on {id}: done"); flight.succeeded(); } Some(Err(message)) => { tracing::warn!("deleting {session} on {id} failed: {message}"); flight.failed(message.clone()); } // `delete` promises an entry per id, so this is a bug rather // than a state -- but a row stuck on "deleting" for ever is a // worse answer than one that says so. None => flight.failed(format!("nothing was reported about {session}")), } } }); Ok(StatusCode::ACCEPTED) } /// Which sessions to delete. See [`delete_importable`] for why it is a list. #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct DeleteBatch { sessions: Vec, } /// Continues Claude Code sessions, in the background. /// /// Separate from `POST /sessions` because the two are asked different /// questions. That one means "start this and take me to it", so it waits and /// answers with the session. This one is the import screen's batch: several at /// once, nobody waiting on any particular one, and the answer arrives as a row /// changing rather than as a reply -- the screen it was started from may well /// be gone by then. /// /// A list for the same reason [`delete_importable`] takes one. async fn start_import( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { // Checked before accepting, so an unknown machine is an error the caller // sees rather than one it has to go and read off a row. setup_by_id(&manager, &id)?; for session in body.sessions { let request = SpawnRequest { setup: id.clone(), provider: body.provider.clone(), // Nothing to say: `spawn` titles an import from the session it // continues, and the cwd comes from the same place. title: None, model: body.model.clone(), cwd: None, permission_mode: body.permission_mode.clone(), effort: body.effort.clone(), params: std::collections::BTreeMap::new(), import: Some(session.clone()), }; let inner = Arc::clone(&manager); in_background( &manager, id.clone(), session, Operation::Importing, async move { spawn(&inner, request) .await .map(|_| ()) .map_err(|err| anyhow::anyhow!("{err}")) }, ); } Ok(StatusCode::ACCEPTED) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct ImportRequest { /// The sessions to continue, all with the settings below -- they were /// picked together on one screen, so there is nothing to say per row. sessions: Vec, provider: String, #[serde(default)] model: Option, #[serde(default)] permission_mode: Option, #[serde(default)] effort: Option, } /// Runs `work` on the server, marked as in flight for as long as it takes. /// Spawned rather than awaited, which is the whole difference: the phone asked /// for it, but the phone leaving must not cancel it. What replaces the reply /// is the pending registry. fn in_background( manager: &Arc, setup: String, session: String, operation: Operation, work: F, ) where F: std::future::Future> + Send + 'static, { let running = manager.pending().begin(&setup, &session, operation); tokio::spawn(async move { match work.await { Ok(()) => { tracing::info!("{} {session} on {setup}: done", operation.label()); running.succeeded(); } Err(err) => { tracing::warn!("{} {session} on {setup} failed: {err:#}", operation.label()); // The server's own words, the way every other failure in this // app reaches a person. running.failed(format!("{err:#}")); } } }); } /// Every change to what is in flight against one machine. Scoped to the setup /// the screen is showing, the same way a session's events are scoped to that /// session. async fn importable_events( State(manager): State>, UrlPath(id): UrlPath, ) -> Sse>> { let live = manager.pending().subscribe(); let stream = BroadcastStream::new(live).filter_map(move |item| { // A lagged subscriber has missed changes it cannot get back here, and // that is what the listing is for: the screen refetches on arrival and // carries the truth whatever this stream missed. let change = item.ok()?; if change.setup() != id { return None; } Some(Ok(SseEvent::default().json_data(&change).ok()?)) }); Sse::new(stream).keep_alive(KeepAlive::default()) } async fn spawn_session( State(manager): State>, axum::Json(body): axum::Json, ) -> Result, ApiError> { spawn(&manager, body).await.map(axum::Json) } /// Starts a session, continuing a Claude Code one where `body.import` names /// it. /// /// A function rather than only a handler because the import screen's batch runs /// this from a background task. Spawning has to mean exactly the same thing /// either way: the same refusal when something else already has the /// conversation open, the same title, the same working directory. async fn spawn(manager: &Arc, body: SpawnRequest) -> Result { // Resolved before the spawn because both halves 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 chosen = crate::session::import::find(&transport, want) .await .map_err(bad_request)? .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 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, and resuming into one // that is gone fails at `cd` before the CLI starts. Starting // somewhere real keeps the conversation, 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 rather than trusted as a choice: a client with nothing to // say sends `""`, which is `Some` and so satisfied `or_else`, and every // import arrived called "claude-cli session". 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, effort: body.effort, 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(info) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct DeleteSessionQuery { /// Also remove the machine's own transcript of this conversation -- the /// file Claude Code keeps under `~/.claude/projects`, which this server's /// delete does not otherwise touch. /// /// Off by default, because the two deletes differ in what they cost: /// leaving the machine's copy is recoverable and removing it is not, and a /// default is the one choice nobody is shown. #[serde(default)] delete_foreign: bool, } async fn delete_session( State(manager): State>, UrlPath(id): UrlPath, Query(query): Query, ) -> Result { // Before the session goes, because only the session record says which file // on which machine this conversation is. let foreign = query .delete_foreign .then(|| manager.foreign_transcript(&id)) .flatten(); // And *deleted* before it too, so a machine that cannot be reached leaves // everything as it was rather than a deleted session and a transcript the // phone has already promised is gone. if let Some((setup, session)) = &foreign { let setup = setup_by_id(&manager, setup)?; let transport = crate::session::transport::Transport::for_setup(&setup); // A batch of one: the same call, so there is one description of what // deleting a foreign transcript means. crate::session::import::delete(&transport, std::slice::from_ref(session)) .await .map_err(bad_request)? .remove(session) .unwrap_or_else(|| Err(format!("nothing was reported about {session}"))) .map_err(|message| bad_request(anyhow::anyhow!("{message}")))?; tracing::info!("deleted Claude Code session {session} with ai-app session {id}"); } 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, } async fn message( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { // For the 404 a session that is not here has always answered with; the send // goes through the manager, which may have to start a process first. 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 UnqueueRequest { /// The id the `messageQueued` event carried, which is what the bubble on /// screen is drawn from. message_id: String, } /// Takes back a message the session has not read yet. /// /// The two failures are separate answers rather than one refusal, because /// they are different things to whoever tapped: `409` means the session has /// already been told, and `404` means nothing is waiting under that id -- a /// bubble something else has already resolved. The Claude driver can only /// ever give the first, since it writes a steer into the CLI on arrival. async fn unqueue( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { match lookup(&manager, &id)?.unqueue(&body.message_id) { Unqueued::Dropped => Ok(StatusCode::NO_CONTENT), Unqueued::AlreadySent => Err(ApiError::Conflict( "the session has already been given this message".to_string(), )), Unqueued::Unknown => Err(ApiError::NotFound( "this message is not waiting to be read".to_string(), )), } } #[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, } async fn answer( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { 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>, UrlPath(id): UrlPath, ) -> Result { lookup(&manager, &id)?.interrupt(); Ok(StatusCode::NO_CONTENT) } /// Ends the session's process. The session stays, and `start` brings it back. /// /// 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>, UrlPath(id): UrlPath, ) -> Result { 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. [`SessionManager::start_session`] refuses unless the session /// is known to have exited. async fn start( State(manager): State>, UrlPath(id): UrlPath, ) -> Result { 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. #[derive(Clone)] pub struct UsageState { monitor: Arc, manager: Arc, } /// 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, manager: Arc, ) -> Router { Router::new() .route("/usage", get(usage)) .with_state(UsageState { monitor, manager }) } async fn usage( State(state): State, ) -> Result>, 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>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { manager .rename_session(&id, &body.title) .map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct CwdRequest { cwd: PathBuf, } /// Moves a session to a different working directory. /// /// The directory is checked here rather than in the manager because checking /// it is an ssh round trip on a remote setup, and the manager is not async. /// /// Checked rather than trusted, and refused rather than corrected: a mistyped /// path that was accepted would leave a session recorded somewhere its process /// cannot start, and the failure would arrive later with nothing pointing at /// the typo. The spawn path corrects instead because it is resuming a /// directory the *machine* recorded, which can be gone through nobody's fault. /// /// It does not start a replacement process; see /// [`SessionManager::set_session_cwd`]. async fn set_cwd( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { let session = manager .sessions() .into_iter() .find(|session| session.id == id) .ok_or_else(|| ApiError::NotFound(format!("no session {id}")))?; // Absolute, because the alternative is relative to whatever the CLI is // launched from, which is not something the person typing it can see. The // same question the explorer asks of every path, asked in one place. let cwd = crate::files::check_path(&body.cwd.to_string_lossy()).map_err(bad_request)?; let setup = setup_by_id(&manager, &session.setup)?; let transport = crate::session::transport::Transport::for_setup(&setup); if !crate::session::import::directory_exists(&transport, &cwd).await { return Err(ApiError::BadRequest(format!( "{} has no directory {cwd}", setup.name ))); } // Stored in the short form, so the one path kept is the one the phone will // draw -- rather than storing `/home/bob/…` and abbreviating it again at // each place it is shown, which is two representations of one directory. // Only where the setup runs here; see `setups::shorten_home`. let stored = if setup.ssh.is_none() { crate::setups::shorten_home(&cwd) } else { cwd.clone() }; manager .set_session_cwd(&id, PathBuf::from(&stored)) .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>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { 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, } /// What new sessions start at. One field today; a struct rather than a bare /// value because "the defaults" is the thing a phone asks for, and the next /// one to move here -- the permission mode, which the spawn screen still /// hardcodes -- must not need a second route. #[derive(Serialize, Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct Defaults { #[serde(default, skip_serializing_if = "Option::is_none")] effort: Option, } async fn defaults(State(manager): State>) -> axum::Json { axum::Json(Defaults { effort: manager.default_effort(), }) } /// Sets what a new session's thinking level is. Applied when a session is /// spawned, so nothing already running changes underneath anybody. async fn set_defaults( State(manager): State>, axum::Json(body): axum::Json, ) -> Result { manager .set_default_effort(body.effort.as_deref()) .map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] struct EffortRequest { /// Absent or null is the CLI's own default, which is a choice somebody can /// make rather than only a state to start in. #[serde(default)] effort: Option, } /// Records how hard this session thinks, and stops the process so the next one /// is launched with it -- `--effort` has no control request behind it. See /// [`SessionManager::set_session_effort`]. async fn set_effort( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { manager .set_session_effort(&id, body.effort.as_deref()) .map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) } async fn set_permission_mode( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { 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>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { manager .set_session_notify(&id, body.notify) .map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) } #[derive(Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] struct AutoResumeRequest { auto_resume: bool, /// What to send when the limit lifts. Absent -- and empty, which is what a /// cleared field sends -- means this app's own default word, which is a /// choice a caller has to be able to make rather than only start in. #[serde(default)] message: Option, } /// Turns auto-resume on or off, and sets what it would say. /// /// One request for both, because they are one decision: switching it on /// without saying what to send is the ordinary case, and changing the words /// while it is off is how somebody sets it up before it is needed. async fn set_auto_resume( State(manager): State>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { manager .set_session_auto_resume(&id, body.auto_resume, body.message.as_deref()) .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, 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>, UrlPath(id): UrlPath, axum::Json(body): axum::Json, ) -> Result { 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. 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>, UrlPath(id): UrlPath, ) -> Result { lookup(&manager, &id)?; manager .run_command(&id, SessionCommand::Compact) .map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) } /// The most one attachment may be. Streamed to disk, so this bounds the /// session directory rather than memory; a day of `perfetto` is under a /// gigabyte, and this leaves room for a few of them. const ATTACHMENT_LIMIT: usize = 4 * 1024 * 1024 * 1024; /// Accepts one file (any multipart field) and stores it under the session; the /// returned id goes into a later `/message`'s attachmentIds. An image is later /// shown to the model, anything else is named to it by path. /// /// Written to disk as it arrives rather than collected first: a trace is bigger /// than this process should hold. Under a `.part` name until it is whole, so a /// tunnel that drops mid-upload leaves nothing a message could reference. /// /// A file for a session on another machine is copied there too, because the /// path the session is told has to exist where the session runs. The copy is /// part of the upload: if it fails, the upload fails and says so, rather than /// a message later naming a file that is not there. async fn upload_attachment( State(manager): State>, UrlPath(id): UrlPath, mut multipart: axum::extract::Multipart, ) -> Result, ApiError> { let session = lookup(&manager, &id)?; let mut 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("application/octet-stream") .to_string(); let file_name = field.file_name().map(str::to_string); let (name, path) = session .new_attachment(&content_type, file_name.as_deref()) .map_err(bad_request)?; let part = path.with_file_name(format!("{name}.part")); let received: Result<(), ApiError> = async { use tokio::io::AsyncWriteExt; let mut file = tokio::fs::File::create(&part) .await .with_context(|| format!("create {}", part.display()))?; while let Some(chunk) = field .chunk() .await .map_err(|err| ApiError::BadRequest(format!("upload read failed: {err}")))? { file.write_all(&chunk) .await .with_context(|| format!("write {}", part.display()))?; } file.flush() .await .with_context(|| format!("finish {}", part.display()))?; Ok(()) } .await; if let Err(err) = received { let _ = tokio::fs::remove_file(&part).await; return Err(err); } tokio::fs::rename(&part, &path) .await .with_context(|| format!("name {}", path.display())) .map_err(ApiError::Internal)?; // Images are not copied: they ride the message itself as base64. if crate::media::media_type_for(&name).is_none() && let Some((ssh, cwd)) = manager.remote_of(&id) { match ship_attachment(&ssh, cwd.as_deref(), &path, &name).await { Ok(remote) => { tokio::fs::write(remote_marker(&path), remote) .await .with_context(|| format!("record where {name} went")) .map_err(ApiError::Internal)?; } Err(err) => { let _ = tokio::fs::remove_file(&path).await; return Err(ApiError::BadRequest(format!( "{name} reached the server but couldn't be copied to {}: {err:#}", ssh.address ))); } } } Ok(axum::Json(serde_json::json!({ "id": name }))) } /// Copies `local` to the machine `ssh` names, into the configured attachments /// directory, else `cwd`, else the login home, and returns the absolute path it /// has there. /// /// One `ssh` invocation does the copy and answers the path: the file goes over /// stdin to `cat`, and `pwd -P` afterwards resolves whatever the directory was /// written as into the path the session will be told. `scp` would need a second /// round trip for that answer. async fn ship_attachment( ssh: &crate::config::SshConfig, cwd: Option<&Path>, local: &Path, name: &str, ) -> anyhow::Result { let dir = ssh.attachments_dir.as_deref().or(cwd); let mut script = String::new(); if let Some(dir) = dir { let dir = crate::ssh::quote_path(&dir.to_string_lossy()); // Created if missing: a configured directory may not exist yet, and a // session's own cwd already does, so this costs it nothing. script.push_str(&format!("mkdir -p {dir} && cd {dir} && ")); } script.push_str(&format!("cat > {} && pwd -P", crate::ssh::quote(name))); let source = std::fs::File::open(local).with_context(|| format!("open {}", local.display()))?; // Through the transport's own "with this on stdin", which the explorer's // write also uses -- one description of what that means rather than an ssh // invocation assembled here as well. let transport = crate::session::transport::Transport::Ssh { name: ssh.address.clone(), ssh: ssh.clone(), }; let launch = crate::session::transport::Launch::new("sh", vec!["-c".to_string(), script], None); let stdout = transport .capture_with_input(&launch, crate::session::transport::Input::File(source)) .await? .ok()?; let dir = String::from_utf8_lossy(&stdout).trim().to_string(); if dir.is_empty() { anyhow::bail!("the remote shell did not say where it put the file"); } Ok(format!("{dir}/{name}")) } /// Where the remote path of a shipped attachment is recorded, beside it. /// Read by `ClaudeDriver`'s `attachment_path`; removed with the session. fn remote_marker(local: &Path) -> std::path::PathBuf { let name = local.file_name().unwrap_or_default().to_string_lossy(); local.with_file_name(format!("{name}.remote")) } /// Serves a session's stored files -- both `files/` (images produced by tools) /// and `attachments/` (uploaded from the phone), by the id events and uploads /// reference. async fn serve_file( State(manager): State>, UrlPath((id, name)): UrlPath<(String, String)>, ) -> Result { // Ids are server-generated -- hex and an extension, or hex and a cleaned // file name; anything else (any path separator in particular) is refused, // not resolved. if !name .chars() .all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-' || 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. let bytes = std::fs::read(path) .with_context(|| format!("read {}", path.display())) .map_err(ApiError::Internal)?; // Every image this server writes has an extension it knows; the rest are // files attached by name, served as the bytes they are. let content_type = crate::media::media_type_for(&name).unwrap_or("application/octet-stream"); 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. 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, #[serde(default = "default_window")] limit: usize, /// Join each reply's streamed deltas into one event, so a page counts rows /// rather than tokens. The scroll-back pager asks for this; the /// anchor-restore path does not, because it counts events to reach a known /// seq. See `read_window`. #[serde(default)] coalesce: bool, /// Return nothing at or below this seq; the page stops here instead of at /// `limit`. The phone passes the end of what it already holds, so a page /// never overlaps it. Exclusive, like the SSE route's `after`. #[serde(default)] after: Option, } 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 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>, UrlPath(id): UrlPath, Query(query): Query, ) -> Result>, ApiError> { let session = lookup(&manager, &id)?; let events = crate::session::transcript::read_window( session.transcript_path(), query.before, query.after, query.limit, query.coalesce, ) .map_err(bad_request)?; // How far back a phone has paged, and what each page cost it, which is the // one question this route raises and nothing else can answer: the app asks // for events and draws rows, and the ratio between them is a property of // the conversation. `RUST_LOG=ai_server=debug`. tracing::debug!( session = %id, before = ?query.before, after = ?query.after, limit = query.limit, got = events.len(), oldest = ?events.first().map(|entry| entry.seq), "transcript page" ); Ok(axum::Json(events)) } async fn events( State(manager): State>, UrlPath(id): UrlPath, Query(query): Query, headers: HeaderMap, ) -> Result>>, 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 sends somebody to a session that may have been /// answered from another device since, and a notification that is wrong costs /// the reader the trip *and* teaches them to distrust the next one. What was /// missed is still on the session list, which answers "what is waiting" /// without claiming to be news. async fn notifications( State(manager): State>, ) -> Sse>> { let live = manager.subscribe_notifications(); let stream = BroadcastStream::new(live).filter_map(|item| { // A lagged subscriber has lost the oldest notifications, and the ones // it still gets are the recent ones -- the ones worth acting on. let notification = item.ok()?; Some(Ok(SseEvent::default().json_data(¬ification).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, tx: mpsc::Sender, ) { 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. async fn send_backlog(transcript: &Path, last: &mut u64, tx: &mpsc::Sender) -> bool { let cursor = *last; let entries = match catch_up(transcript, *last, CATCH_UP_LIMIT) { Ok(CatchUp::Continue(entries)) => { // The pair at debug, because "was this subscriber reset, and how // far behind was it" is a question about a phone that nothing else // here can answer: the app sees a window arrive and cannot tell how // far it had fallen. tracing::debug!(cursor, sent = entries.len(), "stream backlog: continue"); entries } Ok(CatchUp::Restart(entries)) => { tracing::debug!(cursor, sent = entries.len(), "stream backlog: reset"); 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, entry: &SeqEvent, ) -> Result<(), mpsc::error::SendError> { 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) -> 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, downloads: Vec, } async fn list_models( State(store): State>, ) -> Result, 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, ) -> Result>, 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>, Query(query): Query, ) -> Result>, 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>, axum::Json(body): axum::Json, ) -> Result, 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>, axum::Json(body): axum::Json, ) -> Result, ApiError> { let status = store.cancel(&body.key).map_err(bad_request)?; Ok(axum::Json(status)) } async fn delete_model( State(store): State>, axum::Json(body): axum::Json, ) -> Result { store.delete(&body.key).map_err(bad_request)?; Ok(StatusCode::NO_CONTENT) }