Files
codecraft/server/src/api.rs
T
Richard ZinkandClaude Opus 5.5 e8e455b779 Store chat images separately; JPEG for large PNGs; gzip/brotli responses
Images of messages (pasted images, screenshots, generated images) were base64
inside the message JSON, so opening a chat downloaded all of them at once
(66 MB for a chat with 120 screenshots). They now live in a `blobs` table,
addressed by SHA-256; messages keep a reference (blob, size). The UI loads
them from /api/blobs/<hash> lazily with an immutable cache header, the runner
fills the bytes in before model requests, JSON exports stay self-contained.

- Store::add_message moves inline images out; large opaque PNGs become JPEG
  (q85) when smaller. Older chats are converted once at start-up in the
  background, unused images are removed (also after deleting a chat) and the
  database is vacuumed when that frees 32 MB or more.
- Responses are compressed with gzip/brotli (not images, SSE streams or
  binary downloads).

Example (copy of production): chat load 66 MB -> 1.5 MB (370 KB brotli),
database 90 MB -> 27 MB.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-09 20:21:24 +01:00

1900 lines
80 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! HTTP + WebSocket API. Every route except the public auth endpoints requires a
//! signed-in user; admin-only routes check the role.
use crate::app::{Access, App, PermissionReply};
use crate::auth::{self, require_admin};
use crate::config::{self, ConfigPatch};
use crate::prompts;
use crate::runner;
use crate::store::User;
use crate::types::{Event, ImageData, Part, Session, new_id, now_ms};
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::middleware;
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post, put};
use axum::{Extension, Json, Router};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use std::path::PathBuf;
use std::sync::Arc;
pub type AppState = Arc<App>;
pub struct ApiError(pub StatusCode, pub String);
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
(self.0, Json(json!({"error": self.1}))).into_response()
}
}
impl<E: Into<anyhow::Error>> From<E> for ApiError {
fn from(e: E) -> Self {
ApiError(StatusCode::BAD_REQUEST, format!("{:#}", e.into()))
}
}
pub type ApiResult<T = Json<Value>> = Result<T, ApiError>;
fn not_found(what: &str) -> ApiError {
ApiError(StatusCode::NOT_FOUND, format!("{what} not found"))
}
fn forbidden() -> ApiError {
ApiError(StatusCode::FORBIDDEN, "no access to this session".into())
}
pub fn router(app: AppState) -> Router {
let api = Router::new()
// public
.route("/info", get(auth::info))
.route("/auth/login", post(auth::login))
.route("/auth/cli-login", post(auth::cli_login))
.route("/auth/logout", post(auth::logout))
.route("/auth/setup", post(auth::setup))
.route("/auth/oidc/start", get(auth::oidc_start))
.route("/auth/oidc/callback", get(auth::oidc_callback))
// signed in
.route("/auth/password", post(auth::change_password))
.route("/users", get(auth::list_users).post(auth::create_user))
.route("/users/directory", get(auth::directory))
.route("/users/{id}", axum::routing::patch(auth::update_user).delete(auth::delete_user))
.route("/folders", get(list_folders).post(create_folder))
.route("/folders/{id}", delete(delete_folder))
.route("/folders/{id}/access", put(set_folder_access))
.route("/config", get(get_config).patch(patch_config))
.route("/config/raw", get(get_config_raw).put(put_config_raw))
.route("/providers/reload", post(reload_providers))
.route("/models", get(models))
.route("/agents", get(agents))
.route("/agents/tools", get(agent_tools))
.route("/agents/{name}", delete(delete_agent))
.route("/commands", get(commands))
.route("/sessions", get(list_sessions).post(create_session))
.route("/sessions/search", get(search_sessions))
.route("/sessions/{id}/fork", post(fork_session))
.route("/sessions/{id}/title", post(suggest_title))
.route("/sessions/{id}/regenerate", post(regenerate))
.route("/sessions/{id}/worktree", get(worktree_status).post(worktree_action))
.route("/sessions/{id}/tabs", get(session_tabs).put(put_file_tabs))
.route("/sessions/{id}/export", get(export_session))
.route("/blobs/{hash}", get(get_blob))
.route("/sessions/{id}", get(get_session).patch(update_session).delete(delete_session))
.route("/sessions/{id}/prompt", post(prompt))
.route("/sessions/{id}/abort", post(abort))
.route("/sessions/{id}/compact", post(compact))
.route("/sessions/{id}/rewind", post(rewind))
.route("/sessions/{id}/changes", get(changed_files))
.route("/sessions/{id}/shells", get(list_shells))
.route("/sessions/{id}/shells/{shell}/kill", post(kill_shell))
.route("/sessions/{id}/stats", get(session_stats))
.route("/sessions/{id}/shares", get(get_shares).put(put_shares))
.route("/sessions/{id}/queue", get(get_queue))
.route("/sessions/{id}/queue/flush", post(flush_queue))
.route("/sessions/{id}/queue/{item}", delete(delete_queue_item))
.route("/permissions/{id}", post(reply_permission))
.route("/questions/{id}", post(answer_question))
.route("/pending", get(pending))
.route("/mcp", get(mcp_status))
.route("/mcp/reload", post(mcp_reload))
.route("/fs/roots", get(list_roots))
.route("/files/list", get(crate::files::list))
.route("/files/read", get(crate::files::read))
.route("/files/raw", get(crate::files::raw))
.route("/files/stat", get(crate::files::stat))
.route("/files/write", put(crate::files::write))
.route("/files/create", post(crate::files::create))
.route("/files/move", post(crate::files::move_or_copy))
.route("/files/delete", delete(crate::files::remove))
.route("/files/upload", put(crate::files::upload))
.route("/files/search", get(crate::files::search))
.route("/files/preview/pdf-info", get(crate::files::pdf_info))
.route("/files/preview/pdf-page", get(crate::files::pdf_page))
.route("/files/preview/doc", get(crate::files::doc_preview))
.route("/files/preview/module", get(crate::files::module_audio))
.route("/git", post(crate::git::handle))
.route("/fs/dirs", get(list_dirs))
.route("/stats", get(stats))
.route("/budget", get(budget))
.route("/audit", get(audit_log))
.route("/backup", get(backup))
.route("/backups", get(list_backups).post(run_backup))
.route("/backups/{name}", get(download_backup))
.route("/backups/{name}/restore", post(restore_stored_backup))
.route("/backups/restore", post(restore_uploaded_backup).layer(axum::extract::DefaultBodyLimit::max(4 * 1024 * 1024 * 1024)))
.route("/health", get(health))
.route("/tools", get(crate::reverse::status))
.route("/tools/ghidra/install", post(crate::reverse::install_ghidra))
.route("/providers/{name}/prices", get(provider_prices))
.route("/providers/status", get(provider_status).post(recheck_providers))
.route("/providers/test", post(test_provider))
.route("/status", get(status_overview))
.route("/images", get(list_images).post(create_images))
.route("/images/config", get(images_config))
.route("/images/{id}", delete(delete_image))
.route("/images/{id}/file", get(image_file))
.route("/account/git", get(get_git_settings).put(put_git_settings))
.route("/account/prefs", get(crate::cli_api::get_prefs).put(crate::cli_api::put_prefs))
.route("/account/memory", get(crate::cli_api::get_memory).put(crate::cli_api::put_memory))
.route("/account/modes", get(get_personal_modes).put(put_personal_modes))
.route("/account/telegram", get(get_telegram).put(put_telegram).delete(delete_telegram))
.route("/templates", get(crate::cli_api::get_templates))
.route("/templates/{scope}", put(crate::cli_api::put_templates))
.route("/cli/config", get(crate::cli_api::config))
.route("/cli/skill/{name}", get(crate::cli_api::skill))
.route("/cli/tool", post(crate::cli_api::tool))
.route("/llm/stream", post(crate::cli_api::llm_stream))
.route("/account/tokens", get(auth::list_api_tokens).post(auth::create_api_token))
.route("/account/tokens/{id}", delete(auth::delete_api_token))
.route("/jobs", get(list_jobs).post(register_job))
.route("/jobs/create", post(create_job))
.route("/jobs/{id}", axum::routing::patch(update_job).delete(delete_job))
.route("/jobs/{id}/run", post(run_job))
.route("/jobs/{id}/webhook", post(job_webhook_settings))
.route("/terminals", get(list_terminals).post(open_terminal))
.route("/terminals/{id}", delete(close_terminal))
.route("/terminals/{id}/ws", get(terminal_ws))
.route("/push/key", get(push_key))
.route("/push/subscribe", post(push_subscribe).delete(push_unsubscribe))
.route("/push/test", post(push_test))
.route("/push/digest", get(get_digest).put(put_digest).post(send_digest))
.route("/ws", get(ws))
.layer(middleware::from_fn_with_state(app.clone(), auth::require_user))
// Prompts may carry images (base64); the default limit is 2 MB.
.layer(axum::extract::DefaultBodyLimit::max(50 * 1024 * 1024));
let web_dir = std::env::var("CODECRAFT_WEB").map(PathBuf::from).unwrap_or_else(|_| {
["../web/dist", "./web/dist", "/app/web"]
.iter()
.map(PathBuf::from)
.find(|p| p.join("index.html").exists())
.unwrap_or_else(|| PathBuf::from("./web/dist"))
});
let index = web_dir.join("index.html");
let static_files = tower_http::services::ServeDir::new(&web_dir).fallback(tower_http::services::ServeFile::new(index));
Router::new()
.nest("/api", api)
.route("/hooks/{id}", post(incoming_webhook))
.route("/metrics", get(crate::metrics::metrics))
.route("/cli/install.sh", get(cli_install_script))
.route("/cli/download/{target}", get(cli_download))
.fallback_service(static_files)
.layer(compression())
.with_state(app)
}
/// gzip/brotli for JSON, HTML, JS and CSS. Images and SSE streams are left alone (default
/// predicate), and so are binary downloads (already compressed or large).
fn compression() -> tower_http::compression::CompressionLayer<impl tower_http::compression::Predicate + Clone> {
use tower_http::compression::predicate::{DefaultPredicate, NotForContentType, Predicate};
let predicate = DefaultPredicate::new()
.and(NotForContentType::const_new("application/octet-stream"))
.and(NotForContentType::const_new("application/zip"))
.and(NotForContentType::const_new("application/gzip"));
tower_http::compression::CompressionLayer::new().compress_when(predicate)
}
fn cli_dir() -> PathBuf {
std::env::var("CODECRAFT_CLI_DIR").map(PathBuf::from).unwrap_or_else(|_| PathBuf::from("/app/cli"))
}
/// Whether this server ships prebuilt terminal client binaries (the release image does not).
pub(crate) fn cli_builds_available() -> bool {
std::fs::read_dir(cli_dir()).map(|mut d| d.any(|e| e.is_ok_and(|e| e.file_name().to_string_lossy().starts_with("codecraft-")))).unwrap_or(false)
}
const CLI_SOURCE_HINT: &str = "build it from source: cargo install --locked --git https://git.familie-zink.org/zink/codecraft codecraft-cli";
/// The server's public base URL (config, else from the request).
fn public_base(app: &App, headers: &axum::http::HeaderMap) -> String {
let cfg = app.config();
if !cfg.auth.public_url.is_empty() {
return cfg.auth.public_url.trim_end_matches('/').to_string();
}
let host = headers.get("x-forwarded-host").or(headers.get(axum::http::header::HOST)).and_then(|v| v.to_str().ok()).unwrap_or("localhost:8787");
let proto = headers.get("x-forwarded-proto").and_then(|v| v.to_str().ok()).unwrap_or("http");
format!("{proto}://{host}")
}
/// `curl -fsSL <server>/cli/install.sh | sh` installs the terminal client.
async fn cli_install_script(State(app): State<AppState>, headers: axum::http::HeaderMap) -> Response {
let base = public_base(&app, &headers);
let script = include_str!("../resources-cli/install.sh").replace("__CODECRAFT_URL__", &base);
([(axum::http::header::CONTENT_TYPE, "text/x-shellscript; charset=utf-8")], script).into_response()
}
/// Static CLI binaries, e.g. /cli/download/linux-x86_64.
async fn cli_download(Path(target): Path<String>) -> Response {
if !target.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
return (StatusCode::BAD_REQUEST, "invalid target").into_response();
}
let path = cli_dir().join(format!("codecraft-{target}"));
match tokio::fs::read(&path).await {
Ok(bytes) => (
[
(axum::http::header::CONTENT_TYPE, "application/octet-stream".to_string()),
(axum::http::header::CONTENT_DISPOSITION, "attachment; filename=\"codecraft\"".to_string()),
],
bytes,
)
.into_response(),
Err(_) => (StatusCode::NOT_FOUND, format!("no prebuilt codecraft for {target} on this server – {CLI_SOURCE_HINT}\n")).into_response(),
}
}
/// Load a session the user may at least read.
fn readable(app: &App, user: &User, id: &str) -> Result<Session, ApiError> {
let s = app.store.get_session(id)?.ok_or_else(|| not_found("session"))?;
if app.access(user, id) == Access::None {
return Err(forbidden());
}
Ok(s)
}
/// Load a session the user may control (owner or admin).
fn writable(app: &App, user: &User, id: &str) -> Result<Session, ApiError> {
let s = readable(app, user, id)?;
if app.access(user, id) != Access::Full {
return Err(ApiError(StatusCode::FORBIDDEN, "this session is shared with you read-only".into()));
}
Ok(s)
}
// --- config (admin) -----------------------------------------------------------
async fn get_config(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let cfg = app.config();
let default_model = if cfg.server.default_model.is_empty() { app.registry().first_model() } else { Some(cfg.server.default_model.clone()) };
if !user.is_admin() {
// Users only need defaults, not the configuration.
return Ok(Json(json!({"default_model": default_model, "small_model": cfg.server.small_model, "currency": cfg.server.currency})));
}
Ok(Json(json!({
"config": config::redacted_json(&cfg),
"workspace": app.workspace(),
"default_model": default_model,
"small_model": cfg.server.small_model,
"currency": cfg.server.currency,
})))
}
#[derive(Deserialize)]
struct PatchBody {
patches: Vec<ConfigPatch>,
}
async fn patch_config(Extension(user): Extension<User>, State(app): State<AppState>, Json(body): Json<PatchBody>) -> ApiResult {
require_admin(&user)?;
// Masked secrets coming back from the UI must not overwrite real values.
let patches: Vec<ConfigPatch> = body.patches.into_iter().filter(|p| p.value.as_str() != Some("••••••••")).collect();
let paths: Vec<String> = patches.iter().map(|p| p.path.join(".")).collect();
app.patch_config(&patches).await?;
app.audit(Some(&user), "config.patch", paths.join(", "), "");
Ok(Json(json!({"ok": true})))
}
async fn get_config_raw(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
Ok(Json(json!({"text": app.config_text()?})))
}
#[derive(Deserialize)]
struct RawBody {
text: String,
}
async fn put_config_raw(Extension(user): Extension<User>, State(app): State<AppState>, Json(body): Json<RawBody>) -> ApiResult {
require_admin(&user)?;
app.save_config(&body.text).await?;
app.audit(Some(&user), "config.raw", format!("{} bytes", body.text.len()), "");
Ok(Json(json!({"ok": true})))
}
async fn reload_providers(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
app.reload_providers().await;
Ok(Json(app.registry().models_json()))
}
async fn models(State(app): State<AppState>) -> Json<Value> {
Json(app.registry().models_json())
}
#[derive(Deserialize)]
struct CwdQuery {
cwd: Option<String>,
}
/// Root and cwd for listing agents/commands; falls back to the user's first root.
fn scope(app: &App, user: &User, cwd: Option<String>) -> (PathBuf, PathBuf) {
let owner = Some(user.id.as_str());
let roots = app.user_roots(owner);
let cwd = cwd.map(PathBuf::from).filter(|c| roots.iter().any(|r| c.starts_with(r))).unwrap_or_else(|| roots[0].clone());
(app.root_for(owner, &cwd), cwd)
}
async fn agents(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<CwdQuery>) -> Json<Value> {
let (root, cwd) = scope(&app, &user, q.cwd);
Json(json!(prompts::agents_for(&app, &root, &cwd, Some(&user.id))))
}
/// Telegram bot of the user: bot name, pairing state, errors (the token stays hidden).
async fn get_telegram(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let s = crate::telegram::load(&app, &user.id);
let (running, error) = crate::telegram::status(&user.id);
Ok(Json(json!({
"configured": !s.token.is_empty(),
"bot": s.bot,
"paired": s.chat_id.is_some(),
"pair_code": if s.chat_id.is_none() { s.pair_code } else { String::new() },
"running": running,
"error": error,
})))
}
#[derive(Deserialize)]
struct TelegramBody {
token: String,
}
/// Connect a bot: check the token, start polling and create a pairing code.
async fn put_telegram(Extension(user): Extension<User>, State(app): State<AppState>, Json(b): Json<TelegramBody>) -> ApiResult {
let token = crate::telegram::extract_token(&b.token)
.ok_or_else(|| ApiError(StatusCode::BAD_REQUEST, "that is not a bot token – it looks like 123456789:AAH… (paste only the token from @BotFather)".into()))?;
let bot = crate::telegram::check_token(&token).await.map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("token rejected by Telegram: {e:#}")))?;
let s = crate::telegram::Settings { token, bot, pair_code: crate::telegram::new_pair_code(), ..Default::default() };
app.store.set_user_setting(&user.id, "telegram", &s)?;
crate::telegram::restart(&app, &user.id);
app.audit(Some(&user), "telegram.connect", &s.bot, "");
get_telegram(Extension(user), State(app)).await
}
async fn delete_telegram(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
crate::telegram::stop(&user.id);
app.store.set_user_setting(&user.id, "telegram", &crate::telegram::Settings::default())?;
app.audit(Some(&user), "telegram.disconnect", "", "");
Ok(Json(json!({"ok": true})))
}
/// The user's personal modes (name -> settings).
async fn get_personal_modes(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
Ok(Json(json!(prompts::personal_modes(&app, &user.id))))
}
/// Replace the user's personal modes. Names of built-in, admin or project modes are
/// taken; tools and read-only work like in admin modes (the user's sandbox still applies).
async fn put_personal_modes(Extension(user): Extension<User>, State(app): State<AppState>, Json(modes): Json<std::collections::BTreeMap<String, crate::config::AgentConfig>>) -> ApiResult {
if modes.len() > 30 {
return Err(ApiError(StatusCode::BAD_REQUEST, "too many personal modes (max 30)".into()));
}
let root = app.user_roots(Some(&user.id)).into_iter().next().unwrap_or_else(|| app.workspace());
let taken: Vec<String> = prompts::agents(&app, &root, &root).into_iter().map(|a| a.name).collect();
let mut clean = std::collections::BTreeMap::new();
for (name, mut m) in modes {
if name.is_empty() || name.len() > 40 || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
return Err(ApiError(StatusCode::BAD_REQUEST, format!("invalid mode name '{name}' (letters, digits, - and _)")));
}
if taken.contains(&name) {
return Err(ApiError(StatusCode::BAD_REQUEST, format!("the name '{name}' is already used by another mode")));
}
if m.prompt.as_deref().is_some_and(|p| p.len() > 20_000) {
return Err(ApiError(StatusCode::BAD_REQUEST, "prompt too long (max 20,000 characters)".into()));
}
m.mode = Some("primary".into());
m.max_parallel = None;
clean.insert(name, m);
}
app.store.set_user_setting(&user.id, "modes", &clean)?;
Ok(Json(json!(clean)))
}
/// Built-in tools for the mode editors (name, read-only, first line of the description).
async fn agent_tools() -> ApiResult {
let list: Vec<Value> = crate::tools::builtin_tools()
.iter()
.map(|t| json!({"name": t.name(), "read_only": t.read_only(), "description": t.description().split(['.', '\n']).next().unwrap_or("").trim()}))
.collect();
Ok(Json(json!(list)))
}
/// Delete a user-defined agent: its config entry and/or its Markdown file.
async fn delete_agent(Extension(user): Extension<User>, State(app): State<AppState>, Path(name): Path<String>, Query(q): Query<CwdQuery>) -> ApiResult {
let (root, cwd) = scope(&app, &user, q.cwd);
let agent = prompts::agents(&app, &root, &cwd).into_iter().find(|a| a.name == name).ok_or_else(|| not_found("agent"))?;
let mut removed = Vec::new();
if app.config().agents.contains_key(&name) {
require_admin(&user)?;
let patch = ConfigPatch { path: vec!["agents".to_string(), name.clone()], value: Value::Null };
app.patch_config(&[patch]).await.map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
removed.push("config".to_string());
}
if let Some(path) = &agent.path {
let p = std::path::PathBuf::from(path);
// Global agent files are admin-managed; project files must be in the user's folders.
let in_data = p.starts_with(app.data_dir());
let allowed = if in_data { user.is_admin() } else { crate::files::checked_path(&app, &user, path, true).is_ok() };
if !allowed || p.starts_with(prompts::builtin_resources()) {
return Err(ApiError(StatusCode::FORBIDDEN, "this agent file cannot be deleted by you".into()));
}
std::fs::remove_file(&p)?;
removed.push(path.clone());
}
if removed.is_empty() {
return Err(ApiError(StatusCode::BAD_REQUEST, "built-in modes cannot be deleted".into()));
}
app.audit(Some(&user), "agent.delete", format!("{name}: {}", removed.join(", ")), "");
Ok(Json(json!({"ok": true, "removed": removed})))
}
async fn commands(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<CwdQuery>) -> Json<Value> {
let (root, cwd) = scope(&app, &user, q.cwd);
Json(json!({
"commands": prompts::commands(&app, &root, &cwd),
"skills": prompts::skills(&app, &root, &cwd),
}))
}
// --- shared folders (admin) ---------------------------------------------------
async fn list_folders(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
Ok(Json(json!({"workspace": app.workspace(), "folders": app.store.folders()?})))
}
#[derive(Deserialize)]
struct FolderBody {
/// Path relative to the workspace (or absolute inside it). Created if missing.
path: String,
#[serde(default)]
users: Vec<String>,
}
async fn create_folder(Extension(user): Extension<User>, State(app): State<AppState>, Json(body): Json<FolderBody>) -> ApiResult {
require_admin(&user)?;
let ws = app.workspace();
let rel = body.path.trim().trim_start_matches(ws.to_str().unwrap_or("")).trim_matches('/');
if rel.is_empty() || rel.split('/').any(|c| c.is_empty() || c == "." || c == ".." || c.starts_with('.')) {
return Err(ApiError(StatusCode::BAD_REQUEST, "invalid folder path".into()));
}
let path = ws.join(rel);
std::fs::create_dir_all(&path)?;
let path = std::fs::canonicalize(&path)?;
if !path.starts_with(&ws) || path == ws {
return Err(ApiError(StatusCode::BAD_REQUEST, "folder must be inside the workspace".into()));
}
let id = app.store.add_folder(&path.display().to_string())?;
app.audit(Some(&user), "folder.create", path.display().to_string(), "");
app.store.set_folder_access(&id, &body.users)?;
Ok(Json(json!({"id": id, "path": path})))
}
async fn delete_folder(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
require_admin(&user)?;
// Only the share is removed; the directory and its files stay.
app.store.delete_folder(&id)?;
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct AccessBody {
users: Vec<String>,
}
async fn set_folder_access(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<AccessBody>) -> ApiResult {
require_admin(&user)?;
app.store.set_folder_access(&id, &body.users)?;
app.audit(Some(&user), "folder.access", format!("{id}: {} user(s)", body.users.len()), "");
Ok(Json(json!({"ok": true})))
}
// --- sessions ---------------------------------------------------------------
async fn list_sessions(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let filter = if user.is_admin() { None } else { Some(user.id.as_str()) };
let list: Vec<Value> = app
.store
.list_sessions(filter)?
.into_iter()
.map(|s| {
let busy = app.is_busy(&s.id);
let shared = s.owner_id.as_deref() != Some(user.id.as_str());
let mut v = json!(s);
v["busy"] = json!(busy);
v["shared"] = json!(shared);
v
})
.collect();
Ok(Json(json!(list)))
}
#[derive(Deserialize)]
struct CreateSession {
cwd: Option<String>,
model: Option<String>,
agent: Option<String>,
permission_mode: Option<String>,
clean_room: Option<bool>,
}
/// Resolve a directory and check it lies inside one of the user's roots.
fn validate_cwd(app: &App, user: &User, cwd: Option<String>) -> Result<PathBuf, ApiError> {
let roots = app.user_roots(Some(&user.id));
let p = match cwd.filter(|c| !c.trim().is_empty()) {
Some(c) if std::path::Path::new(&c).is_absolute() => PathBuf::from(c),
Some(c) => roots[0].join(c),
None => roots[0].clone(),
};
let p = std::fs::canonicalize(&p).map_err(|_| ApiError(StatusCode::BAD_REQUEST, format!("directory does not exist: {}", p.display())))?;
if !p.is_dir() || !roots.iter().any(|r| p.starts_with(r)) {
return Err(ApiError(StatusCode::FORBIDDEN, "directory is outside your folders".into()));
}
Ok(p)
}
async fn create_session(Extension(user): Extension<User>, State(app): State<AppState>, Json(body): Json<CreateSession>) -> ApiResult {
let cwd = validate_cwd(&app, &user, body.cwd)?;
let cfg = app.config();
let model = body
.model
.filter(|m| !m.is_empty())
.or_else(|| Some(cfg.server.default_model.clone()).filter(|m| !m.is_empty()))
.or_else(|| app.registry().first_model())
.ok_or_else(|| ApiError(StatusCode::BAD_REQUEST, "no model configured – ask an admin to add a provider".into()))?;
let mode = body.permission_mode.unwrap_or_else(|| "auto".into());
if !crate::permission::MODES.contains(&mode.as_str()) {
return Err(ApiError(StatusCode::BAD_REQUEST, "invalid permission mode".into()));
}
let now = now_ms();
let s = Session {
id: new_id(),
title: String::new(),
cwd: cwd.display().to_string(),
model,
agent: body.agent.unwrap_or_else(|| "build".into()),
permission_mode: mode,
parent_id: None,
owner_id: Some(user.id.clone()),
created_at: now,
updated_at: now,
todos: Value::Null,
last_usage: None,
clean_room: body.clean_room,
phase: None,
worktree: None,
pinned: false,
archived: false,
folder: None,
instructions: String::new(),
};
let s = app.save_session(s)?;
Ok(Json(json!(s)))
}
#[derive(Deserialize)]
struct SearchQ {
q: String,
}
async fn search_sessions(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<SearchQ>) -> ApiResult {
if q.q.trim().len() < 2 {
return Ok(Json(json!([])));
}
let filter = if user.is_admin() { None } else { Some(user.id.as_str()) };
let hits: Vec<Value> = app
.store
.search_sessions(filter, q.q.trim())?
.into_iter()
.map(|(s, snippet)| json!({"id": s.id, "title": s.title, "cwd": s.cwd, "agent": s.agent, "updated_at": s.updated_at, "snippet": snippet}))
.collect();
Ok(Json(json!(hits)))
}
#[derive(Deserialize)]
struct RegenerateBody {
/// An assistant message of the turn to redo.
message_id: String,
model: Option<String>,
effort: Option<String>,
}
/// Redo a turn: drop everything after the user's prompt, restore files changed
/// since, and run again (optionally with another model / effort).
async fn regenerate(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(b): Json<RegenerateBody>) -> ApiResult {
let mut s = writable(&app, &user, &id)?;
if app.is_busy(&id) {
return Err(ApiError(StatusCode::CONFLICT, "session is busy".into()));
}
let msgs = app.store.messages(&id)?;
let idx = msgs.iter().position(|m| m.id == b.message_id).ok_or_else(|| not_found("message"))?;
// The prompt of this turn: the last user message with real content before it.
let prompt_idx = msgs[..idx]
.iter()
.rposition(|m| matches!(m.role, crate::types::Role::User) && m.parts.iter().any(|p| matches!(p, Part::Text { .. } | Part::Image { .. })))
.ok_or_else(|| ApiError(StatusCode::BAD_REQUEST, "no prompt before this message".into()))?;
let first_dropped = msgs.get(prompt_idx + 1).ok_or_else(|| ApiError(StatusCode::BAD_REQUEST, "nothing to regenerate".into()))?;
if let Some(m) = b.model.filter(|m| !m.is_empty()) {
app.registry().resolve(&m).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
s.model = m;
}
// Files the dropped steps changed go back to their state before.
let since = first_dropped.created_at;
let roots = app.user_roots(s.owner_id.as_deref());
for (path, content) in app.store.checkpoints_since(&id, since)? {
let p = PathBuf::from(&path);
if !roots.iter().any(|r| p.starts_with(r)) {
continue;
}
match content {
Some(bytes) => {
let _ = std::fs::write(&p, bytes);
}
None => {
let _ = std::fs::remove_file(&p);
}
}
}
app.store.delete_checkpoints_since(&id, since)?;
app.store.truncate_from(&id, &first_dropped.id)?;
s.last_usage = None;
app.save_session(s)?;
set_turn_effort(&app, &id, b.effort.as_deref())?;
runner::submit(app.clone(), id.clone(), vec![], "user").map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")))?;
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct WorktreeBody {
action: String,
}
async fn worktree_status(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let mut s = readable(&app, &user, &id)?;
if s.worktree.is_none() {
return Ok(Json(json!({"worktree": null})));
}
Ok(Json(crate::git::worktree(&app, &user, &mut s, "status").await?))
}
async fn worktree_action(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(b): Json<WorktreeBody>) -> ApiResult {
let mut s = writable(&app, &user, &id)?;
if app.is_busy(&id) && b.action != "status" {
return Err(ApiError(StatusCode::CONFLICT, "the session is busy – wait until the agent is done".into()));
}
let res = crate::git::worktree(&app, &user, &mut s, &b.action).await?;
if b.action == "remove" {
s.worktree = None;
}
if b.action != "status" {
app.save_session(s)?;
app.audit(Some(&user), &format!("worktree.{}", b.action), &id, "");
}
Ok(Json(res))
}
#[derive(Deserialize, Default)]
struct TitleBody {
model: Option<String>,
}
/// Generate a title suggestion with a chosen model (not saved).
async fn suggest_title(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, body: Option<Json<TitleBody>>) -> ApiResult {
writable(&app, &user, &id)?;
let model = body.and_then(|b| b.0.model);
let title = crate::runner::suggest_title(app.clone(), &id, model).await.map_err(|e| ApiError(StatusCode::BAD_GATEWAY, format!("{e:#}")))?;
Ok(Json(json!({"title": title})))
}
#[derive(Deserialize, Default)]
struct ForkBody {
/// Last message to keep (inclusive); all messages when missing.
message_id: Option<String>,
}
/// Copy a session (up to a message) into a new one to try another path.
async fn fork_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, body: Option<Json<ForkBody>>) -> ApiResult {
let src = writable(&app, &user, &id)?;
let body = body.map(|b| b.0).unwrap_or_default();
let messages = app.store.messages(&id)?;
let end = match &body.message_id {
Some(mid) => messages.iter().position(|m| &m.id == mid).ok_or_else(|| not_found("message"))? + 1,
None => messages.len(),
};
let now = now_ms();
let title = if src.title.is_empty() { "Fork".to_string() } else { format!("{} (fork)", src.title) };
let fork = Session {
id: new_id(),
title,
parent_id: None,
owner_id: Some(user.id.clone()),
created_at: now,
updated_at: now,
last_usage: None,
phase: None,
worktree: None,
pinned: false,
archived: false,
..src.clone()
};
let fork = app.save_session(fork)?;
for m in &messages[..end] {
let mut m = m.clone();
m.id = new_id();
app.store.add_message(&fork.id, &m)?;
}
app.audit(Some(&user), "session.fork", &format!("{id} -> {}", fork.id), "");
Ok(Json(json!(fork)))
}
#[derive(Deserialize)]
struct ExportQ {
#[serde(default)]
format: String,
}
/// Download a session as Markdown (default) or JSON.
async fn export_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<ExportQ>) -> Result<axum::response::Response, ApiError> {
let s = readable(&app, &user, &id)?;
// JSON exports are self-contained: images inline instead of store references.
let messages = if q.format == "json" { app.store.messages_for_model(&id)? } else { app.store.messages(&id)? };
let base: String = s.title.chars().map(|c| if c.is_alphanumeric() || c == '-' { c } else { '_' }).collect::<String>().trim_matches('_').to_string();
let base = if base.is_empty() { format!("session-{}", &id[..8.min(id.len())]) } else { base };
let (body, mime, ext) = if q.format == "json" {
(serde_json::to_string_pretty(&json!({"session": s, "messages": messages}))?, "application/json", "json")
} else {
(crate::export::markdown(&s, &messages), "text/markdown; charset=utf-8", "md")
};
Ok((
[
(axum::http::header::CONTENT_TYPE, mime.to_string()),
(axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{base}.{ext}\"")),
],
body,
)
.into_response())
}
async fn get_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let s = readable(&app, &user, &id)?;
let messages = app.store.messages(&id)?;
let busy = app.is_busy(&id);
let can_write = app.access(&user, &id) == Access::Full;
let queue = app.store.queue_list(&id)?;
let step = app.with_runtime(&id, |r| r.step.clone());
Ok(Json(json!({"session": s, "messages": messages, "busy": busy, "can_write": can_write, "queue": queue, "step": step})))
}
#[derive(Deserialize)]
struct UpdateSession {
title: Option<String>,
model: Option<String>,
agent: Option<String>,
permission_mode: Option<String>,
clean_room: Option<bool>,
pinned: Option<bool>,
archived: Option<bool>,
/// "" removes the folder.
folder: Option<String>,
/// Instructions for this chat ("" clears them).
instructions: Option<String>,
}
async fn update_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<UpdateSession>) -> ApiResult {
writable(&app, &user, &id)?;
// Organising the list must not move the session to "today".
let organise_only = body.title.is_none() && body.model.is_none() && body.agent.is_none() && body.clean_room.is_none() && body.permission_mode.is_none() && body.instructions.is_none();
if body.instructions.as_ref().is_some_and(|i| i.len() > 20_000) {
return Err(ApiError(StatusCode::BAD_REQUEST, "instructions too long (max 20,000 characters)".into()));
}
if let Some(m) = &body.model {
app.registry().resolve(m)?;
}
if body.permission_mode.as_deref().is_some_and(|p| !crate::permission::MODES.contains(&p)) {
return Err(ApiError(StatusCode::BAD_REQUEST, "invalid permission mode".into()));
}
// Atomic: a running turn updates the same session (usage, title, todos) meanwhile.
let s = app.modify_session(&id, !organise_only, |s| {
if let Some(t) = body.title {
s.title = t;
}
if let Some(m) = body.model {
s.model = m;
}
if let Some(a) = body.agent {
s.agent = a;
}
if let Some(c) = body.clean_room {
s.clean_room = Some(c);
}
if let Some(p) = body.pinned {
s.pinned = p;
}
if let Some(a) = body.archived {
s.archived = a;
if a {
s.pinned = false;
}
}
if let Some(f) = body.folder {
let f = f.trim().chars().take(60).collect::<String>();
s.folder = (!f.is_empty()).then_some(f);
}
if let Some(p) = body.permission_mode {
s.permission_mode = p;
}
if let Some(i) = body.instructions {
s.instructions = i.trim().to_string();
}
})?;
Ok(Json(json!(s)))
}
#[derive(Deserialize, Serialize)]
struct FileTab {
id: String,
dir: String,
}
fn file_tabs_key(session_id: &str) -> String {
format!("file_tabs:{session_id}")
}
/// Tabs of a chat for this user: its terminals and file managers (kept on the server,
/// so they are the same on every device).
async fn session_tabs(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
readable(&app, &user, &id)?;
let files: Vec<FileTab> = app.store.user_setting(&user.id, &file_tabs_key(&id))?;
Ok(Json(json!({"terminals": app.terminals.for_session(&id, &user.id), "files": files})))
}
async fn put_file_tabs(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(tabs): Json<Vec<FileTab>>) -> ApiResult {
readable(&app, &user, &id)?;
if tabs.len() > 20 {
return Err(ApiError(StatusCode::BAD_REQUEST, "too many file manager tabs".into()));
}
let roots = app.user_roots(Some(&user.id));
for t in &tabs {
let p = crate::tools::normalize(&PathBuf::from(&t.dir));
if t.id.len() > 64 || !roots.iter().any(|r| p.starts_with(r)) {
return Err(ApiError(StatusCode::FORBIDDEN, "folder is outside your folders".into()));
}
}
app.store.set_user_setting(&user.id, &file_tabs_key(&id), &tabs)?;
Ok(Json(json!({"ok": true})))
}
async fn delete_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
writable(&app, &user, &id)?;
// Notify before deleting: visibility is checked against the stored session.
app.emit(Event::SessionDeleted { session_id: id.clone() });
runner::abort(&app, &id);
crate::shells::kill_session(&app, &id);
app.terminals.kill_session(&id);
app.store.delete_session(&id)?;
// Images only this session used.
let a = app.clone();
tokio::task::spawn_blocking(move || {
if let Err(e) = a.store.gc_blobs() {
tracing::warn!("image cleanup: {e:#}");
}
});
Ok(Json(json!({"ok": true})))
}
/// An image of a chat message (see blobs.rs). Addressed by the SHA-256 of its bytes,
/// so it never changes and the browser may keep it.
async fn get_blob(Extension(_user): Extension<User>, State(app): State<AppState>, Path(hash): Path<String>) -> Result<axum::response::Response, ApiError> {
if !crate::blobs::is_hash(&hash) {
return Err(not_found("image"));
}
let (mime, bytes) = app.store.blob(&hash)?.ok_or_else(|| not_found("image"))?;
Ok((
[
(axum::http::header::CONTENT_TYPE, mime),
(axum::http::header::CACHE_CONTROL, "private, max-age=31536000, immutable".to_string()),
],
bytes,
)
.into_response())
}
#[derive(Deserialize)]
struct PromptBody {
text: String,
#[serde(default)]
images: Vec<ImageData>,
/// Reasoning effort for this turn (low, medium, high, xhigh, max).
effort: Option<String>,
}
const EFFORTS: &[&str] = &["low", "medium", "high", "xhigh", "max"];
fn set_turn_effort(app: &App, id: &str, effort: Option<&str>) -> Result<(), ApiError> {
match effort.filter(|e| !e.is_empty()) {
Some(e) if !EFFORTS.contains(&e) => Err(ApiError(StatusCode::BAD_REQUEST, format!("invalid effort '{e}'"))),
Some(e) if !app.is_busy(id) => {
app.with_runtime(id, |r| r.effort = Some(e.to_string()));
Ok(())
}
_ => Ok(()),
}
}
async fn prompt(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<PromptBody>) -> ApiResult {
let session = writable(&app, &user, &id)?;
send_prompt(&app, session, body.text, body.images, body.effort.as_deref()).map(Json)
}
/// Handle a user message for a session (web UI and Telegram): built-in commands,
/// slash command expansion, then start or queue the turn.
pub(crate) fn send_prompt(app: &AppState, session: crate::types::Session, text: String, images: Vec<ImageData>, effort: Option<&str>) -> Result<Value, ApiError> {
let id = session.id.clone();
let text = text.trim().to_string();
// Built-in commands.
if text == "/compact" {
if app.is_busy(&id) {
return Err(ApiError(StatusCode::CONFLICT, "session is busy".into()));
}
tokio::spawn(do_compact(app.clone(), id.clone()));
return Ok(json!({"ok": true, "command": "compact"}));
}
let cwd = PathBuf::from(&session.cwd);
let root = app.root_for(session.owner_id.as_deref(), &cwd);
let mut prompt_text = text.clone();
if text.starts_with('/') {
if let Some((expanded, agent, clean_room)) = prompts::expand_command(app, &root, &cwd, &text) {
prompt_text = expanded;
if agent.is_some() || clean_room.is_some() {
app.update_session(&id, |s| {
if let Some(a) = agent {
s.agent = a;
}
if clean_room.is_some() {
s.clean_room = clean_room;
}
})?;
}
}
}
set_turn_effort(app, &id, effort)?;
let mut parts = vec![Part::Text { text: prompt_text }];
for image in images {
parts.push(Part::Image { image });
}
match runner::submit(app.clone(), id, parts, "user")? {
runner::Submitted::Started => Ok(json!({"ok": true})),
runner::Submitted::Queued(item) => Ok(json!({"ok": true, "queued": item})),
}
}
async fn get_queue(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
readable(&app, &user, &id)?;
Ok(Json(json!(app.store.queue_list(&id)?)))
}
async fn delete_queue_item(Extension(user): Extension<User>, State(app): State<AppState>, Path((id, item)): Path<(String, String)>) -> ApiResult {
writable(&app, &user, &id)?;
if !app.store.queue_remove(&id, &item)? {
return Err(ApiError(StatusCode::CONFLICT, "already delivered to the agent".into()));
}
let items = app.store.queue_list(&id)?;
app.emit(Event::QueueUpdated { session_id: id, items: json!(items) });
Ok(Json(json!({"ok": true})))
}
async fn flush_queue(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
writable(&app, &user, &id)?;
Ok(Json(json!({"started": runner::flush_queue(app.clone(), &id)?})))
}
pub(crate) async fn do_compact(app: AppState, id: String) {
let cancel = tokio_util::sync::CancellationToken::new();
app.with_runtime(&id, |r| r.cancel = Some(cancel.clone()));
app.emit(Event::SessionStatus { session_id: id.clone(), busy: true });
if let Err(e) = runner::compact(app.clone(), &id, cancel, None).await {
app.emit(Event::Error { session_id: id.clone(), message: format!("compaction failed: {e:#}") });
}
app.with_runtime(&id, |r| r.cancel = None);
app.emit(Event::SessionStatus { session_id: id, busy: false });
}
async fn abort(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
writable(&app, &user, &id)?;
runner::abort(&app, &id);
Ok(Json(json!({"ok": true})))
}
async fn compact(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
writable(&app, &user, &id)?;
if app.is_busy(&id) {
return Err(ApiError(StatusCode::CONFLICT, "session is busy".into()));
}
tokio::spawn(do_compact(app.clone(), id));
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct RewindBody {
message_id: String,
/// Also restore files the agent changed since that message.
#[serde(default)]
restore_files: bool,
}
/// Remove a message and everything after it (edit & resend). Files are not restored.
async fn rewind(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<RewindBody>) -> ApiResult {
let mut s = writable(&app, &user, &id)?;
if app.is_busy(&id) {
return Err(ApiError(StatusCode::CONFLICT, "session is busy".into()));
}
let since = app.store.message_created_at(&id, &body.message_id)?.ok_or_else(|| not_found("message"))?;
let mut restored = Vec::new();
if body.restore_files {
let roots = app.user_roots(s.owner_id.as_deref());
for (path, content) in app.store.checkpoints_since(&id, since)? {
let p = PathBuf::from(&path);
// Only inside the owner's folders (they may have changed since).
if !roots.iter().any(|r| p.starts_with(r)) {
continue;
}
match content {
Some(bytes) => {
if let Some(dir) = p.parent() {
let _ = std::fs::create_dir_all(dir);
}
std::fs::write(&p, bytes)?;
}
None => {
let _ = std::fs::remove_file(&p);
}
}
restored.push(path);
}
}
app.store.delete_checkpoints_since(&id, since)?;
app.store.truncate_from(&id, &body.message_id)?;
s.last_usage = None;
app.save_session(s)?;
Ok(Json(json!({"ok": true, "restored": restored})))
}
/// Background shells of a session (and its subagents).
async fn list_shells(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
readable(&app, &user, &id)?;
let root = app.root_session_id(&id);
let list: Vec<Value> = app
.shells
.lock()
.unwrap()
.values()
.filter(|s| app.root_session_id(&s.session_id) == root)
.map(|s| s.json(20_000))
.collect();
Ok(Json(json!(list)))
}
async fn kill_shell(Extension(user): Extension<User>, State(app): State<AppState>, Path((id, shell)): Path<(String, String)>) -> ApiResult {
writable(&app, &user, &id)?;
let root = app.root_session_id(&id);
let s = app.shells.lock().unwrap().get(&shell).cloned().filter(|s| app.root_session_id(&s.session_id) == root).ok_or_else(|| not_found("shell"))?;
s.kill();
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct ChangesQuery {
since_message: String,
}
/// Files the agent changed since a message (for the rewind dialog).
async fn changed_files(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<ChangesQuery>) -> ApiResult {
readable(&app, &user, &id)?;
let since = app.store.message_created_at(&id, &q.since_message)?.unwrap_or(i64::MAX);
let files: Vec<Value> = app.store.checkpoints_since(&id, since)?.into_iter().map(|(p, c)| json!({"path": p, "created": c.is_none()})).collect();
Ok(Json(json!(files)))
}
async fn session_stats(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
readable(&app, &user, &id)?;
let mut v = app.store.session_stats(&id)?;
v["currency"] = json!(app.config().server.currency);
Ok(Json(v))
}
async fn get_shares(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
writable(&app, &user, &id)?;
Ok(Json(json!(app.store.shares(&id)?)))
}
async fn put_shares(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<AccessBody>) -> ApiResult {
let s = writable(&app, &user, &id)?;
if s.parent_id.is_some() {
return Err(ApiError(StatusCode::BAD_REQUEST, "share the parent session instead".into()));
}
let users: Vec<String> = body.users.into_iter().filter(|u| Some(u) != s.owner_id.as_ref()).collect();
app.set_shares(&id, &users)?;
app.audit(Some(&user), "session.share", format!("{id}: {} user(s)", users.len()), "");
Ok(Json(json!({"ok": true})))
}
// --- interaction ------------------------------------------------------------
/// Only users who control the session may answer its permission requests and questions.
fn check_pending(app: &App, user: &User, id: &str) -> Result<(), ApiError> {
let sid = app.pending_session(id).ok_or_else(|| not_found("request"))?;
if app.access(user, &sid) != Access::Full {
return Err(forbidden());
}
Ok(())
}
#[derive(Deserialize)]
struct PermissionBody {
reply: PermissionReply,
message: Option<String>,
/// Accepted diff hunks of a file edit (indices); missing = all.
hunks: Option<Vec<usize>>,
}
async fn reply_permission(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<PermissionBody>) -> ApiResult {
check_pending(&app, &user, &id)?;
let what = app.pending_title(&id).unwrap_or_default();
app.reply_permission(&id, body.reply, body.message, body.hunks)?;
app.audit(Some(&user), &format!("permission.{}", serde_json::to_value(body.reply).ok().and_then(|v| v.as_str().map(String::from)).unwrap_or_default()), what, "");
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct QuestionBody {
/// question text → answer (string or array of strings); null = dismissed
answers: Option<Map<String, Value>>,
}
async fn answer_question(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(body): Json<QuestionBody>) -> ApiResult {
check_pending(&app, &user, &id)?;
app.answer_question(&id, body.answers)?;
Ok(Json(json!({"ok": true})))
}
async fn pending(Extension(user): Extension<User>, State(app): State<AppState>) -> Json<Value> {
let list: Vec<Event> = app.pending_events().into_iter().filter(|e| app.event_visible(&user, e)).collect();
Json(json!(list))
}
async fn mcp_status(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
Ok(Json(app.mcp.status_json()))
}
async fn mcp_reload(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
let mcp = app.mcp.clone();
let servers = app.config().mcp.clone();
let weak = Arc::downgrade(&app);
mcp.start_all(weak, servers).await;
Ok(Json(app.mcp.status_json()))
}
// --- files ------------------------------------------------------------------
/// The directories the user may work in (own folder, shared folders).
async fn list_roots(Extension(user): Extension<User>, State(app): State<AppState>) -> Json<Value> {
let ws = app.workspace();
let roots: Vec<Value> = app
.user_roots(Some(&user.id))
.into_iter()
.map(|r| {
let rel = r.strip_prefix(&ws).map(|p| p.display().to_string()).unwrap_or_default();
json!({"path": r, "name": if rel.is_empty() { "/".to_string() } else { rel }})
})
.collect();
Json(json!(roots))
}
#[derive(Deserialize)]
struct DirQuery {
path: Option<String>,
}
/// Subdirectories for the project picker (inside the user's roots only).
async fn list_dirs(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<DirQuery>) -> ApiResult {
let dir = validate_cwd(&app, &user, q.path)?;
let mut dirs = Vec::new();
for e in std::fs::read_dir(&dir)?.flatten() {
let name = e.file_name().to_string_lossy().to_string();
if name.starts_with('.') || !e.file_type().map(|t| t.is_dir()).unwrap_or(false) {
continue;
}
dirs.push(json!({"name": name, "path": e.path(), "git": e.path().join(".git").exists()}));
}
dirs.sort_by(|a, b| a["name"].as_str().cmp(&b["name"].as_str()));
let root = app.root_for(Some(&user.id), &dir);
Ok(Json(json!({
"path": dir,
"parent": if dir == root { Value::Null } else { json!(dir.parent()) },
"root": root,
"workspace": app.workspace(),
"dirs": dirs,
})))
}
// --- audit & backup -------------------------------------------------------------
#[derive(Deserialize)]
struct AuditQuery {
limit: Option<i64>,
action: Option<String>,
}
async fn audit_log(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<AuditQuery>) -> ApiResult {
require_admin(&user)?;
Ok(Json(json!(app.store.audit_list(q.limit.unwrap_or(300).min(5000), q.action.as_deref().filter(|a| !a.is_empty()))?)))
}
/// ZIP with config.toml and a consistent database snapshot.
async fn backup(Extension(user): Extension<User>, State(app): State<AppState>) -> Result<Response, ApiError> {
require_admin(&user)?;
let zip_bytes = crate::backup::create_zip(&app).await?;
app.audit(Some(&user), "backup.download", format!("{} bytes", zip_bytes.len()), "");
let name = format!("codecraft-backup-{}.zip", chrono::Local::now().format("%Y%m%d-%H%M"));
Ok((
[
(axum::http::header::CONTENT_TYPE, "application/zip".to_string()),
(axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{name}\"")),
],
zip_bytes,
)
.into_response())
}
/// Automatic backups: stored files, last run, settings summary.
async fn list_backups(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
let cfg = app.config().backup.clone();
Ok(Json(json!({
"files": crate::backup::list(&app),
"dir": crate::backup::backup_dir(&app).display().to_string(),
"last": *app.backup.last.lock().unwrap(),
"enabled": cfg.enabled,
"hour": cfg.hour,
"keep": cfg.keep,
"s3": !cfg.s3.endpoint.trim().is_empty(),
})))
}
async fn run_backup(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
let res = crate::backup::run(&app).await?;
app.audit(Some(&user), "backup.run", &res.file, "");
*app.backup.last.lock().unwrap() = Some(res.clone());
Ok(Json(json!(res)))
}
/// Stage a restore and restart: the database is replaced before it is opened again.
/// In Docker the restart policy brings the server back up.
async fn restore_and_restart(app: &App, user: &User, bytes: Vec<u8>, source: &str) -> ApiResult {
crate::backup::stage_restore(app, bytes).await.map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
app.audit(Some(user), "backup.restore", source, "");
tracing::warn!(source, "backup staged for restore – restarting");
tokio::spawn(async {
tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
std::process::exit(0);
});
Ok(Json(json!({"ok": true, "restart": true})))
}
async fn restore_stored_backup(Extension(user): Extension<User>, State(app): State<AppState>, Path(name): Path<String>) -> ApiResult {
require_admin(&user)?;
if !crate::backup::list(&app).iter().any(|b| b.name == name) {
return Err(ApiError(StatusCode::NOT_FOUND, "no such backup".into()));
}
let bytes = tokio::fs::read(crate::backup::backup_dir(&app).join(&name)).await?;
restore_and_restart(&app, &user, bytes, &name).await
}
async fn restore_uploaded_backup(Extension(user): Extension<User>, State(app): State<AppState>, body: axum::body::Bytes) -> ApiResult {
require_admin(&user)?;
restore_and_restart(&app, &user, body.to_vec(), &format!("upload ({} bytes)", body.len())).await
}
async fn download_backup(Extension(user): Extension<User>, State(app): State<AppState>, Path(name): Path<String>) -> Result<Response, ApiError> {
require_admin(&user)?;
if !crate::backup::list(&app).iter().any(|b| b.name == name) {
return Err(ApiError(StatusCode::NOT_FOUND, "no such backup".into()));
}
let bytes = tokio::fs::read(crate::backup::backup_dir(&app).join(&name)).await?;
app.audit(Some(&user), "backup.download", &name, "");
Ok((
[
(axum::http::header::CONTENT_TYPE, "application/zip".to_string()),
(axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{name}\"")),
],
bytes,
)
.into_response())
}
/// Liveness for monitoring (no auth needed – see auth::is_public).
async fn health(State(app): State<AppState>) -> Json<Value> {
Json(json!({"ok": true, "version": crate::update::VERSION, "commit": crate::update::commit(), "sandbox": crate::sandbox::available(), "providers": app.registry().providers.len()}))
}
/// Status overlay: every enabled provider and MCP server with its state. Error texts can
/// contain URLs or hosts, so only admins get them.
async fn status_overview(Extension(user): Extension<User>, State(app): State<AppState>) -> Json<Value> {
let admin = user.is_admin();
let cfg = app.config();
let registry = app.registry();
let health = app.health.all();
let providers: serde_json::Map<String, Value> = cfg
.providers
.iter()
.filter(|(_, p)| p.enabled)
.map(|(name, _)| {
let models = registry.providers.get(name).map(|e| e.models.len()).unwrap_or(0);
let v = match health.get(name) {
Some(h) => json!({
"state": h.state,
"latency_ms": h.latency_ms,
"checked_at": h.checked_at,
"error": if admin { h.error.clone() } else { None },
"models": models,
}),
None => json!({"state": "unknown", "models": models}),
};
let mut v = v;
if let Some(limit) = cfg.providers.get(name).and_then(|p| p.plan_limit).filter(|l| *l > 0.0) {
v["plan"] = crate::budget::plan_usage(&app, name, limit).unwrap_or(Value::Null);
}
(name.clone(), v)
})
.collect();
let mcp: serde_json::Map<String, Value> = app
.mcp
.states()
.into_iter()
.filter(|(_, st)| !matches!(st, crate::mcp::McpState::Disabled))
.map(|(name, st)| {
let v = match st {
crate::mcp::McpState::Connected { tools } => json!({"state": "connected", "tools": tools}),
crate::mcp::McpState::Failed { error } => json!({"state": "failed", "error": if admin { Some(error) } else { None }}),
_ => json!({"state": "connecting"}),
};
(name, v)
})
.collect();
// Budgets and plan windows (admins: all; plan windows are shown per provider anyway).
let budgets = if admin { crate::budget::meters(&app).unwrap_or_default() } else { vec![] };
Json(json!({"version": crate::update::VERSION, "providers": providers, "mcp": mcp, "budgets": budgets, "currency": cfg.server.currency}))
}
async fn provider_status(State(app): State<AppState>) -> Json<Value> {
Json(json!(app.health.all()))
}
#[derive(Deserialize)]
struct TestProviderBody {
/// Existing provider: its stored key is used when `api_key` is empty.
#[serde(default)]
name: Option<String>,
kind: String,
base_url: String,
#[serde(default)]
api_key: String,
#[serde(default)]
auth_prefix: Option<String>,
}
/// Check URL and key from the edit form without saving them.
async fn test_provider(Extension(user): Extension<User>, State(app): State<AppState>, Json(body): Json<TestProviderBody>) -> ApiResult {
require_admin(&user)?;
let stored = body.name.as_ref().and_then(|n| app.config().providers.get(n).cloned());
let mut value = stored.as_ref().map(|p| serde_json::to_value(p)).transpose()?.unwrap_or_else(|| json!({}));
value["kind"] = json!(body.kind);
value["base_url"] = json!(body.base_url);
if !body.api_key.is_empty() {
value["api_key"] = json!(body.api_key);
}
if let Some(prefix) = body.auth_prefix {
value["auth_prefix"] = if prefix.is_empty() { Value::Null } else { json!(prefix) };
}
let pc: crate::config::ProviderConfig = serde_json::from_value(value).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("invalid provider: {e}")))?;
let provider = crate::provider::build(crate::provider::http_client(), &pc);
let (status, models) = crate::health::probe(provider).await;
Ok(Json(json!({"status": status, "models": models})))
}
async fn recheck_providers(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
crate::health::check_all(&app).await;
Ok(Json(json!(app.health.all())))
}
// --- budget -------------------------------------------------------------------
/// Budgets concern the whole installation: admins only.
async fn budget(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
require_admin(&user)?;
Ok(Json(crate::budget::overview(&app).await?))
}
async fn provider_prices(Extension(user): Extension<User>, State(app): State<AppState>, Path(name): Path<String>) -> ApiResult {
require_admin(&user)?;
Ok(Json(crate::budget::discover_prices(&app, &name).await?))
}
// --- images -------------------------------------------------------------------
async fn images_config(State(app): State<AppState>) -> Json<Value> {
let cfg = app.config();
let ic = &cfg.images;
Json(json!({
"enabled": ic.enabled && !crate::config::expand_env(&ic.api_key).is_empty(),
"default_model": ic.default_model,
"output_dir": ic.output_dir,
"sizes": crate::images::SIZES,
"models": ic.models.iter().map(|m| json!({"id": m.id, "name": m.name.clone().unwrap_or_else(|| m.id.clone()), "price": m.price})).collect::<Vec<_>>(),
"currency": cfg.server.currency,
}))
}
async fn list_images(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
Ok(Json(json!(app.store.images(Some(&user.id), 200)?)))
}
#[derive(Deserialize)]
struct ImageBody {
prompt: String,
model: Option<String>,
size: Option<String>,
#[serde(default = "one")]
count: u32,
/// Target directory (absolute inside the user's folders); default: first folder/output_dir.
directory: Option<String>,
}
fn one() -> u32 {
1
}
async fn create_images(Extension(user): Extension<User>, State(app): State<AppState>, Json(body): Json<ImageBody>) -> ApiResult {
if body.prompt.trim().is_empty() {
return Err(ApiError(StatusCode::BAD_REQUEST, "prompt is empty".into()));
}
let roots = app.user_roots(Some(&user.id));
let dir = match body.directory.filter(|d| !d.trim().is_empty()) {
Some(d) => {
let p = crate::tools::normalize(&PathBuf::from(d));
if !roots.iter().any(|r| p.starts_with(r)) {
return Err(ApiError(StatusCode::FORBIDDEN, "directory is outside your folders".into()));
}
p
}
None => roots[0].join(&app.config().images.output_dir),
};
let records = crate::images::generate(
&app,
crate::images::GenerateRequest {
prompt: body.prompt.trim().to_string(),
model: body.model,
size: body.size,
count: body.count,
dir,
user_id: Some(user.id.clone()),
session_id: None,
references: vec![],
},
)
.await?;
Ok(Json(json!(records)))
}
fn own_image(app: &App, user: &User, id: &str) -> Result<crate::store::ImageRecord, ApiError> {
let img = app.store.image(id)?.ok_or_else(|| not_found("image"))?;
if !user.is_admin() && img.user_id.as_deref() != Some(user.id.as_str()) {
return Err(ApiError(StatusCode::FORBIDDEN, "not your image".into()));
}
Ok(img)
}
async fn image_file(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> Result<Response, ApiError> {
let img = own_image(&app, &user, &id)?;
let bytes = tokio::fs::read(&img.path).await.map_err(|_| not_found("image file"))?;
let ctype = match std::path::Path::new(&img.path).extension().and_then(|e| e.to_str()) {
Some("jpg") | Some("jpeg") => "image/jpeg",
Some("webp") => "image/webp",
_ => "image/png",
};
Ok(([(axum::http::header::CONTENT_TYPE, ctype), (axum::http::header::CACHE_CONTROL, "private, max-age=86400")], bytes).into_response())
}
async fn delete_image(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let img = own_image(&app, &user, &id)?;
// Only delete files inside the owner's folders.
let roots = app.user_roots(img.user_id.as_deref());
let p = PathBuf::from(&img.path);
if roots.iter().any(|r| p.starts_with(r)) {
let _ = tokio::fs::remove_file(&p).await;
}
app.store.delete_image(&id)?;
Ok(Json(json!({"ok": true})))
}
// --- statistics -------------------------------------------------------------
#[derive(Deserialize)]
struct StatsQuery {
from: Option<i64>,
to: Option<i64>,
/// Minutes east of UTC (JS: -new Date().getTimezoneOffset()).
tz: Option<i64>,
/// Admins: user id to filter by (omitted = all users).
user: Option<String>,
}
async fn stats(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<StatsQuery>) -> ApiResult {
let to = q.to.unwrap_or_else(|| now_ms() + 1);
let from = q.from.unwrap_or(to - 30 * 86_400_000);
let filter = if user.is_admin() { q.user.filter(|u| !u.is_empty()) } else { Some(user.id.clone()) };
let mut v = app.store.global_stats(from, to, q.tz.unwrap_or(0), filter.as_deref())?;
v["currency"] = json!(app.config().server.currency);
v["from"] = json!(from);
v["to"] = json!(to);
Ok(Json(v))
}
// --- websocket --------------------------------------------------------------
// --- automations (scheduled agents) ------------------------------------------------
fn own_job(app: &App, user: &User, id: &str) -> Result<crate::jobs::Job, ApiError> {
let job = app.store.job(id)?.ok_or_else(|| not_found("automation"))?;
if job.owner_id != user.id && !user.is_admin() {
return Err(forbidden());
}
Ok(job)
}
async fn list_jobs(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let jobs: Vec<Value> = app.store.jobs(Some(&user.id))?.iter().map(crate::jobs::describe).collect();
Ok(Json(json!(jobs)))
}
#[derive(Deserialize)]
struct RegisterJob {
path: String,
}
async fn register_job(Extension(user): Extension<User>, State(app): State<AppState>, Json(b): Json<RegisterJob>) -> ApiResult {
let p = crate::files::checked_path(&app, &user, &b.path, true)?;
crate::jobs::parse(&p).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
let job = app.store.job_register(&user.id, &p.display().to_string())?;
app.audit(Some(&user), "job.register", &job.path, "");
Ok(Json(crate::jobs::describe(&job)))
}
#[derive(Deserialize)]
struct CreateJob {
/// Project folder; the file goes to <cwd>/.codecraft/jobs/<name>.md.
cwd: String,
#[serde(flatten)]
job: crate::jobs::NewJob,
}
async fn create_job(Extension(user): Extension<User>, State(app): State<AppState>, Json(b): Json<CreateJob>) -> ApiResult {
let dir = crate::files::checked_path(&app, &user, &b.cwd, true)?;
if b.job.name.trim().is_empty() || b.job.prompt.trim().is_empty() {
return Err(ApiError(StatusCode::BAD_REQUEST, "name and prompt are required".into()));
}
if !crate::permission::MODES.contains(&b.job.permission_mode.as_str()) {
return Err(ApiError(StatusCode::BAD_REQUEST, "invalid permission mode".into()));
}
// Inside the checked folder: the file name is a slug of the name.
let path = crate::jobs::new_job_path(&dir, &b.job.name);
crate::jobs::write_file(&path, &b.job).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
let job = app.store.job_register(&user.id, &path.display().to_string())?;
app.audit(Some(&user), "job.create", &job.path, "");
Ok(Json(crate::jobs::describe(&job)))
}
#[derive(Deserialize)]
struct UpdateJob {
enabled: Option<bool>,
}
async fn update_job(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Json(b): Json<UpdateJob>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
if let Some(e) = b.enabled {
app.store.job_set_enabled(&job.id, e)?;
}
let job = app.store.job(&id)?.ok_or_else(|| not_found("automation"))?;
Ok(Json(crate::jobs::describe(&job)))
}
async fn delete_job(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
app.store.job_delete(&job.id)?;
app.audit(Some(&user), "job.delete", &job.path, "");
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct WebhookSettings {
enable: bool,
}
/// Enable (new secret, shown once) or disable the webhook of an automation.
async fn job_webhook_settings(Extension(user): Extension<User>, State(app): State<AppState>, headers: axum::http::HeaderMap, Path(id): Path<String>, Json(b): Json<WebhookSettings>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
if !b.enable {
app.store.job_set_webhook(&job.id, None)?;
app.audit(Some(&user), "job.webhook.off", &job.path, "");
return Ok(Json(json!({"ok": true})));
}
let secret = crate::auth::random_token(24);
app.store.job_set_webhook(&job.id, Some(&secret))?;
app.audit(Some(&user), "job.webhook.on", &job.path, "");
let base = public_base(&app, &headers);
Ok(Json(json!({"url": format!("{base}/hooks/{}", job.id), "secret": secret})))
}
#[derive(Deserialize)]
struct HookQuery {
secret: Option<String>,
}
/// Public webhook endpoint: starts the automation with the request body as input.
async fn incoming_webhook(State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<HookQuery>, headers: axum::http::HeaderMap, body: bytes::Bytes) -> Response {
let ip = crate::auth::client_ip(&headers);
let Ok(Some(job)) = app.store.job(&id) else { return (StatusCode::NOT_FOUND, "unknown webhook").into_response() };
let Ok(Some(secret)) = app.store.job_webhook_secret(&id) else { return (StatusCode::NOT_FOUND, "unknown webhook").into_response() };
if !crate::jobs::verify_webhook(&secret, &headers, q.secret.as_deref(), &body) {
app.store.audit(None, "", "webhook.denied", &id, &ip);
return (StatusCode::UNAUTHORIZED, "invalid secret or signature").into_response();
}
if !job.enabled {
return (StatusCode::CONFLICT, "the automation is disabled").into_response();
}
if job.last_session_id.as_deref().is_some_and(|s| app.is_busy(s)) {
return (StatusCode::CONFLICT, "the automation is still running").into_response();
}
if job.last_run.is_some_and(|t| now_ms() - t < 10_000) {
return (StatusCode::TOO_MANY_REQUESTS, "too many calls – wait a few seconds").into_response();
}
let payload = crate::jobs::payload_text(&headers, &body);
match crate::jobs::run_with(&app, &job, "webhook", Some(payload)).await {
Ok(sid) => {
app.store.audit(None, "", "webhook.run", &id, &ip);
(StatusCode::ACCEPTED, Json(json!({"ok": true, "session_id": sid}))).into_response()
}
Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
}
}
async fn run_job(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
let sid = crate::jobs::run(&app, &job, "manual").await.map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
Ok(Json(json!({"session_id": sid})))
}
// --- git credentials ---------------------------------------------------------------
async fn get_git_settings(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let s: crate::gitcreds::GitSettings = app.store.user_setting(&user.id, "git")?;
Ok(Json(json!(crate::gitcreds::masked(&s))))
}
async fn put_git_settings(Extension(user): Extension<User>, State(app): State<AppState>, headers: axum::http::HeaderMap, Json(b): Json<crate::gitcreds::GitSettings>) -> ApiResult {
let old: crate::gitcreds::GitSettings = app.store.user_setting(&user.id, "git")?;
let new = crate::gitcreds::merge(&old, b);
crate::gitcreds::write_files(&app, &user, &new).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")))?;
app.store.set_user_setting(&user.id, "git", &new)?;
let hosts: Vec<&str> = new.credentials.iter().map(|c| c.host.as_str()).collect();
app.audit(Some(&user), "git.credentials", hosts.join(", "), &crate::auth::client_ip(&headers));
Ok(Json(json!(crate::gitcreds::masked(&new))))
}
// --- web terminal ------------------------------------------------------------------
#[derive(Deserialize)]
struct TerminalQuery {
session: Option<String>,
}
/// All terminals of the user, or the tabs of one chat (?session=…).
async fn list_terminals(Extension(user): Extension<User>, State(app): State<AppState>, Query(q): Query<TerminalQuery>) -> ApiResult {
Ok(Json(json!(match q.session {
Some(s) => app.terminals.for_session(&s, &user.id),
None => app.terminals.list(&user.id),
})))
}
#[derive(Deserialize)]
struct OpenTerminal {
cwd: Option<String>,
/// Open as a tab of this chat (in its folder unless `cwd` is given).
session_id: Option<String>,
cols: Option<u16>,
rows: Option<u16>,
}
async fn open_terminal(Extension(user): Extension<User>, State(app): State<AppState>, headers: axum::http::HeaderMap, Json(b): Json<OpenTerminal>) -> ApiResult {
let session = match &b.session_id {
Some(id) => Some(writable(&app, &user, id)?),
None => None,
};
let cwd = validate_cwd(&app, &user, b.cwd.or_else(|| session.as_ref().map(|s| s.cwd.clone())))?;
let term = crate::terminal::spawn(&app, &user, &cwd, session.map(|s| s.id), b.cols.unwrap_or(100), b.rows.unwrap_or(30)).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
app.audit(Some(&user), "terminal.open", cwd.display().to_string(), &crate::auth::client_ip(&headers));
Ok(Json(term.info()))
}
async fn close_terminal(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
Ok(Json(json!({"ok": app.terminals.kill(&id, &user.id)})))
}
async fn terminal_ws(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, upgrade: WebSocketUpgrade) -> Response {
match app.terminals.get(&id, &user.id) {
Some(term) => upgrade.on_upgrade(move |socket| crate::terminal::attach(term, socket)),
None => (StatusCode::NOT_FOUND, "terminal not found").into_response(),
}
}
// --- web push -------------------------------------------------------------------
async fn push_key(State(app): State<AppState>) -> ApiResult {
Ok(Json(json!({"key": app.push.public_key()})))
}
#[derive(Deserialize)]
struct PushKeys {
p256dh: String,
auth: String,
}
#[derive(Deserialize)]
struct PushSubscription {
endpoint: String,
keys: PushKeys,
}
async fn push_subscribe(Extension(user): Extension<User>, State(app): State<AppState>, Json(sub): Json<PushSubscription>) -> ApiResult {
if !sub.endpoint.starts_with("https://") {
return Err(ApiError(StatusCode::BAD_REQUEST, "invalid push endpoint".into()));
}
app.store.push_add(&user.id, &sub.endpoint, &sub.keys.p256dh, &sub.keys.auth)?;
Ok(Json(json!({"ok": true})))
}
#[derive(Deserialize)]
struct PushEndpoint {
endpoint: String,
}
async fn push_unsubscribe(Extension(user): Extension<User>, State(app): State<AppState>, Json(b): Json<PushEndpoint>) -> ApiResult {
// Only the owner's own subscription.
if app.store.push_subscriptions(&user.id)?.iter().any(|s| s.endpoint == b.endpoint) {
app.store.push_remove(&b.endpoint)?;
}
Ok(Json(json!({"ok": true})))
}
async fn get_digest(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let on: bool = app.store.user_setting(&user.id, "cost_push")?;
Ok(Json(json!({"enabled": on})))
}
#[derive(Deserialize)]
struct DigestBody {
enabled: bool,
}
async fn put_digest(Extension(user): Extension<User>, State(app): State<AppState>, Json(b): Json<DigestBody>) -> ApiResult {
app.store.set_user_setting(&user.id, "cost_push", &b.enabled)?;
Ok(Json(json!({"enabled": b.enabled})))
}
/// Send today's summary right away (preview).
async fn send_digest(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let payload = crate::digest::summary(&app, &user.id).unwrap_or_else(|| json!({"kind": "digest", "title": "CodeCraft", "body": "No model requests today.", "tag": "cost-digest", "url": "/#/stats"}));
let (sent, errors) = crate::push::send_to_user(&app, &user.id, &payload).await;
Ok(Json(json!({"sent": sent, "errors": errors, "body": payload["body"]})))
}
async fn push_test(Extension(user): Extension<User>, State(app): State<AppState>) -> ApiResult {
let (sent, errors) = crate::push::send_to_user(&app, &user.id, &json!({"kind": "test", "title": "CodeCraft", "body": "Notifications work.", "tag": "test", "url": "/"})).await;
Ok(Json(json!({"sent": sent, "errors": errors})))
}
async fn ws(Extension(user): Extension<User>, State(app): State<AppState>, upgrade: WebSocketUpgrade) -> Response {
upgrade.on_upgrade(move |socket| ws_loop(app, user, socket))
}
async fn ws_loop(app: AppState, user: User, mut socket: WebSocket) {
let mut rx = app.subscribe();
for ev in app.pending_events() {
if app.event_visible(&user, &ev) && send_event(&mut socket, &ev).await.is_err() {
return;
}
}
let mut ping = tokio::time::interval(std::time::Duration::from_secs(25));
// Re-check the login regularly so disabled users lose their live connection.
let mut recheck = tokio::time::interval(std::time::Duration::from_secs(60));
let mut visible = false;
loop {
tokio::select! {
ev = rx.recv() => match ev {
Ok(ev) => {
if app.event_visible(&user, &ev) && send_event(&mut socket, &ev).await.is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
// Client fell behind: tell it to resync.
let msg = json!({"type": "resync", "skipped": n}).to_string();
if socket.send(WsMessage::Text(msg.into())).await.is_err() {
break;
}
}
Err(_) => break,
},
msg = socket.recv() => match msg {
Some(Ok(WsMessage::Close(_))) | None | Some(Err(_)) => break,
Some(Ok(WsMessage::Text(t))) => {
// The client reports whether its window is visible (push only when none is).
if let Ok(v) = serde_json::from_str::<Value>(&t) {
if v["type"] == "visibility" {
let now = v["visible"].as_bool().unwrap_or(false);
if now != visible {
app.push.set_visible(&user.id, if now { 1 } else { -1 });
visible = now;
}
}
}
}
_ => {}
},
_ = ping.tick() => {
if socket.send(WsMessage::Ping(Vec::new().into())).await.is_err() {
break;
}
}
_ = recheck.tick() => {
if !matches!(app.store.user(&user.id), Ok(Some(u)) if !u.disabled) {
break;
}
}
}
}
if visible {
app.push.set_visible(&user.id, -1);
}
}
async fn send_event(socket: &mut WebSocket, ev: &Event) -> Result<(), axum::Error> {
let text = serde_json::to_string(ev).unwrap_or_default();
socket.send(WsMessage::Text(text.into())).await
}