zink/codecraft
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity
Files
c3df4a23cc62ce7df0c5d01daec52fb4ad5928f7
codecraft/server/src/api.rs
T
Richard ZinkandClaude Opus 5.5 1e8418336d Value opencode Go plan requests; per-duration cache prices; usage under answers 
- opencode Go reports plan requests as cost 0 with the usage in allowance;
  ModelConfig::cost values them at the model's list prices (charged amounts
  and free requests stay as reported). Rows booked at 0 are valued once on
  start (idempotent).
- opencode's price list now has per-duration cache prices for some models
  ({"30m": ..., "12h": ...}): the shortest duration is used, also in tiers.
  Saving the provider's models failed with "untagged enum ModelEntry".
- Each answer shows tokens sent/received and the cost of the turn's requests
  (usage.cost filled when messages are saved and loaded).
- The chat title keeps a minimum width in a crowded top bar.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-12 07:58:23 +01:00

2189 lines
94 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}/messages", get(session_messages))
.route("/sessions/{id}/search", get(search_in_session))
.route("/sessions/{id}/media", get(session_media))
.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/sidebar", get(crate::cli_api::get_sidebar).put(crate::cli_api::put_sidebar))
.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/schedule", get(schedule_preview))
.route("/jobs/{id}", get(get_job).patch(update_job).delete(delete_job))
.route("/jobs/{id}/assistant", post(job_assistant))
.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 static_files = tower_http::services::ServeDir::new(&web_dir).fallback(missing_static_file(web_dir.clone()));
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)
}
/// Paths that are not a file of the web build. Usually a client-side route → index.html,
/// except for:
/// - `/sw-<build>.js` of an earlier build: installed apps check their service worker at
/// that URL for updates. Answering with index.html (not a script) makes the update
/// fail, so the old worker kept serving the old app shell forever – which broke when
/// the API changed. They get the kill switch /sw.js instead: it removes the old worker
/// and reloads, and the page then registers the current worker.
/// - other missing scripts/styles (old chunks): 404 instead of HTML.
fn missing_static_file(web_dir: PathBuf) -> axum::routing::MethodRouter {
use axum::http::{header, StatusCode};
use axum::response::IntoResponse;
axum::routing::any(move |uri: axum::http::Uri| {
let web_dir = web_dir.clone();
async move {
let path = uri.path();
let name = path.trim_start_matches('/');
if name.starts_with("sw-") && name.ends_with(".js") && !name.contains('/') {
return match tokio::fs::read(web_dir.join("sw.js")).await {
Ok(js) => ([(header::CONTENT_TYPE, "text/javascript"), (header::CACHE_CONTROL, "no-cache, no-store")], js).into_response(),
Err(_) => StatusCode::NOT_FOUND.into_response(),
};
}
if [".js", ".mjs", ".css", ".map"].iter().any(|ext| path.ends_with(ext)) {
return StatusCode::NOT_FOUND.into_response();
}
match tokio::fs::read(web_dir.join("index.html")).await {
Ok(html) => ([(header::CONTENT_TYPE, "text/html; charset=utf-8"), (header::CACHE_CONTROL, "no-cache")], html).into_response(),
Err(_) => StatusCode::NOT_FOUND.into_response(),
}
}
})
}
/// 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(),
// New chats keep their approvals and questions out of Telegram (⋯ menu turns it on).
telegram_quiet: true,
};
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())
}
#[derive(Deserialize)]
struct WindowQ {
/// Only the end of the chat (from the last summary on, at most this many messages);
/// without it, all messages (terminal client).
limit: Option<i64>,
before: Option<String>,
after: Option<String>,
around: Option<String>,
/// Loading older messages stops at a summary (the history before it on request).
#[serde(default)]
stop_at_summary: bool,
}
impl WindowQ {
fn window(&self) -> Option<crate::store::MessageWindow> {
use crate::store::MessageWindow as W;
let limit = self.limit.unwrap_or(200).clamp(1, 1000);
Some(match (&self.before, &self.after, &self.around) {
(Some(id), _, _) => W::Before { id: id.clone(), limit, stop_at_summary: self.stop_at_summary },
(_, Some(id), _) => W::After { id: id.clone(), limit },
(_, _, Some(id)) => W::Around { id: id.clone(), limit },
_ if self.limit.is_some() => W::Tail { limit },
_ => return None,
})
}
}
/// The cost of each answer for the chat view (usage.cost: charged, or valued at list prices).
fn with_costs(app: &App, mut messages: Vec<crate::types::Message>) -> Vec<crate::types::Message> {
let registry = app.registry();
for m in &mut messages {
if let (Some(u), Some(spec)) = (m.usage.as_mut(), m.model.as_deref()) {
if let Ok((_, model)) = registry.resolve(spec) {
u.cost = Some(model.cost(u));
}
}
}
messages
}
async fn get_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<WindowQ>) -> ApiResult {
let s = readable(&app, &user, &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());
let mut out = json!({"session": s, "busy": busy, "can_write": can_write, "queue": queue, "step": step});
match q.window() {
Some(w) => {
let (messages, older, newer) = app.store.messages_window(&id, &w)?;
out["messages"] = json!(with_costs(&app, messages));
out["has_older"] = json!(older);
out["has_newer"] = json!(newer);
}
None => out["messages"] = json!(app.store.messages(&id)?),
}
Ok(Json(out))
}
/// More messages of a chat: before / after / around a message.
async fn session_messages(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<WindowQ>) -> ApiResult {
readable(&app, &user, &id)?;
let w = q.window().ok_or_else(|| ApiError(StatusCode::BAD_REQUEST, "before, after, around or limit is required".into()))?;
let (messages, older, newer) = app.store.messages_window(&id, &w).map_err(|e| ApiError(StatusCode::NOT_FOUND, format!("{e:#}")))?;
Ok(Json(json!({"messages": with_costs(&app, messages), "has_older": older, "has_newer": newer})))
}
/// Images, videos and audio files shown in a chat (all of it, with subagents).
async fn session_media(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
readable(&app, &user, &id)?;
let a = app.clone();
let items = tokio::task::spawn_blocking(move || crate::media::collect(&a, &id)).await.map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))??;
Ok(Json(json!(items)))
}
/// Search within one chat (all of it, not only the loaded part).
async fn search_in_session(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>, Query(q): Query<SearchQ>) -> ApiResult {
readable(&app, &user, &id)?;
if q.q.trim().chars().count() < 2 {
return Ok(Json(json!([])));
}
let hits: Vec<Value> = app
.store
.search_messages(&id, q.q.trim(), 100)?
.into_iter()
.map(|(mid, role, at, snippet)| json!({"id": mid, "role": role, "created_at": at, "snippet": snippet}))
.collect();
Ok(Json(json!(hits)))
}
#[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>,
/// Approvals and questions of this chat are not sent to Telegram.
telegram_quiet: Option<bool>,
}
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();
}
if let Some(q) = body.telegram_quiet {
s.telegram_quiet = q;
}
})?;
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 ctype_of = |p: &str| match std::path::Path::new(p).extension().and_then(|e| e.to_str()).map(|e| e.to_ascii_lowercase()).as_deref() {
Some("jpg") | Some("jpeg") => "image/jpeg",
Some("webp") => "image/webp",
_ => "image/png",
};
let respond = |ctype: String, bytes: Vec<u8>| ([(axum::http::header::CONTENT_TYPE, ctype), (axum::http::header::CACHE_CONTROL, "private, max-age=86400".to_string())], bytes).into_response();
if let Ok(bytes) = tokio::fs::read(&img.path).await {
return Ok(respond(ctype_of(&img.path).into(), bytes));
}
// The file was moved or renamed (agents tidy up folders): with a copy in the image store,
// answer from there and look for the file in the background; otherwise look now.
if let Some((mime, bytes)) = img.blob.as_deref().map(|h| app.store.blob(h)).transpose()?.flatten() {
let a = app.clone();
tokio::task::spawn_blocking(move || crate::images::relocate(&a, &img));
return Ok(respond(mime, bytes));
}
let a = app.clone();
let found = tokio::task::spawn_blocking(move || crate::images::relocate(&a, &img)).await.ok().flatten().ok_or_else(|| not_found("image file"))?;
let bytes = tokio::fs::read(&found).await.map_err(|_| not_found("image file"))?;
Ok(respond(ctype_of(&found.display().to_string()).into(), bytes))
}
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)))
}
async fn get_job(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
Ok(Json(crate::jobs::describe(&job)))
}
#[derive(Deserialize)]
struct RegisterJob {
path: String,
}
/// Register an existing Markdown file (older files: settings from their frontmatter).
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)?;
if !p.is_file() {
return Err(ApiError(StatusCode::BAD_REQUEST, format!("{} is not a file", p.display())));
}
let job = app.store.job_register(&user.id, &p.display().to_string())?;
let job = crate::jobs::adopt(&app.store, job).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
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,
name: String,
/// Initial content of the instructions file (example or template).
prompt: String,
#[serde(default)]
schedule: Option<String>,
#[serde(default)]
agent: Option<String>,
#[serde(default)]
permission_mode: Option<String>,
/// Default false: the user switches it on once prompt and schedule are ready.
#[serde(default)]
enabled: bool,
}
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.prompt.trim().is_empty() {
return Err(ApiError(StatusCode::BAD_REQUEST, "the instructions are empty".into()));
}
let mut settings = crate::jobs::JobSettings::new(&b.name);
if let Some(x) = b.schedule {
settings.schedule = x;
}
if let Some(x) = b.agent {
settings.agent = x;
}
if let Some(x) = b.permission_mode {
settings.permission_mode = x;
}
// Inside the checked folder: the file name is a slug of the name.
let path = crate::jobs::new_job_path(&dir, &b.name);
let job = crate::jobs::create(&app.store, &user.id, &path, settings, &b.prompt, b.enabled).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
app.audit(Some(&user), "job.create", &job.path, "");
Ok(Json(crate::jobs::describe(&job)))
}
#[derive(Deserialize)]
struct UpdateJob {
enabled: Option<bool>,
name: Option<String>,
schedule: Option<String>,
agent: Option<String>,
/// "" = the server's default model.
model: Option<String>,
permission_mode: Option<String>,
timeout_minutes: Option<u64>,
}
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)?;
let job = crate::jobs::adopt(&app.store, job).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
let changes = b.name.is_some() || b.schedule.is_some() || b.agent.is_some() || b.model.is_some() || b.permission_mode.is_some() || b.timeout_minutes.is_some();
if changes {
let mut st = job.settings.clone().unwrap_or_else(|| crate::jobs::JobSettings::new("automation"));
if let Some(x) = b.name {
st.name = x;
}
if let Some(x) = b.schedule {
st.schedule = x;
}
if let Some(x) = b.agent {
st.agent = x;
}
if let Some(x) = b.model {
if !x.is_empty() {
app.registry().resolve(&x)?;
}
st.model = Some(x);
}
if let Some(x) = b.permission_mode {
st.permission_mode = x;
}
if let Some(x) = b.timeout_minutes {
st.timeout_minutes = x;
}
st.validate().map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
app.store.job_set_settings(&job.id, &st)?;
app.audit(Some(&user), "job.settings", &job.path, &st.schedule);
}
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)))
}
/// Chat in which the user works out the instructions with a model (created once, reused
/// while it exists). The model edits the job file directly.
async fn job_assistant(Extension(user): Extension<User>, State(app): State<AppState>, Path(id): Path<String>) -> ApiResult {
let job = own_job(&app, &user, &id)?;
let job = crate::jobs::adopt(&app.store, job).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("{e:#}")))?;
let mut st = job.settings.clone().ok_or_else(|| ApiError(StatusCode::BAD_REQUEST, "automation without settings".into()))?;
if let Some(sid) = st.assistant_session.as_deref() {
if app.store.get_session(sid)?.is_some_and(|s| s.owner_id.as_deref() == Some(user.id.as_str())) {
return Ok(Json(json!({"session_id": sid})));
}
}
let path = std::path::Path::new(&job.path);
let cwd = crate::jobs::cwd_of(&job.path, st.cwd.as_deref());
let cfg = app.config();
let model = st
.model
.clone()
.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 now = now_ms();
let instructions = format!(
"You help the user write the instructions of the automation \"{name}\". They live in the Markdown file {file} \
and are executed unattended by an agent (mode {agent}) on a schedule the user sets in the UI. That run starts a \
fresh session and sees only this file – no conversation, nobody to ask.\n\
- Work on that file directly with your edit tools and keep it up to date after every round; the user sees it next to this chat.\n\
- Find out what the automation should achieve, then shape the file into clear, self-contained instructions: \
goal, steps, expected result (e.g. a short report), limits (what must not be done).\n\
- Write only the instructions: no frontmatter, no schedule or settings in the file.\n\
- Do not carry out the automation's task yourself and do not create other files. Use the language the user writes in.",
name = st.name,
file = path.display(),
agent = st.agent,
);
let s = Session {
id: new_id(),
title: format!("{} – {}", st.name, if crate::telegram::german(&app, &user.id) { "Anweisungen" } else { "instructions" }),
cwd: cwd.display().to_string(),
model,
agent: "build".into(),
permission_mode: "auto".into(),
parent_id: None,
owner_id: Some(user.id.clone()),
created_at: now,
updated_at: now,
todos: Value::Null,
last_usage: None,
clean_room: None,
phase: None,
worktree: None,
pinned: false,
archived: false,
folder: None,
instructions,
telegram_quiet: true,
};
let s = app.save_session(s)?;
st.assistant_session = Some(s.id.clone());
app.store.job_set_settings(&job.id, &st)?;
Ok(Json(json!({"session_id": s.id})))
}
#[derive(Deserialize)]
struct ScheduleQ {
expr: String,
}
/// Check a cron expression and list its next runs (for the schedule editor).
async fn schedule_preview(Query(q): Query<ScheduleQ>) -> ApiResult {
let expr = q.expr.trim();
if let Err(e) = crate::jobs::check_schedule(expr) {
return Ok(Json(json!({"ok": false, "error": format!("{e:#}")})));
}
let mut next = Vec::new();
let mut t = now_ms();
for _ in 0..5 {
match crate::jobs::next_run(expr, t) {
Some(n) => {
next.push(n);
t = n + 1000;
}
None => break,
}
}
// Wall-clock times of the server: cron works in its time zone, the browser may be elsewhere.
let local: Vec<String> = next.iter().filter_map(|n| chrono::DateTime::from_timestamp_millis(*n)).map(|t| t.with_timezone(&chrono::Local).format("%Y-%m-%dT%H:%M").to_string()).collect();
let tz = std::env::var("TZ").ok().filter(|z| !z.is_empty()).unwrap_or_else(|| chrono::Local::now().format("UTC%:z").to_string());
Ok(Json(json!({"ok": true, "next": next, "next_local": local, "tz": tz})))
}
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
}
Reference in new issue View Git Blame Copy Permalink