irisandClaude Opus 5
32b2a4871a
The composer's settings row is outlined bubbles opening round menus, and the message box is a TextFieldValue so anything put into it without being typed -- a draft, a share, a slash command -- leaves the cursor at the end. A tap that puts a text selection away no longer also collapses the card the text was drawn in: every open and close on the session screen goes through one guard that spends such a press on the selection. The usage bar and the usage dialog were two polls of one measurement and disagreed for up to a minute at a time; they are one feed now, and the countdown rounds up to the minute in the one place both read. A working directory typed as ~/repos/ai-app was four literal characters on the local transport and as an argument on both, so the existence check refused every home-relative path. It is checked by entering the directory now, expanded for a local spawn the way the remote shell expands it, and stored short so the phone draws what somebody would write. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1789 lines
70 KiB
Rust
1789 lines
70 KiB
Rust
//! 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}/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}/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
|
|
//! 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::{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<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))
|
|
// A batch at a time, never a session at a time -- see
|
|
// [`delete_importable`]. There is no `{session}` route to collide
|
|
// with, so all three of these are plain static segments.
|
|
.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),
|
|
)
|
|
.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}/notify", post(set_notify))
|
|
.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 (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),
|
|
/// 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") -- 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>,
|
|
/// Where attached files land on that machine; see `SshConfig`.
|
|
#[serde(default)]
|
|
attachments_dir: Option<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(),
|
|
// Not `tidy`: that expands `~` to *this* machine's home, and
|
|
// this path is on the other one. The remote shell expands it
|
|
// there (`ssh::quote_path`).
|
|
attachments_dir: self
|
|
.attachments_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<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<ImportableRow>>, 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.
|
|
//
|
|
// 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 be
|
|
// showing nothing at all where the thing it is waiting for used to be.
|
|
// A row with an operation on it stays until the operation settles.
|
|
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, out of range, or freshly opened never heard
|
|
// the events -- see `pending`. An operation is *not* filtered out
|
|
// above: a row being imported has to stay visible, marked, or the list
|
|
// would say the work never started.
|
|
let present: Vec<String> = found.iter().map(|row| row.id.clone()).collect();
|
|
manager.pending().prune(&id, &present);
|
|
let rows: Vec<ImportableRow> = 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 between "what the machine
|
|
/// said" and "what we are doing about it".
|
|
#[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<String>,
|
|
}
|
|
|
|
/// Removes Claude Code sessions from a machine.
|
|
///
|
|
/// The transcript *is* the session, so this ends any chance of resuming
|
|
/// those conversations -- including from an ai-app session already
|
|
/// importing one. 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 the whole reason this is a
|
|
/// POST with a body rather than a `DELETE` on each id. The phone used to
|
|
/// send one request per row, and a handover was then only as atomic as the
|
|
/// network was reliable: 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, so the answer to
|
|
/// "did my batch start" is one answer for the batch.
|
|
///
|
|
/// Registering is what has to be atomic; the work itself does not. The
|
|
/// batch runs as one command on the machine -- see
|
|
/// [`crate::session::import::delete`] for why it is not one per id -- but
|
|
/// each row still 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, and pretending otherwise would mean holding five
|
|
/// sessions hostage to the one that failed.
|
|
async fn delete_importable(
|
|
State(manager): State<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
axum::Json(body): axum::Json<DeleteBatch>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
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, so they all say so.
|
|
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<String>,
|
|
}
|
|
|
|
/// 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 -- which is the whole
|
|
/// point, since the screen it was started from may well be gone by then.
|
|
///
|
|
/// A list for the same reason [`delete_importable`] takes one: the batch is
|
|
/// handed over in a single request, so it cannot half-arrive.
|
|
async fn start_import(
|
|
State(manager): State<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
axum::Json(body): axum::Json<ImportRequest>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
// Checked before accepting, so an unknown machine is still an error the
|
|
// caller sees rather than a failure 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(),
|
|
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<String>,
|
|
provider: String,
|
|
#[serde(default)]
|
|
model: Option<String>,
|
|
#[serde(default)]
|
|
permission_mode: Option<String>,
|
|
}
|
|
|
|
/// 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 -- the row says what is happening to
|
|
/// it, whoever is looking and whenever they look.
|
|
fn in_background<F>(
|
|
manager: &Arc<SessionManager>,
|
|
setup: String,
|
|
session: String,
|
|
operation: Operation,
|
|
work: F,
|
|
) where
|
|
F: std::future::Future<Output = anyhow::Result<()>> + 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 -- a phone watching one machine's
|
|
/// import list has no use for another's.
|
|
async fn importable_events(
|
|
State(manager): State<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
) -> Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>> {
|
|
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<Arc<SessionManager>>,
|
|
axum::Json(body): axum::Json<SpawnRequest>,
|
|
) -> Result<axum::Json<SessionInfo>, 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 -- see [`start_import`]. 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<SessionManager>, body: SpawnRequest) -> Result<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 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 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(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 behind 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<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
Query(query): Query<DeleteSessionQuery>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
// 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. The phone can
|
|
// then retry, or turn the toggle off.
|
|
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. Its outcome is this
|
|
// request's outcome, since there is only the one row.
|
|
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<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 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 the message is on its way into the conversation,
|
|
/// and `404` means nothing is waiting under that id -- a bubble on screen
|
|
/// that something else has already resolved. See [`Driver::unqueue`]; the
|
|
/// Claude driver can only ever give the first, since it writes a steer into
|
|
/// the CLI the moment it arrives.
|
|
async fn unqueue(
|
|
State(manager): State<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
axum::Json(body): axum::Json<UnqueueRequest>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
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<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(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 -- the same division `POST /sessions` already makes for the
|
|
/// directory an import was recorded in.
|
|
///
|
|
/// 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, as a
|
|
/// session that would not come back, 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; a path
|
|
/// somebody has just typed is different.
|
|
///
|
|
/// Note what this does not do: it does not start a replacement process.
|
|
/// See [`SessionManager::set_session_cwd`].
|
|
async fn set_cwd(
|
|
State(manager): State<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
axum::Json(body): axum::Json<CwdRequest>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
let session = manager
|
|
.sessions()
|
|
.into_iter()
|
|
.find(|session| session.id == id)
|
|
.ok_or_else(|| ApiError::NotFound(format!("no session {id}")))?;
|
|
let cwd = body.cwd.to_string_lossy().trim().to_string();
|
|
if cwd.is_empty() {
|
|
return Err(ApiError::BadRequest(
|
|
"a working directory is a path, and this one is empty".to_string(),
|
|
));
|
|
}
|
|
// Absolute, because the alternative is relative to whatever the CLI is
|
|
// launched from, which is not something the person typing it can see.
|
|
if !cwd.starts_with('/') && !cwd.starts_with('~') {
|
|
return Err(ApiError::BadRequest(format!(
|
|
"{cwd} is not an absolute path, so where it would be depends on where the \
|
|
session happens to start"
|
|
)));
|
|
}
|
|
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 that is 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 and a second rule to keep in step.
|
|
// 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<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)
|
|
}
|
|
|
|
/// 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 -- see `ClaudeDriver::send_user_message`.
|
|
///
|
|
/// Written to disk as it arrives rather than collected first: a trace is
|
|
/// bigger than this process should hold, and the phone streams it for the
|
|
/// same reason. 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<Arc<SessionManager>>,
|
|
UrlPath(id): UrlPath<String>,
|
|
mut multipart: axum::extract::Multipart,
|
|
) -> Result<axum::Json<serde_json::Value>, 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 -- a `~`, a relative name, a symlink -- into
|
|
/// the path the session will be told, which is the one a CLI's file tools
|
|
/// take. `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<String> {
|
|
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()))?;
|
|
let mut command = tokio::process::Command::from(crate::ssh::command(
|
|
Some(ssh),
|
|
"sh",
|
|
&["-c".to_string(), script],
|
|
None,
|
|
));
|
|
command
|
|
.stdin(source)
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped());
|
|
let output = command.output().await.context("run ssh")?;
|
|
if !output.status.success() {
|
|
anyhow::bail!("{}", String::from_utf8_lossy(&output.stderr).trim());
|
|
}
|
|
let dir = String::from_utf8_lossy(&output.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<Arc<SessionManager>>,
|
|
UrlPath((id, name)): UrlPath<(String, String)>,
|
|
) -> Result<Response, ApiError> {
|
|
// Ids are server-generated -- hex and an extension, or hex and a
|
|
// cleaned file name (`safe_file_name`); anything else (and 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 -- 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)?;
|
|
// 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 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,
|
|
/// 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,
|
|
}
|
|
|
|
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,
|
|
query.coalesce,
|
|
)
|
|
.map_err(bad_request)?;
|
|
// How far back a phone has paged, and how much each page cost it to get
|
|
// there, 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,
|
|
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<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(¬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<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)
|
|
}
|