#![deny(unsafe_code)]
use axum::extract::State;
use axum::http::{HeaderValue, StatusCode};
use axum::response::sse::{Event, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::{Json, Router};
use futures_util::StreamExt;
use futures_util::stream;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
pub const FREEZE_VERSION: &str = "1.18.31";
pub fn deep_merge(base: &mut serde_json::Value, patch: serde_json::Value) {
match (base, patch) {
(serde_json::Value::Object(b), serde_json::Value::Object(p)) => {
for (k, v) in p {
match b.get_mut(&k) {
Some(bv) => deep_merge(bv, v),
None => {
b.insert(k, v);
}
}
}
}
(b, p) => *b = p,
}
}
pub fn config_write_path() -> std::path::PathBuf {
let home = std::env::var("HOME").unwrap_or_else(|_| "/".into());
if std::env::var("OCSERVE_CONFIG_WRITE")
.map(|v| v == "overlay")
.unwrap_or(false)
{
std::path::PathBuf::from(home).join(".config/ocserve/config.json")
} else {
std::path::PathBuf::from(home).join(".config/opencode/opencode.json")
}
}
pub type ConfigReloader =
std::sync::Arc<dyn Fn() -> anyhow::Result<(Payloads, LlmRegistry)> + Send + Sync>;
#[derive(Default)]
pub struct Payloads {
pub config: serde_json::Value,
pub agent: Vec<serde_json::Value>,
pub api_agent: Vec<serde_json::Value>,
pub command: Vec<serde_json::Value>,
pub config_providers: serde_json::Value,
pub provider: serde_json::Value,
pub console: serde_json::Value,
pub capabilities: serde_json::Value,
pub compaction: serde_json::Value,
pub catalog: std::sync::Arc<serde_json::Value>,
}
async fn run_blocking<T, F>(db: &std::path::Path, op: F) -> anyhow::Result<T>
where
F: FnOnce(&std::path::Path) -> anyhow::Result<T> + Send + 'static,
T: Send + 'static,
{
let db = db.to_path_buf();
tokio::task::spawn_blocking(move || op(&db))
.await
.map_err(|e| anyhow::anyhow!("blocking task join: {e}"))?
}
pub mod compress;
pub mod doc;
pub mod embedded_ui;
pub mod pty;
pub mod tui;
pub mod ui;
pub mod v2;
pub mod vcs;
pub struct ApiError {
pub status: StatusCode,
pub name: &'static str,
pub message: String,
}
impl ApiError {
pub fn not_found(msg: impl Into<String>) -> Self {
Self {
status: StatusCode::NOT_FOUND,
name: "NotFoundError",
message: msg.into(),
}
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let body = json!({"name": self.name, "data": {"message": self.message}});
(self.status, Json(body)).into_response()
}
}
pub mod sync;
pub mod watch;
pub enum HttpError {
Api(ApiError),
Tagged {
status: StatusCode,
tag: &'static str,
},
Query {
message: String,
},
Payload {
message: String,
},
Params {
message: String,
},
TaggedData {
status: StatusCode,
tag: &'static str,
fields: serde_json::Value,
},
}
impl From<ApiError> for HttpError {
fn from(e: ApiError) -> Self {
HttpError::Api(e)
}
}
impl IntoResponse for HttpError {
fn into_response(self) -> Response {
match self {
HttpError::Api(e) => e.into_response(),
HttpError::Tagged { status, tag } => {
let body = format!("{{\"_tag\":\"{tag}\"}}");
(
status,
[(axum::http::header::CONTENT_TYPE, "application/json")],
body,
)
.into_response()
}
HttpError::TaggedData {
status,
tag,
fields,
} => {
let mut obj = serde_json::Map::new();
obj.insert("_tag".into(), serde_json::Value::String(tag.into()));
if let serde_json::Value::Object(f) = fields {
obj.extend(f);
}
(
status,
[(axum::http::header::CONTENT_TYPE, "application/json")],
serde_json::Value::Object(obj).to_string(),
)
.into_response()
}
HttpError::Query { message } => {
let body = serde_json::json!({
"name": "BadRequest",
"data": {"message": message, "kind": "Query"}
})
.to_string();
(
StatusCode::BAD_REQUEST,
[(axum::http::header::CONTENT_TYPE, "application/json")],
body,
)
.into_response()
}
HttpError::Payload { message } => {
let body = serde_json::json!({
"name": "BadRequest",
"data": {"message": message, "kind": "Payload"}
})
.to_string();
(
StatusCode::BAD_REQUEST,
[(axum::http::header::CONTENT_TYPE, "application/json")],
body,
)
.into_response()
}
HttpError::Params { message } => {
let body = serde_json::json!({
"name": "BadRequest",
"data": {"message": message, "kind": "Params"}
})
.to_string();
(
StatusCode::BAD_REQUEST,
[(axum::http::header::CONTENT_TYPE, "application/json")],
body,
)
.into_response()
}
}
}
}
pub type PromptTask = (u64, tokio::task::JoinHandle<()>);
pub struct AppState {
pub payloads: parking_lot::RwLock<Payloads>,
pub wire: parking_lot::RwLock<HashMap<&'static str, WireEntry>>,
pub wire_off: std::sync::atomic::AtomicBool,
pub reloader: parking_lot::RwLock<Option<ConfigReloader>>,
pub paths: serde_json::Value,
pub auth: Option<(String, String)>,
pub requests: std::sync::atomic::AtomicU64,
pub bus: ocserve_core::EventBus,
pub db: std::path::PathBuf,
pub blobs: std::sync::Arc<ocserve_store::BlobStore>,
pub writer: std::sync::Arc<ocserve_store::Writer>,
pub llm: parking_lot::RwLock<LlmRegistry>,
pub watch: parking_lot::RwLock<watch::WatchState>,
pub reconcile_lock: tokio::sync::Mutex<()>,
pub max_rounds: usize,
pub cost_ceiling: f64,
pub gate: std::sync::Arc<ocserve_core::PermissionGate>,
pub mcp: std::sync::OnceLock<std::sync::Arc<ocserve_mcp::McpHub>>,
pub plugins: std::sync::OnceLock<std::sync::Arc<tokio::sync::Mutex<ocserve_plugin::Sidecar>>>,
pub prompt_locks: std::sync::Arc<
parking_lot::Mutex<
std::collections::HashMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>,
>,
>,
pub prompt_tasks:
std::sync::Arc<parking_lot::Mutex<std::collections::HashMap<String, PromptTask>>>,
pub prompt_gen: std::sync::Arc<std::sync::atomic::AtomicU64>,
pub question_gate: std::sync::Arc<ocserve_core::question::QuestionGate>,
pub tui: crate::tui::TuiControl,
pub ui: crate::ui::UiProxy,
pub pty: std::sync::Arc<ocserve_pty::PtyManager>,
pub cors_extra: Vec<String>,
}
#[derive(Default)]
pub struct LlmRegistry {
pub endpoints: std::collections::HashMap<String, (String, String)>,
pub pricing: std::collections::HashMap<(String, String), (f64, f64, f64)>,
pub limits: std::collections::HashMap<(String, String), serde_json::Value>,
pub default_model: (String, String),
pub systems: std::collections::HashMap<String, String>,
pub default_agent: String,
}
pub struct Wires {
pub db: std::path::PathBuf,
pub blobs: std::sync::Arc<ocserve_store::BlobStore>,
pub writer: std::sync::Arc<ocserve_store::Writer>,
pub llm: LlmRegistry,
}
#[derive(Clone)]
pub struct WireEntry {
pub identity: bytes::Bytes,
pub br: bytes::Bytes,
pub gzip: bytes::Bytes,
pub etag: String,
}
pub(crate) fn rebuild_wire(
p: &Payloads,
off: bool,
location: &Value,
) -> HashMap<&'static str, WireEntry> {
let mut m = HashMap::new();
if off {
return m;
}
fn one<T: serde::Serialize>(m: &mut HashMap<&'static str, WireEntry>, k: &'static str, v: &T) {
if let Ok(b) = serde_json::to_vec(v) {
let identity = bytes::Bytes::from(b);
let etag = crate::compress::etag_for(&identity);
let br = bytes::Bytes::from(crate::compress::compress(
&identity,
crate::compress::Encoding::Brotli,
));
let gzip = bytes::Bytes::from(crate::compress::compress(
&identity,
crate::compress::Encoding::Gzip,
));
m.insert(
k,
WireEntry {
identity,
br,
gzip,
etag,
},
);
}
}
{
let mut gc = p.config.clone();
if let Some(obj) = gc.as_object_mut() {
for k in ["agent", "command", "mode", "username"] {
obj.remove(k);
}
}
one(&mut m, "global_config", &gc);
}
one(&mut m, "config", &p.config);
one(&mut m, "agent", &p.agent);
one(&mut m, "command", &p.command);
one(&mut m, "config_providers", &p.config_providers);
one(&mut m, "provider", &p.provider);
one(&mut m, "console", &p.console);
one(&mut m, "capabilities", &p.capabilities);
one(&mut m, "api_model", &crate::v2::models_body(p, location));
one(
&mut m,
"api_provider",
&crate::v2::providers_body(p, location),
);
one(
&mut m,
"api_integration",
&crate::v2::integrations_body(p, location),
);
one(&mut m, "api_agent", &api_agent_body(p, location));
one(&mut m, "api_command", &api_command_body(p, location));
m
}
fn wire_json_h(
st: &AppState,
key: &'static str,
accept_encoding: Option<&str>,
if_none_match: Option<&str>,
) -> Option<axum::response::Response> {
if st.wire_off.load(std::sync::atomic::Ordering::Relaxed) {
return None;
}
let e = st.wire.read().get(key).cloned()?;
let encoding = crate::compress::negotiate(accept_encoding);
if crate::compress::if_none_match_matches(if_none_match, &e.etag) {
return Some(
axum::response::Response::builder()
.status(axum::http::StatusCode::NOT_MODIFIED)
.header(axum::http::header::ETAG, &e.etag)
.header(axum::http::header::VARY, "Accept-Encoding")
.body(axum::body::Body::empty())
.expect("static 304"),
);
}
let mut b = axum::response::Response::builder()
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::ETAG, &e.etag)
.header(axum::http::header::VARY, "Accept-Encoding");
let body = match encoding {
Some(crate::compress::Encoding::Brotli) => {
b = b.header(axum::http::header::CONTENT_ENCODING, "br");
e.br.clone()
}
Some(crate::compress::Encoding::Gzip) => {
b = b.header(axum::http::header::CONTENT_ENCODING, "gzip");
e.gzip.clone()
}
None => e.identity.clone(),
};
Some(
b.body(axum::body::Body::from(body))
.expect("static response"),
)
}
#[cfg(test)]
fn wire_json(st: &AppState, key: &'static str) -> Option<axum::response::Response> {
wire_json_h(st, key, None, None)
}
impl AppState {
pub fn new() -> Arc<Self> {
Self::with_auth(None)
}
pub fn with_auth(auth: Option<(String, String)>) -> Arc<Self> {
Self::with_payloads(auth, Payloads::default())
}
pub fn with_payloads(auth: Option<(String, String)>, p: Payloads) -> Arc<Self> {
static NEW_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let dir = std::env::temp_dir().join(format!(
"ocserve-http-{}-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0),
NEW_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
));
std::fs::create_dir_all(&dir).expect("temp data dir");
let db = ocserve_store::writer::db_path(&dir);
let writer =
std::sync::Arc::new(ocserve_store::Writer::spawn(db.clone()).expect("temp writer"));
let blobs = std::sync::Arc::new(
ocserve_store::BlobStore::new(dir.join("blobs")).expect("temp blobs"),
);
Self::with_wiring(
auth,
p,
Wires {
db,
blobs,
writer,
llm: LlmRegistry::default(),
},
)
}
pub fn with_wiring(auth: Option<(String, String)>, p: Payloads, w: Wires) -> Arc<Self> {
let home = std::env::var("HOME").unwrap_or_default();
let worktree = std::env::current_dir()
.map(|p| p.display().to_string())
.unwrap_or_else(|_| "/".into());
let wire_off = std::env::var("OCSERVE_WIRE_CACHE")
.map(|v| v == "0")
.unwrap_or(false);
let location = json!({
"directory": worktree,
"project": {"id": "global", "directory": "/"},
});
let wire_map = rebuild_wire(&p, wire_off, &location);
let bus = ocserve_core::EventBus::new();
let pty = {
let mgr = ocserve_pty::PtyManager::new(worktree.clone());
let bus_for_pty = bus.clone();
let dir_for_pty = worktree.clone();
mgr.set_event_sink(std::sync::Arc::new(move |event_type, properties| {
bus_for_pty.publish(ocserve_core::event::frame(
&dir_for_pty,
event_type,
properties,
));
}));
mgr
};
Arc::new(Self {
payloads: parking_lot::RwLock::new(p),
wire: parking_lot::RwLock::new(wire_map),
wire_off: std::sync::atomic::AtomicBool::new(wire_off),
reloader: parking_lot::RwLock::new(None),
paths: json!({
"home": home,
"state": format!("{home}/.local/state/opencode"),
"config": format!("{home}/.config/opencode"),
"worktree": worktree,
"directory": worktree,
}),
auth,
requests: std::sync::atomic::AtomicU64::new(0),
bus,
gate: ocserve_core::PermissionGate::new(),
mcp: std::sync::OnceLock::new(),
plugins: std::sync::OnceLock::new(),
prompt_locks: std::sync::Arc::new(parking_lot::Mutex::new(
std::collections::HashMap::new(),
)),
prompt_tasks: std::sync::Arc::new(parking_lot::Mutex::new(
std::collections::HashMap::new(),
)),
prompt_gen: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
question_gate: ocserve_core::question::QuestionGate::new(),
tui: crate::tui::TuiControl::default(),
ui: crate::ui::UiProxy::from_env(),
pty,
cors_extra: read_cors_env(),
db: w.db,
blobs: w.blobs,
writer: w.writer,
llm: parking_lot::RwLock::new(w.llm),
watch: parking_lot::RwLock::new(watch::WatchState::default()),
reconcile_lock: tokio::sync::Mutex::new(()),
max_rounds: std::env::var("OCSERVE_PROMPT_MAX_ROUNDS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0),
cost_ceiling: std::env::var("OCSERVE_PROMPT_MAX_COST_USD")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0.0),
})
}
}
async fn health() -> impl IntoResponse {
Json(json!({"healthy": true, "version": FREEZE_VERSION}))
}
async fn get_doc() -> impl IntoResponse {
(
[(axum::http::header::CONTENT_TYPE, "application/json")],
crate::doc::SPEC_JSON,
)
}
async fn get_config(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "config", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().config.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
async fn get_agent(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "agent", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().agent.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
async fn get_command(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "command", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().command.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
async fn get_config_providers(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "config_providers", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().config_providers.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
async fn get_provider(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "provider", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().provider.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
async fn get_console(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "console", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().console.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
async fn get_capabilities(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "capabilities", ae, inm).unwrap_or_else(|| {
let v = st.payloads.read().capabilities.clone();
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
pub(crate) fn api_location(st: &AppState) -> Value {
let dir = st.paths["directory"].as_str().unwrap_or("/");
json!({
"directory": dir,
"project": {"id": "global", "directory": "/"},
})
}
async fn get_api_agent(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
crate::wire_json_h(&st, "api_agent", ae, inm).unwrap_or_else(|| {
let payloads = st.payloads.read();
Json(api_agent_body(&payloads, &api_location(&st))).into_response()
})
}
pub(crate) fn api_agent_body(payloads: &Payloads, location: &Value) -> Value {
json!({"location": location, "data": api_agent_data(payloads)})
}
fn api_agent_data(payloads: &Payloads) -> Vec<Value> {
let data: Vec<Value> = payloads
.api_agent
.iter()
.map(|a| {
let perms: Vec<Value> = a["permission"]
.as_array()
.map(|rules| {
rules
.iter()
.map(|r| {
json!({
"action": r["permission"],
"resource": r["pattern"],
"effect": r["action"],
})
})
.collect()
})
.unwrap_or_default();
let mut out = json!({
"id": a["name"],
"hidden": a.get("hidden").and_then(|v| v.as_bool()).unwrap_or(false),
"mode": a["mode"],
"permissions": perms,
"request": {"headers": {}, "body": {}},
});
if let Some(d) = a.get("description") {
out["description"] = d.clone();
}
if let Some(p) = a.get("prompt") {
out["system"] = p.clone();
} else if a["name"] == "build" {
out["system"] = json!(BUILD_SYSTEM_BLURB);
}
out
})
.collect();
data
}
const BUILD_SYSTEM_BLURB: &str = "You are an AI coding agent. Help the user accomplish software \
engineering tasks by inspecting the workspace, making targeted changes, and using tools according \
to the configured permissions.";
async fn get_api_command(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
crate::wire_json_h(&st, "api_command", ae, inm).unwrap_or_else(|| {
let payloads = st.payloads.read();
Json(api_command_body(&payloads, &api_location(&st))).into_response()
})
}
pub(crate) fn api_command_body(payloads: &Payloads, location: &Value) -> Value {
let data: Vec<Value> = payloads
.command
.iter()
.map(|c| {
json!({
"name": c["name"],
"description": c.get("description").and_then(|v| v.as_str()).unwrap_or(""),
"template": c.get("template").and_then(|v| v.as_str()).unwrap_or(""),
})
})
.collect();
json!({"location": location, "data": data})
}
async fn get_experimental_tool(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, ApiError> {
let has = |k: &str| q.get(k).map(|s| !s.is_empty()).unwrap_or(false);
if !has("provider") || !has("model") {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "Invalid query: provider, model required".into(),
});
}
let mut schemas = ocserve_tools::schemas();
if let Some(hub) = st.mcp.get() {
let mcp_tools =
tokio::time::timeout(std::time::Duration::from_secs(10), hub.tool_schemas())
.await
.unwrap_or_else(|_| {
tracing::warn!("mcp tool schema fetch timed out (/experimental/tool)");
Vec::new()
});
schemas.extend(mcp_tools);
}
let items: Vec<Value> = schemas
.iter()
.filter_map(|t| {
let f = t.get("function")?;
Some(json!({
"id": f.get("name")?,
"description": f.get("description")?,
"parameters": f.get("parameters")?,
}))
})
.collect();
Ok(Json(Value::Array(items)))
}
async fn get_experimental_tool_ids() -> Json<Value> {
let ids: Vec<Value> = ocserve_tools::schemas()
.iter()
.filter_map(|t| t.pointer("/function/name").and_then(|n| n.as_str()))
.map(|n| json!(n))
.collect();
Json(Value::Array(ids))
}
async fn experimental_resource() -> Json<Value> {
Json(json!({}))
}
async fn experimental_workspace() -> Json<Value> {
Json(json!([]))
}
async fn experimental_workspace_status() -> Json<Value> {
Json(json!([]))
}
async fn formatter_list() -> Json<Value> {
Json(json!([]))
}
async fn lsp_list() -> Json<Value> {
Json(json!([]))
}
async fn project_directories() -> Json<Value> {
Json(json!([]))
}
async fn get_mcp(State(st): State<Arc<AppState>>) -> Json<Value> {
let hub = st
.mcp
.get()
.cloned()
.unwrap_or_else(|| std::sync::Arc::new(ocserve_mcp::McpHub::default()));
Json(hub.statuses())
}
async fn post_mcp(
State(st): State<Arc<AppState>>,
Json(body): Json<Value>,
) -> Result<Json<Value>, HttpError> {
let name = body
.get("name")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.ok_or_else(|| mcp_payload_err("name"))?;
let config = body
.get("config")
.filter(|c| c.is_object())
.ok_or_else(|| mcp_payload_err("config"))?;
let hub = st.mcp.get().cloned().ok_or_else(|| {
HttpError::Api(ApiError {
status: StatusCode::SERVICE_UNAVAILABLE,
name: "McpUnavailable",
message: "MCP hub not initialized".into(),
})
})?;
let mut one = serde_json::Map::new();
one.insert(name.to_string(), config.clone());
let cfg = ocserve_mcp::parse_config(&Value::Object(one))
.into_iter()
.next()
.ok_or_else(|| mcp_payload_err("config"))?;
hub.upsert_cfg(cfg);
if let Err(e) = hub.connect(name).await {
tracing::info!("mcp add: connect {name} failed (recorded): {e:#}");
}
Ok(Json(hub.statuses()))
}
fn mcp_payload_err(field: &str) -> HttpError {
HttpError::Payload {
message: format!("Missing key\n at [\"{field}\"]"),
}
}
pub(crate) fn decode_v1<T>(r: Result<T, ocserve_core::wire::WireError>) -> Result<T, HttpError> {
r.map_err(|e| HttpError::Payload { message: e.message })
}
pub(crate) fn decode_v2<T>(r: Result<T, ocserve_core::wire::WireError>) -> Result<T, HttpError> {
r.map_err(|e| HttpError::TaggedData {
status: StatusCode::BAD_REQUEST,
tag: "InvalidRequestError",
fields: json!({"message": e.message, "kind": "Payload"}),
})
}
async fn get_experimental_sessions(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Json<Value> {
let mut v = run_blocking(&st.db, ocserve_store::load_sessions_wire)
.await
.unwrap_or_else(|e| {
tracing::error!("session list read failed: {e:#}");
Vec::new()
});
if let Some(search) = q.get("search").filter(|s| !s.is_empty()) {
let needle = search.to_lowercase();
v.retain(|s| {
s["title"]
.as_str()
.unwrap_or_default()
.to_lowercase()
.contains(&needle)
});
}
if let Some(limit) = q.get("limit").and_then(|l| l.parse::<usize>().ok()) {
v.truncate(limit);
}
for row in &mut v {
row["project"] = json!({"id": "global", "worktree": "/"});
}
Json(Value::Array(v))
}
async fn post_permission_reply(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(body): Json<Value>,
) -> Result<Json<Value>, HttpError> {
let reply = validate_permission_reply(&body)?;
if !st.gate.reply(&id, reply) {
return Err(HttpError::TaggedData {
status: StatusCode::NOT_FOUND,
tag: "PermissionNotFoundError",
fields: json!({"requestID": id, "message": format!("Permission request not found: {id}")}),
});
}
Ok(Json(json!({})))
}
pub(crate) fn validate_permission_reply(body: &Value) -> Result<&'static str, HttpError> {
validate_permission_reply_key(body, "reply")
}
pub(crate) fn validate_permission_reply_key(
body: &Value,
key: &str,
) -> Result<&'static str, HttpError> {
let Some(v) = body.get(key) else {
return Err(HttpError::Payload {
message: format!("Missing key\n at [\"{key}\"]"),
});
};
match v.as_str() {
Some("once") => Ok("once"),
Some("always") => Ok("always"),
Some("reject") => Ok("reject"),
_ => Err(HttpError::Payload {
message: format!(
"Expected \"once\" | \"always\" | \"reject\", got {}\n at [\"{key}\"]",
v
),
}),
}
}
async fn get_permissions(State(st): State<Arc<AppState>>) -> Json<Value> {
Json(serde_json::Value::Array(st.gate.list()))
}
async fn vcs_info(State(st): State<Arc<AppState>>) -> impl IntoResponse {
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let info = run_blocking(&st.db, move |_db| Ok::<_, anyhow::Error>(vcs::info(&dir)))
.await
.unwrap_or_else(|_| json!({"branch": null, "default_branch": null}));
Json(info)
}
async fn vcs_status(State(st): State<Arc<AppState>>) -> Response {
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
match run_blocking(&st.db, move |_db| Ok::<_, anyhow::Error>(vcs::status(&dir))).await {
Ok(v) => axum::Json(v).into_response(),
Err(_) => axum::Json(Value::Array(vec![])).into_response(),
}
}
async fn vcs_diff(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
headers: axum::http::HeaderMap,
) -> Response {
let mode = q.get("mode").map(String::as_str).unwrap_or("");
if mode != "git" && mode != "branch" {
return HttpError::Query {
message: format!("Expected \"git\" | \"branch\", got \"{mode}\"\n at [\"mode\"]"),
}
.into_response();
}
let context: Option<u32> = q.get("context").and_then(|v| v.parse().ok());
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let mode_owned = mode.to_string();
let v = run_blocking(&st.db, move |_db| {
Ok::<_, anyhow::Error>(vcs::diff(&dir, &mode_owned, context))
})
.await
.unwrap_or_else(|_| Value::Array(vec![]));
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
}
async fn vcs_diff_raw(State(st): State<Arc<AppState>>) -> Response {
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let (text, truncated) = run_blocking(&st.db, move |_db| {
Ok::<_, anyhow::Error>(vcs::diff_raw_bounded(&dir))
})
.await
.unwrap_or_else(|_| (String::new(), false));
Response::builder()
.header(
axum::http::header::CONTENT_TYPE,
"text/plain; charset=utf-8",
)
.header(axum::http::header::VARY, "Accept-Encoding")
.header("x-ocserve-truncated", if truncated { "1" } else { "0" })
.body(axum::body::Body::from(text))
.expect("static response")
}
async fn file_status() -> impl IntoResponse {
axum::Json(Value::Array(vec![]))
}
async fn find_symbol() -> impl IntoResponse {
axum::Json(Value::Array(vec![]))
}
async fn global_config_get(
State(st): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
) -> Response {
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
wire_json_h(&st, "global_config", ae, inm).unwrap_or_else(|| {
let mut v = st.payloads.read().config.clone();
if let Some(obj) = v.as_object_mut() {
for k in ["agent", "command", "mode", "username"] {
obj.remove(k);
}
}
let body = serde_json::to_vec(&v).unwrap_or_default();
let etag = crate::compress::etag_for(&body);
crate::compress::encode_response(
StatusCode::OK,
"application/json",
bytes::Bytes::from(body),
Some(&etag),
ae,
inm,
&[],
)
})
}
pub const PAGE_MEMO_ENTRY_CAP: usize = 4 * 1024 * 1024;
pub const PAGE_MEMO_MAX_ENTRIES: usize = 2048;
pub const PAGE_MEMO_MAX_BYTES: usize = 64 * 1024 * 1024;
struct PageEntry {
body: bytes::Bytes,
next: Option<String>,
}
struct PageMemoState {
epoch: u64,
entries: std::collections::VecDeque<(String, PageEntry)>,
bytes: usize,
}
impl PageMemoState {
fn new() -> Self {
Self {
epoch: 0,
entries: std::collections::VecDeque::new(),
bytes: 0,
}
}
fn get(&mut self, key: &str, epoch: u64) -> Option<PageEntry> {
if self.epoch != epoch {
self.entries.clear();
self.bytes = 0;
self.epoch = epoch;
}
self.entries
.iter()
.find(|(k, _)| k == key)
.map(|(_, e)| PageEntry {
body: e.body.clone(),
next: e.next.clone(),
})
}
fn put(&mut self, key: String, entry: PageEntry, epoch: u64) {
if entry.body.len() > PAGE_MEMO_ENTRY_CAP.min(PAGE_MEMO_MAX_BYTES) {
return;
}
if self.epoch != epoch {
self.entries.clear();
self.bytes = 0;
self.epoch = epoch;
}
if let Some(pos) = self.entries.iter().position(|(k, _)| k == &key) {
let (_, old) = self.entries.remove(pos).expect("position found");
self.bytes -= old.body.len();
}
let budget = std::env::var("OCSERVE_PAGE_MEMO_MB")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.map(|mb| mb * 1024 * 1024)
.unwrap_or(PAGE_MEMO_MAX_BYTES)
.min(PAGE_MEMO_MAX_BYTES);
self.bytes += entry.body.len();
self.entries.push_back((key, entry));
while self.entries.len() > PAGE_MEMO_MAX_ENTRIES
|| (self.bytes > budget && self.entries.len() > 1)
{
if let Some((_, old)) = self.entries.pop_front() {
self.bytes -= old.body.len();
} else {
break;
}
}
}
fn len(&self) -> usize {
self.entries.len()
}
}
static PAGE_MEMO: std::sync::OnceLock<parking_lot::Mutex<PageMemoState>> =
std::sync::OnceLock::new();
static PAGE_MEMO_HITS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static PAGE_MEMO_MISSES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
fn page_memo() -> &'static parking_lot::Mutex<PageMemoState> {
PAGE_MEMO.get_or_init(|| parking_lot::Mutex::new(PageMemoState::new()))
}
pub fn page_memo_enabled() -> bool {
std::env::var("OCSERVE_PAGE_MEMO")
.map(|v| v != "0")
.unwrap_or(true)
}
pub fn page_memo_stats() -> (u64, u64, usize) {
(
PAGE_MEMO_HITS.load(std::sync::atomic::Ordering::Relaxed),
PAGE_MEMO_MISSES.load(std::sync::atomic::Ordering::Relaxed),
page_memo().lock().len(),
)
}
fn page_response(
headers: &axum::http::HeaderMap,
uri: &axum::http::Uri,
page_n: &Option<u64>,
page_next: &Option<String>,
body: axum::body::Body,
) -> Result<axum::response::Response, ApiError> {
let mut builder = axum::response::Response::builder()
.header(axum::http::header::CONTENT_TYPE, "application/json");
if let (Some(n), Some(cur)) = (page_n, page_next) {
let scheme = headers
.get("x-forwarded-proto")
.and_then(|v| v.to_str().ok())
.unwrap_or("http");
let host = headers
.get("host")
.and_then(|v| v.to_str().ok())
.unwrap_or("localhost");
let link = format!(
"<{scheme}://{host}{}?limit={n}&before={cur}>; rel=\"next\"",
uri.path()
);
builder = builder
.header("x-next-cursor", cur.as_str())
.header("link", link)
.header("access-control-expose-headers", "Link, X-Next-Cursor");
}
builder.body(body).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("response build: {e}"),
})
}
async fn get_sessions(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
headers: axum::http::HeaderMap,
) -> axum::response::Response {
let query = ocserve_store::SessionListQuery {
limit: q
.get("limit")
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0),
roots: q.get("roots").map(|v| v == "true").unwrap_or(false),
search: q.get("search").filter(|s| !s.is_empty()).cloned(),
start: q.get("start").and_then(|v| v.parse::<i64>().ok()),
directory: q.get("directory").filter(|s| !s.is_empty()).cloned(),
scope_project: q.get("scope").map(|v| v == "project").unwrap_or(false),
};
let ae = headers
.get(axum::http::header::ACCEPT_ENCODING)
.and_then(|v| v.to_str().ok());
let inm = headers
.get(axum::http::header::IF_NONE_MATCH)
.and_then(|v| v.to_str().ok());
let qc = query.clone();
match run_blocking(&st.db, move |db| {
ocserve_store::load_sessions_wire_bytes_filtered(db, &qc)
})
.await
{
Ok(b) => {
let etag = crate::compress::etag_for(&b);
crate::compress::encode_response(
axum::http::StatusCode::OK,
"application/json",
(*b).clone(),
Some(&etag),
ae,
inm,
&[],
)
}
Err(e) => {
tracing::error!("session list read failed: {e:#}");
Json(Vec::<serde_json::Value>::new()).into_response()
}
}
}
async fn get_session(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<impl IntoResponse, ApiError> {
let wire_id = id.clone();
run_blocking(&st.db, move |db| {
ocserve_store::load_session_wire(db, &wire_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.map(Json)
.ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))
}
async fn session_status(State(st): State<Arc<AppState>>) -> impl IntoResponse {
let mut out = serde_json::Map::new();
let mut busy = |sid: &str| {
out.insert(sid.to_string(), serde_json::json!({"type": "busy"}));
};
for (sid, arc) in st.prompt_locks.lock().iter() {
if arc.try_lock().is_err() {
busy(sid);
}
}
for (sid, (_gen, handle)) in st.prompt_tasks.lock().iter() {
if !handle.is_finished() {
busy(sid);
}
}
Json(serde_json::Value::Object(out))
}
async fn get_path(State(st): State<Arc<AppState>>) -> impl IntoResponse {
Json(st.paths.clone())
}
async fn get_projects(State(st): State<Arc<AppState>>) -> impl IntoResponse {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
let mut dirs = ocserve_store::distinct_session_directories(&st.db).unwrap_or_default();
let worktree = st.paths["worktree"].as_str().unwrap_or("/").to_string();
if !dirs.iter().any(|d| d == &worktree) {
dirs.insert(0, worktree);
}
let project = |id: &str, wt: &str| {
json!({
"id": id,
"worktree": wt,
"time": {"created": now_ms, "updated": now_ms, "initialized": now_ms},
"sandboxes": [],
})
};
let mut out = vec![project("global", "/")];
for d in dirs {
out.push(project(&d, &d));
}
Json(Value::Array(out))
}
async fn get_project_current(State(st): State<Arc<AppState>>) -> impl IntoResponse {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
let worktree = st.paths["worktree"].as_str().unwrap_or("/");
Json(json!({
"id": "global",
"worktree": worktree,
"time": {"created": now_ms, "updated": now_ms, "initialized": now_ms},
"sandboxes": [],
}))
}
async fn metrics(State(_st): State<Arc<AppState>>) -> impl IntoResponse {
ocserve_metrics::sample_rss();
ocserve_metrics::gauge(
"ocserve_splice_rows_total",
ocserve_store::splice_rows() as i64,
);
ocserve_metrics::gauge(
"ocserve_splice_fallbacks_total",
ocserve_store::splice_fallbacks() as i64,
);
let (search_hits, search_misses, list_hits, list_misses, _pad, search_len, _epoch) =
ocserve_store::memo_stats();
ocserve_metrics::gauge("ocserve_memo_search_hits", search_hits as i64);
ocserve_metrics::gauge("ocserve_memo_search_misses", search_misses as i64);
ocserve_metrics::gauge("ocserve_memo_list_hits", list_hits as i64);
ocserve_metrics::gauge("ocserve_memo_list_misses", list_misses as i64);
ocserve_metrics::gauge("ocserve_memo_search_entries", search_len as i64);
let body = ocserve_metrics::render();
(
[(
axum::http::header::CONTENT_TYPE,
"text/plain; version=0.0.4",
)],
body,
)
}
fn slug_for(id: &str) -> String {
const A: &[&str] = &[
"stellar", "amber", "quiet", "cobalt", "rapid", "lunar", "vivid", "crisp", "amber", "ivory",
];
const B: &[&str] = &[
"orchid", "falcon", "meadow", "canyon", "ember", "drift", "quartz", "harbor", "summit",
"ripple",
];
let h = id
.bytes()
.fold(0u64, |acc, b| acc.wrapping_mul(31).wrapping_add(b as u64));
format!(
"{}-{}",
A[(h % A.len() as u64) as usize],
B[((h >> 16) % B.len() as u64) as usize]
)
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
fn iso8601_z(ms: i64) -> String {
let ms = ms.rem_euclid(86_400_000_000_000);
let (days, rem) = (ms.div_euclid(86_400_000), ms.rem_euclid(86_400_000));
let (h, rem) = (rem.div_euclid(3_600_000), rem.rem_euclid(3_600_000));
let (mi, rem) = (rem.div_euclid(60_000), rem.rem_euclid(60_000));
let (sec, milli) = (rem.div_euclid(1_000), rem.rem_euclid(1_000));
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if m <= 2 { y + 1 } else { y };
format!("{y:04}-{m:02}-{d:02}T{h:02}:{mi:02}:{sec:02}.{milli:03}Z")
}
pub fn js_to_date_string(ms: i64) -> String {
const WDAY: [&str; 7] = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"];
const MON: [&str; 12] = [
"Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
];
let days = ms.div_euclid(86_400_000);
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 } as usize;
let y = if m <= 2 { y + 1 } else { y };
let wd = (days + 4).rem_euclid(7) as usize;
format!("{} {} {:02} {}", WDAY[wd], MON[m - 1], d, y)
}
fn default_session_title(now_ms: i64) -> String {
format!("New session - {}", iso8601_z(now_ms))
}
pub fn forked_title(title: &str) -> String {
const PAT: &str = " (fork #";
if let Some(idx) = title.rfind(PAT)
&& idx > 0
&& let Some(num) = title[idx + PAT.len()..].strip_suffix(')')
&& !num.is_empty()
&& num.bytes().all(|b| b.is_ascii_digit())
&& let Ok(n) = num.parse::<u64>()
{
return format!("{} (fork #{})", &title[..idx], n + 1);
}
format!("{title} (fork #1)")
}
pub fn start_plugin_event_pump(st: &Arc<AppState>) {
let mut rx = st.bus.subscribe();
let st = st.clone();
tokio::spawn(async move {
loop {
match rx.recv().await {
Ok(frame_str) => {
let Some(plug) = st.plugins.get() else {
continue; };
let Ok(v) = serde_json::from_str::<Value>(&frame_str) else {
continue;
};
let Some(payload) = v.get("payload").cloned() else {
continue;
};
let mut guard = plug.lock().await;
if let Err(e) = guard.emit_event(payload).await {
tracing::debug!("plugin event pump: {e:#}");
}
}
Err(ocserve_core::event::RecvError::Lagged(n)) => {
tracing::warn!("plugin event pump lagged {n} frames (skipped)");
}
Err(ocserve_core::event::RecvError::Closed) => break,
}
}
});
}
async fn post_session(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
Json(body): Json<Value>,
) -> Result<impl IntoResponse, ApiError> {
let resolved =
resolve_create_dir_path(&st, &body, q.get("directory").map(String::as_str)).await;
Ok(Json(create_session_record(&st, &body, resolved)?))
}
async fn resolve_create_dir_path(
st: &Arc<AppState>,
body: &Value,
query_dir: Option<&str>,
) -> (String, String) {
let requested = resolve_create_dir(st, body, query_dir);
let st = st.clone();
tokio::task::spawn_blocking(move || freeze_dir_path(&requested))
.await
.unwrap_or_else(|_| {
let d = resolve_create_dir(&st, body, query_dir);
(d.clone(), d.trim_start_matches('/').to_string())
})
}
fn resolve_create_dir(st: &Arc<AppState>, body: &Value, query_dir: Option<&str>) -> String {
query_dir
.filter(|s| !s.is_empty())
.map(String::from)
.or_else(|| {
body.pointer("/location/directory")
.and_then(|d| d.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
})
.unwrap_or_else(|| st.paths["worktree"].as_str().unwrap_or("/").to_string())
}
fn freeze_dir_path(dir: &str) -> (String, String) {
let dir = dir.trim_end_matches('/');
if dir.is_empty() {
return ("/".to_string(), String::new());
}
match git_toplevel(dir) {
Some(root) => {
let path = dir
.strip_prefix(&root)
.unwrap_or("")
.trim_start_matches('/')
.to_string();
(root, path)
}
None => {
let path = dir.trim_start_matches('/').to_string();
(dir.to_string(), path)
}
}
}
fn git_toplevel(dir: &str) -> Option<String> {
let out = std::process::Command::new("git")
.arg("-C")
.arg(dir)
.args(["rev-parse", "--show-toplevel"])
.output()
.ok()?;
if !out.status.success() {
return None;
}
let root = String::from_utf8_lossy(&out.stdout).trim().to_string();
(!root.is_empty()).then_some(root)
}
pub(crate) fn create_session_record(
st: &Arc<AppState>,
body: &Value,
(worktree, rel_path): (String, String),
) -> Result<Value, ApiError> {
let id = ocserve_core::ids::ses_id();
let now = now_ms();
let title = body
.get("title")
.and_then(|t| t.as_str())
.unwrap_or("")
.to_string();
let title = if title.trim().is_empty() {
default_session_title(now)
} else {
title
};
let agent = body
.get("agent")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.unwrap_or("build");
let model = body.get("model").cloned().unwrap_or(Value::Null);
let mut info = json!({
"id": id,
"projectID": "global",
"directory": worktree,
"path": rel_path,
"slug": slug_for(&id),
"title": title,
"version": FREEZE_VERSION,
"time": {"created": now, "updated": now},
"cost": 0,
"tokens": {"input": 0, "output": 0, "reasoning": 0,
"cache": {"read": 0, "write": 0}},
});
if body.get("agent").is_some() {
info["agent"] = json!(agent);
}
if model.is_object() {
info["model"] = model;
}
ocserve_store::insert_session(&st.writer, &info).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
st.bus.publish(ocserve_core::event::frame(
&worktree,
"session.created",
json!({"sessionID": id, "info": info}),
));
Ok(info)
}
async fn post_fork(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
body: axum::body::Bytes,
) -> Result<Json<Value>, HttpError> {
let upto: Option<String> = {
let text = String::from_utf8_lossy(&body);
if text.trim().is_empty() {
None } else {
let v: Value = serde_json::from_str(&text).map_err(|_| tagged_bad_request())?;
if !v.is_object() {
return Err(tagged_bad_request()); }
match v.get("messageID") {
None => None,
Some(Value::String(s)) => Some(s.clone()),
Some(_) => return Err(tagged_bad_request()), }
}
};
let src_id = id.clone();
let src = run_blocking(&st.db, move |db| {
ocserve_store::load_session_wire(db, &src_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))?;
let new_id = ocserve_core::ids::ses_id();
let now = now_ms();
let worktree = st.paths["worktree"].as_str().unwrap_or("/").to_string();
let info = json!({
"id": new_id,
"projectID": "global",
"directory": worktree,
"path": worktree.trim_start_matches('/'),
"slug": slug_for(&new_id),
"title": forked_title(src["title"].as_str().unwrap_or("")),
"version": FREEZE_VERSION,
"time": {"created": now, "updated": now},
"cost": 0,
"tokens": {"input": 0, "output": 0, "reasoning": 0,
"cache": {"read": 0, "write": 0}},
});
let writer = st.writer.clone();
let blobs = st.blobs.clone();
let db = st.db.clone();
let fork_info = info.clone();
let fork_id = id.clone();
let fork_upto = upto.clone();
let stats = run_blocking(&db, move |db| {
let env = ocserve_store::fork::ForkEnv {
writer: &writer,
blobs: Some(&*blobs),
db,
};
ocserve_store::fork_session(
&env,
&fork_info,
&fork_id,
fork_upto.as_deref(),
ocserve_core::ids::msg_id,
ocserve_core::ids::prt_id,
)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
ocserve_metrics::counter("ocserve_session_fork_total", 1);
tracing::info!(
fork = %new_id,
source = %id,
upto = ?upto,
messages = stats.messages,
parts = stats.parts,
"session forked"
);
st.bus.publish(ocserve_core::event::frame(
st.paths["directory"].as_str().unwrap_or("/"),
"session.created",
json!({"sessionID": new_id, "info": info}),
));
Ok(Json(info))
}
async fn get_messages(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
uri: axum::http::Uri,
headers: axum::http::HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
let limit: Option<u64> = match q.get("limit") {
None => None,
Some(raw) if raw.is_empty() => None, Some(raw) => Some(parse_limit(raw).map_err(|message| HttpError::Query { message })?),
};
let before: Option<(String, i64)> = match (q.get("before").map(String::as_str), limit) {
(Some(_), None) => return Err(tagged_bad_request()),
(Some(b), Some(_)) => {
Some(ocserve_store::decode_cursor(b).map_err(|_| tagged_bad_request())?)
}
(None, _) => None,
};
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")).into());
}
let mut page_n: Option<u64> = None;
let mut page_next: Option<String> = None;
let walk: ocserve_store::MessageWalk = match limit.filter(|n| *n > 0) {
Some(n) => {
let page_id = id.clone();
let before_owned = before.clone();
let (rows, _more, next) = run_blocking(&st.db, move |db| {
ocserve_store::page_messages(
db,
&page_id,
n,
before_owned.as_ref().map(|(b, t)| (b.as_str(), *t)),
)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
page_n = Some(n);
page_next = next;
ocserve_store::MessageWalk::Window(rows)
}
None => ocserve_store::MessageWalk::Seq { limit: None },
};
if page_n.is_some() && page_memo_enabled() {
let memo_key = format!("{}\u{1}{id}\u{1}{limit:?}\u{1}{before:?}", st.db.display());
let e0 = ocserve_store::write_epoch();
{
let mut m = page_memo().lock();
if let Some(e) = m.get(&memo_key, e0) {
drop(m);
PAGE_MEMO_HITS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
return page_response(
&headers,
&uri,
&page_n,
&page_next,
axum::body::Body::from(e.body),
)
.map_err(Into::into);
}
}
PAGE_MEMO_MISSES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let walk_asm = walk.clone();
let db_asm = st.db.clone();
let sid_asm = id.clone();
let asm: anyhow::Result<std::result::Result<Vec<u8>, ()>> =
run_blocking(&db_asm, move |db| {
let mut buf: Vec<u8> = Vec::new();
let mut total = 0usize;
let mut first = true;
let r = ocserve_store::for_each_message_json(db, &sid_asm, walk_asm, |chunk| {
let framed = if first {
first = false;
format!("[{chunk}")
} else {
format!(",{chunk}")
};
if total + framed.len() > PAGE_MEMO_ENTRY_CAP {
return Err(anyhow::anyhow!("page memo entry cap"));
}
total += framed.len();
buf.extend_from_slice(framed.as_bytes());
Ok(())
});
match r {
Ok(()) => {
if first {
buf.extend_from_slice(b"[]");
} else {
buf.push(b']');
}
Ok(Ok(buf))
}
Err(e) if e.to_string().contains("entry cap") => Ok(Err(())),
Err(e) => Err(e),
}
})
.await;
if let Ok(Ok(buf)) = asm {
let bytes = bytes::Bytes::from(buf);
let entry = PageEntry {
body: bytes.clone(),
next: page_next.clone(),
};
page_memo().lock().put(memo_key, entry, e0);
return page_response(
&headers,
&uri,
&page_n,
&page_next,
axum::body::Body::from(bytes),
)
.map_err(Into::into);
}
}
let (tx, rx) = tokio::sync::mpsc::channel::<String>(4);
let db = st.db.clone();
let sid = id.clone();
tokio::task::spawn_blocking(move || {
let mut first = true;
let r = ocserve_store::for_each_message_json(&db, &sid, walk, |chunk| {
let framed = if first {
first = false;
format!("[{chunk}")
} else {
format!(",{chunk}")
};
tx.blocking_send(framed)
.map_err(|_| anyhow::anyhow!("client disconnected"))
});
if let Err(e) = r {
tracing::error!("message stream aborted session={sid}: {e:#}");
}
let _ = tx.blocking_send(if first {
"[]".to_string()
} else {
"]".to_string()
});
});
let body_stream = futures_util::stream::unfold(rx, |mut rx| async move {
rx.recv().await.map(|chunk| {
(
Ok::<axum::body::Bytes, std::convert::Infallible>(axum::body::Bytes::from(chunk)),
rx,
)
})
});
page_response(
&headers,
&uri,
&page_n,
&page_next,
axum::body::Body::from_stream(body_stream),
)
.map_err(Into::into)
}
async fn patch_config(
State(st): State<Arc<AppState>>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let path = config_write_path();
let mut base: Value = if path.exists() {
let raw = std::fs::read_to_string(&path).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("read {}: {e}", path.display()),
})?;
serde_json::from_str(&raw).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("parse {}: {e}", path.display()),
})?
} else {
json!({})
};
deep_merge(&mut base, payload.clone());
let dir = path.parent().ok_or_else(|| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: "config path has no parent".into(),
})?;
std::fs::create_dir_all(dir).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("mkdir {}: {e}", dir.display()),
})?;
let tmp = dir.join(format!(".opencode.json.tmp-{}", std::process::id()));
let data = serde_json::to_vec_pretty(&base).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("serialize: {e}"),
})?;
std::fs::write(&tmp, &data).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("write {}: {e}", tmp.display()),
})?;
if path.exists() {
let _ = std::fs::copy(&path, dir.join(".opencode.json.bak"));
}
std::fs::rename(&tmp, &path).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("rename {}: {e}", path.display()),
})?;
if let Err(e) = watch::reconcile(&st).await {
tracing::warn!("config reload failed (restart to apply): {e:#}");
}
tracing::info!("config patched: {}", path.display());
Ok(Json(payload))
}
async fn post_summarize(
State(st): State<Arc<AppState>>,
axum::extract::Path(sid): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let exists_id = sid.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {sid}")));
}
let provider = payload
.get("providerID")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty());
let model_id = payload
.get("modelID")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty());
let (Some(provider), Some(model_id)) = (provider, model_id) else {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "providerID and modelID are required".into(),
});
};
let release = lock_session(&st, &sid).await?;
let Ok(_guard) = release.arc().try_lock() else {
return Err(session_busy(&sid));
};
let history = run_blocking(&st.db, {
let sid = sid.clone();
move |db| ocserve_store::load_messages(db, &sid, None) })
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
let agent = history
.iter()
.rev()
.find(|(info, _)| info["role"] == "user")
.and_then(|(info, _)| info["agent"].as_str())
.unwrap_or("build")
.to_string();
let prompt_payload = ocserve_core::wire::PromptRequest {
model: Some(ocserve_core::wire::ModelRef {
provider_id: provider.to_string(),
model_id: model_id.to_string(),
}),
agent: Some(agent.clone()),
..Default::default()
};
let ctx = build_prompt_context(&st, &prompt_payload, &sid)?;
let writer = st.writer.clone();
ocserve_core::compaction::persist_anchor(&ctx, &writer, &sid, &agent, false, false)
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
ocserve_core::prompt::run_prompt_with(
&ctx,
&writer,
&sid,
&prompt_payload,
ocserve_core::prompt::RunOpts {
persist_user: false,
skip_history: false,
prelude: None,
tools_enabled: false,
..Default::default()
},
)
.await
.map_err(prompt_err)?;
st.bus.publish(ocserve_core::event::frame(
st.paths["directory"].as_str().unwrap_or("/"),
"session.compacted",
json!({"sessionID": sid}),
));
Ok(Json(json!(true)))
}
pub(crate) async fn run_compact(
st: &Arc<AppState>,
ctx: &ocserve_core::prompt::PromptContext,
writer: &ocserve_store::Writer,
sid: &str,
) -> anyhow::Result<()> {
let agent = ctx.agent.clone();
let prompt_payload = ocserve_core::wire::PromptRequest {
model: Some(ocserve_core::wire::ModelRef {
provider_id: ctx.provider_id.clone(),
model_id: ctx.model_id.clone(),
}),
agent: Some(agent.clone()),
..Default::default()
};
ocserve_core::compaction::persist_anchor(ctx, writer, sid, &agent, false, false).await?;
ocserve_core::prompt::run_prompt_with(
ctx,
writer,
sid,
&prompt_payload,
ocserve_core::prompt::RunOpts {
persist_user: false,
skip_history: false,
prelude: None,
tools_enabled: false,
..Default::default()
},
)
.await?;
st.bus.publish(ocserve_core::event::frame(
st.paths["directory"].as_str().unwrap_or("/"),
"session.compacted",
json!({"sessionID": sid}),
));
Ok(())
}
fn mcp_not_found(name: &str) -> HttpError {
HttpError::TaggedData {
status: StatusCode::NOT_FOUND,
tag: "McpServerNotFoundError",
fields: json!({"name": name, "message": format!("MCP server not found: {name}")}),
}
}
async fn mcp_action(
State(st): State<Arc<AppState>>,
axum::extract::Path((name, action)): axum::extract::Path<(String, String)>,
) -> Result<Json<Value>, HttpError> {
let hub = st.mcp.get().cloned().ok_or_else(|| HttpError::TaggedData {
status: StatusCode::SERVICE_UNAVAILABLE,
tag: "McpUnavailable",
fields: json!({"message": "MCP hub not initialized"}),
})?;
let res = match action.as_str() {
"connect" => hub.connect(&name).await,
"disconnect" => hub.disconnect(&name).await,
_ => Err(anyhow::anyhow!("unknown MCP action: {action}")),
};
res.map_err(|e| {
if e.to_string().contains("not found") {
mcp_not_found(&name)
} else {
HttpError::Api(ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})
}
})?;
Ok(Json(json!(true)))
}
async fn mcp_auth_remove(
State(st): State<Arc<AppState>>,
axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<Json<Value>, HttpError> {
let known = st
.mcp
.get()
.map(|h| {
let m = h.statuses();
m.as_object()
.map(|o| o.contains_key(&name))
.unwrap_or(false)
|| h.known(&name)
})
.unwrap_or(false);
if !known {
return Err(mcp_not_found(&name));
}
Ok(Json(json!({"success": true})))
}
fn auth_overlay_path(st: &AppState) -> std::path::PathBuf {
st.db
.parent()
.unwrap_or(std::path::Path::new("."))
.join("auth-overlay.json")
}
fn write_json_atomic(path: &std::path::Path, value: &Value) -> Result<(), ApiError> {
let dir = path.parent().ok_or_else(|| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: "path has no parent".into(),
})?;
std::fs::create_dir_all(dir).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("mkdir: {e}"),
})?;
let tmp = dir.join(format!(".tmp-{}-auth", std::process::id()));
let data = serde_json::to_vec_pretty(value).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("serialize: {e}"),
})?;
std::fs::write(&tmp, data).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("write: {e}"),
})?;
std::fs::rename(&tmp, path).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("rename: {e}"),
})?;
Ok(())
}
async fn auth_put(
State(st): State<Arc<AppState>>,
axum::extract::Path(pid): axum::extract::Path<String>,
Json(body): Json<Value>,
) -> Result<Json<Value>, ApiError> {
if !body.is_object() {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "auth payload must be an object".into(),
});
}
let path = auth_overlay_path(&st);
let mut overlay: Value = if path.exists() {
std::fs::read_to_string(&path)
.ok()
.and_then(|r| serde_json::from_str(&r).ok())
.unwrap_or_else(|| json!({}))
} else {
json!({})
};
overlay[&pid] = body; write_json_atomic(&path, &overlay)?;
if let Err(e) = watch::reconcile(&st).await {
tracing::warn!("payload reload failed: {e:#}");
}
tracing::info!("auth overlay set provider={pid}");
Ok(Json(json!(true)))
}
async fn auth_delete(
State(st): State<Arc<AppState>>,
axum::extract::Path(pid): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
let path = auth_overlay_path(&st);
if path.exists() {
let mut overlay: Value = std::fs::read_to_string(&path)
.ok()
.and_then(|r| serde_json::from_str(&r).ok())
.unwrap_or_else(|| json!({}));
if let Some(obj) = overlay.as_object_mut() {
obj.remove(&pid);
}
write_json_atomic(&path, &overlay)?;
if let Err(e) = watch::reconcile(&st).await {
tracing::warn!("payload reload failed: {e:#}");
}
}
tracing::info!("auth overlay remove provider={pid}");
Ok(Json(json!(true)))
}
async fn provider_auth_methods(State(st): State<Arc<AppState>>) -> Json<Value> {
let payloads = st.payloads.read();
let mut out = serde_json::Map::new();
if let Some(list) = payloads
.config_providers
.get("providers")
.and_then(|p| p.as_array())
{
for e in list {
if let Some(pid) = e.get("id").and_then(|v| v.as_str()) {
out.insert(
pid.to_string(),
json!([{"type": "api", "label": "API Key"}]),
);
}
}
}
Json(serde_json::Value::Object(out))
}
async fn global_dispose(State(st): State<Arc<AppState>>) -> Result<Json<Value>, ApiError> {
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let props = json!({"directory": dir});
ocserve_store::append_event(&st.writer, None, "server.instance.disposed", &props).map_err(
|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
},
)?;
st.bus.publish(ocserve_core::event::frame(
&dir,
"server.instance.disposed",
props,
));
if let Err(e) = watch::reconcile(&st).await {
tracing::warn!("payload reload failed: {e:#}");
}
tracing::info!("global dispose: emitted + payloads reloaded");
Ok(Json(json!(true)))
}
async fn post_revert(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<axum::response::Response, ApiError> {
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")));
}
let message_id = payload
.get("messageID")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if message_id.is_empty() {
return Ok((
StatusCode::BAD_REQUEST,
axum::Json(json!({
"name": "BadRequest",
"data": {
"message": "Missing key\n at [\"messageID\"]",
"kind": "Payload"
}
})),
)
.into_response());
}
let part_id = payload.get("partID").cloned();
let mut marker = json!({"messageID": message_id});
if let Some(p) = part_id
&& !p.is_null()
{
marker["partID"] = p;
}
ocserve_store::set_session_revert(&st.writer, &id, Some(&marker)).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let info = session_wire_or_partial_http(&st, &id, json!({"revert": marker})).await;
st.bus.publish(ocserve_core::event::frame(
&dir,
"session.updated",
json!({"sessionID": id, "info": info}),
));
let sid = id.clone();
run_blocking(&st.db, move |db| ocserve_store::load_session_wire(db, &sid))
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.map(|j| axum::Json(j).into_response())
.ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))
}
async fn post_unrevert(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")));
}
ocserve_store::set_session_revert(&st.writer, &id, None).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let info = session_wire_or_partial_http(&st, &id, json!({"revert": Value::Null})).await;
st.bus.publish(ocserve_core::event::frame(
&dir,
"session.updated",
json!({"sessionID": id, "info": info}),
));
let sid = id.clone();
run_blocking(&st.db, move |db| ocserve_store::load_session_wire(db, &sid))
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.map(Json)
.ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))
}
async fn instance_dispose(State(st): State<Arc<AppState>>) -> Result<Json<Value>, ApiError> {
let _disposed = global_dispose(State(st)).await?;
Ok(Json(json!(true)))
}
async fn post_init(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<axum::response::Response, HttpError> {
let provider = payload.get("providerID").and_then(|v| v.as_str());
let model = payload.get("modelID").and_then(|v| v.as_str());
let (Some(provider), Some(model)) = (provider, model) else {
let missing = if provider.is_none() {
"providerID"
} else {
"modelID"
};
return Ok((
StatusCode::BAD_REQUEST,
axum::Json(json!({
"name": "BadRequest",
"data": {
"message": format!("Missing key\n at [\"{missing}\"]"),
"kind": "Payload"
}
})),
)
.into_response());
};
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")).into());
}
let cmd_payload = json!({
"command": "init",
"arguments": "",
"model": format!("{provider}/{model}"),
"messageID": payload.get("messageID").cloned().unwrap_or(Value::Null),
});
post_command(State(st), axum::extract::Path(id), Json(cmd_payload))
.await
.map(|j| j.into_response())
}
async fn post_provider_oauth(
axum::extract::Path(_id): axum::extract::Path<String>,
) -> axum::response::Response {
(
StatusCode::BAD_REQUEST,
axum::Json(json!({"name": "BadRequest", "data": {}})),
)
.into_response()
}
async fn post_provider_oauth_callback(
axum::extract::Path(_id): axum::extract::Path<String>,
) -> axum::response::Response {
(
StatusCode::BAD_REQUEST,
axum::Json(json!({"name": "BadRequest", "data": {}})),
)
.into_response()
}
async fn post_mcp_auth_authenticate(
State(st): State<Arc<AppState>>,
axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<axum::response::Response, HttpError> {
mcp_auth_gate(&st, &name)?;
Err(HttpError::TaggedData {
status: StatusCode::BAD_REQUEST,
tag: "McpUnsupportedOAuthError",
fields: json!({"error": format!("MCP server {name} does not support OAuth")}),
})
}
async fn post_mcp_auth_callback(
State(st): State<Arc<AppState>>,
axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<axum::response::Response, HttpError> {
mcp_auth_gate(&st, &name)?;
Err(HttpError::TaggedData {
status: StatusCode::BAD_REQUEST,
tag: "McpUnsupportedOAuthError",
fields: json!({"error": format!("MCP server {name} does not support OAuth")}),
})
}
fn mcp_auth_gate(st: &Arc<AppState>, name: &str) -> Result<(), HttpError> {
let known = st
.mcp
.get()
.map(|h| h.statuses().get(name).is_some())
.unwrap_or(false);
if known {
return Ok(());
}
Err(HttpError::TaggedData {
status: StatusCode::NOT_FOUND,
tag: "McpServerNotFoundError",
fields: json!({
"name": name,
"message": format!("MCP server not found: {name}"),
}),
})
}
async fn get_message_by_id(
State(st): State<Arc<AppState>>,
axum::extract::Path((id, mid)): axum::extract::Path<(String, String)>,
) -> Result<axum::response::Response, ApiError> {
let db = st.db.clone();
let sid = id.clone();
let mid_c = mid.clone();
let found = run_blocking(&db, move |db| {
ocserve_store::message_by_id(db, &sid, &mid_c)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
match found {
Some(v) => Ok(axum::Json(v).into_response()),
None => Err(ApiError::not_found(format!("Message not found: {mid}"))),
}
}
async fn post_session_permission_respond(
State(st): State<Arc<AppState>>,
axum::extract::Path((_id, pid)): axum::extract::Path<(String, String)>,
Json(body): Json<Value>,
) -> Result<axum::response::Response, HttpError> {
let reply = validate_permission_reply_key(&body, "response")?;
if !st.gate.reply(&pid, reply) {
return Err(HttpError::TaggedData {
status: StatusCode::NOT_FOUND,
tag: "PermissionNotFoundError",
fields: json!({
"requestID": pid,
"message": format!("Permission request not found: {pid}"),
}),
});
}
Ok(axum::Json(json!(true)).into_response())
}
async fn post_log(Json(body): Json<Value>) -> Result<Json<Value>, HttpError> {
let level = body.get("level").and_then(|v| v.as_str()).unwrap_or("info");
let service = body
.get("service")
.and_then(|v| v.as_str())
.unwrap_or("app");
let message = body.get("message").and_then(|v| v.as_str()).unwrap_or("");
match level {
"debug" => tracing::debug!(target: "ocserve::client", service, "{message}"),
"warn" => tracing::warn!(target: "ocserve::client", service, "{message}"),
"error" => tracing::error!(target: "ocserve::client", service, "{message}"),
_ => tracing::info!(target: "ocserve::client", service, "{message}"),
}
Ok(Json(json!(true)))
}
async fn search_messages(
State(st): State<Arc<AppState>>,
Json(body): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let query = body
.get("query")
.and_then(|v| v.as_str())
.unwrap_or("")
.trim()
.to_string();
if query.is_empty() || query.chars().count() > 512 {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "query must be a non-empty string of at most 512 characters".into(),
});
}
let scope = body
.get("sessionID")
.and_then(|v| v.as_str())
.map(str::to_string);
let limit = body
.get("limit")
.and_then(|v| v.as_u64())
.unwrap_or(50)
.clamp(1, 200) as u32;
let offset = body.get("offset").and_then(|v| v.as_u64()).unwrap_or(0) as u32;
let needle = query.clone();
let (hits, truncated) = run_blocking(&st.db, move |db| {
ocserve_store::search_parts(db, &needle, scope.as_deref(), limit, offset)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
let out = json!({
"hits": hits
.iter()
.map(|h| json!({
"sessionID": h.session_id,
"messageID": h.message_id,
"partID": h.part_id,
"role": h.role,
"time": h.time,
"snippet": ocserve_store::snippet(&h.text, &query),
}))
.collect::<Vec<_>>(),
"truncated": truncated,
});
Ok(Json(out))
}
fn resolve_model(
st: &Arc<AppState>,
model: Option<&ocserve_core::wire::ModelRef>,
session_id: &str,
) -> (String, String) {
if let Some(m) = model {
return (m.provider_id.clone(), m.model_id.clone());
}
if let Ok(conn) = ocserve_store::pragma::open_reader(&st.db) {
let stored: Option<String> = conn
.query_row(
"SELECT model FROM session WHERE id = ?1",
[session_id],
|r| r.get(0),
)
.ok()
.flatten();
if let Some(raw) = stored
&& let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw)
{
let p = v.get("providerID").and_then(|x| x.as_str()).unwrap_or("");
let m = v.get("id").and_then(|x| x.as_str()).unwrap_or("");
if !p.is_empty() && !m.is_empty() {
return (p.to_string(), m.to_string());
}
}
}
st.llm.read().default_model.clone()
}
pub(crate) fn agent_rules(st: &Arc<AppState>, agent: &str) -> Vec<ocserve_tools::Rule> {
st.payloads
.read()
.agent
.iter()
.find(|a| a["name"].as_str() == Some(agent))
.and_then(|a| a["permission"].as_array())
.map(|arr| {
arr.iter()
.filter_map(|r| {
Some(ocserve_tools::Rule {
permission: r["permission"].as_str()?.to_string(),
pattern: r["pattern"].as_str()?.to_string(),
action: r["action"].as_str()?.to_string(),
})
})
.collect()
})
.unwrap_or_default()
}
fn skill_entries(st: &Arc<AppState>) -> Vec<(String, String, String, String)> {
let payloads = st.payloads.read();
payloads
.command
.iter()
.filter(|c| c.get("source").and_then(|v| v.as_str()) == Some("skill"))
.filter_map(|c| {
let name = c.get("name").and_then(|v| v.as_str())?.to_string();
let description = c
.get("description")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let location = c
.get("location")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.unwrap_or("<built-in>")
.to_string();
let content = c
.get("template")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
Some((name, description, location, content))
})
.collect()
}
pub(crate) fn build_prompt_context(
st: &Arc<AppState>,
req: &ocserve_core::wire::PromptRequest,
session_id: &str,
) -> Result<ocserve_core::prompt::PromptContext, ApiError> {
let agent = req
.agent
.as_deref()
.filter(|a| !a.is_empty())
.map(String::from)
.unwrap_or_else(|| st.llm.read().default_agent.clone());
let system = st
.llm
.read()
.systems
.get(&agent)
.cloned()
.unwrap_or_default();
let (pid, mid) = resolve_model(st, req.model.as_ref(), session_id);
let (base_url, api_key) =
st.llm
.read()
.endpoints
.get(&pid)
.cloned()
.ok_or_else(|| ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: format!("no endpoint configured for provider {pid}"),
})?;
let pricing = st
.llm
.read()
.pricing
.get(&(pid.clone(), mid.clone()))
.copied();
let rules = agent_rules(st, &agent);
let model_limit = st
.llm
.read()
.limits
.get(&(pid.clone(), mid.clone()))
.cloned()
.unwrap_or(serde_json::Value::Null);
let compaction = {
let payloads = st.payloads.read();
ocserve_core::compact::CompactionCfg::from_value(Some(&payloads.compaction))
};
let compaction_system = st
.llm
.read()
.systems
.get("compaction")
.cloned()
.unwrap_or_else(|| {
ocserve_core::system_prompt::base_prompt(None, &mid, &pid)
});
let session_directory = ocserve_store::session_directory(&st.db, session_id)
.ok()
.flatten()
.unwrap_or_else(|| st.paths["directory"].as_str().unwrap_or("/").to_string());
let base = ocserve_core::system_prompt::base_prompt(
if system.is_empty() {
None
} else {
Some(&system)
},
&mid,
&pid,
);
let worktree = vcs::toplevel(std::path::Path::new(&session_directory))
.unwrap_or_else(|| session_directory.clone());
let is_git =
worktree != session_directory || std::path::Path::new(&worktree).join(".git").exists();
let env_block = ocserve_core::system_prompt::environment_block(
&mid,
&pid,
&session_directory,
&worktree,
is_git,
std::env::consts::OS,
&js_to_date_string(now_ms()),
);
let skills = skill_entries(st);
let skill_rows: Vec<(String, String, String)> = skills
.iter()
.map(|(n, d, l, _)| (n.clone(), d.clone(), l.clone()))
.collect();
let skills_blk = ocserve_core::system_prompt::skills_block(&skill_rows);
let system = ocserve_core::system_prompt::compose(&base, &[Some(env_block), skills_blk], None);
let ctx = ocserve_core::prompt::PromptContext {
db: st.db.clone(),
blobs: st.blobs.clone(),
bus: st.bus.clone(),
directory: session_directory,
agent: agent.clone(),
system,
endpoint: ocserve_core::prompt::LlmEndpoint {
base_url,
api_key,
pricing,
},
model_id: mid,
provider_id: pid,
rules,
gate: st.gate.clone(),
mcp: st.mcp.get().cloned(),
plugins: st.plugins.get().cloned(),
questions: st.question_gate.clone(),
skills,
model_limit,
max_rounds: st.max_rounds,
cost_ceiling: st.cost_ceiling,
compaction,
compaction_system,
};
Ok(ctx)
}
fn fmt_js(v: f64) -> String {
if v.is_finite() && v.abs() >= 1e21 {
let raw = format!("{v:e}"); if let Some(pos) = raw.find('e') {
return format!("{}e+{}", &raw[..pos], &raw[pos + 1..]);
}
raw
} else {
format!("{v}")
}
}
fn parse_limit(raw: &str) -> Result<u64, String> {
const INT_MSG: &str = "Expected an integer, got {}\n at [\"limit\"]";
match raw.parse::<f64>() {
Err(_) => Err(INT_MSG.replace("{}", "NaN")),
Ok(v) if v.is_nan() => Err(INT_MSG.replace("{}", "NaN")),
Ok(v) if v.is_infinite() => {
Err(INT_MSG.replace("{}", if v > 0.0 { "Infinity" } else { "-Infinity" }))
}
Ok(v) if v.trunc() != v => Err(INT_MSG.replace("{}", &fmt_js(v))),
Ok(v) if v.abs() > 9007199254740991.0 => Err(INT_MSG.replace("{}", &fmt_js(v))),
Ok(v) if v < 0.0 => Err(format!(
"Expected a value greater than or equal to 0, got {}\n at [\"limit\"]",
fmt_js(v)
)),
Ok(v) => Ok(v as u64),
}
}
fn tagged_bad_request() -> HttpError {
HttpError::Tagged {
status: StatusCode::BAD_REQUEST,
tag: "BadRequest",
}
}
struct LockRelease {
locks: std::sync::Arc<
parking_lot::Mutex<
std::collections::HashMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>,
>,
>,
sid: String,
lock: std::sync::Arc<tokio::sync::Mutex<()>>,
}
impl LockRelease {
fn arc(&self) -> &std::sync::Arc<tokio::sync::Mutex<()>> {
&self.lock
}
}
impl Drop for LockRelease {
fn drop(&mut self) {
let mut map = self.locks.lock();
if let Some(arc) = map.get(&self.sid)
&& std::sync::Arc::strong_count(arc) <= 2
{
map.remove(&self.sid);
}
}
}
pub(crate) async fn lock_session(st: &Arc<AppState>, sid: &str) -> Result<LockRelease, ApiError> {
let lock = {
let mut map = st.prompt_locks.lock();
if map.len() >= 64 && !map.contains_key(sid) {
return Err(ApiError {
status: StatusCode::SERVICE_UNAVAILABLE,
name: "ServiceUnavailableError",
message: "too many concurrent prompts".into(),
});
}
map.entry(sid.to_string())
.or_insert_with(|| std::sync::Arc::new(tokio::sync::Mutex::new(())))
.clone()
};
Ok(LockRelease {
locks: std::sync::Arc::clone(&st.prompt_locks),
sid: sid.to_string(),
lock,
})
}
pub(crate) fn prompt_err(e: anyhow::Error) -> ApiError {
let msg = format!("{e:#}");
if msg.starts_with("Session not found") {
ApiError::not_found(msg)
} else {
ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: msg,
}
}
}
async fn post_message(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<impl IntoResponse, HttpError> {
let req = decode_v1(ocserve_core::wire::PromptRequest::decode(&payload))?;
let ctx = build_prompt_context(&st, &req, &id)?;
let _release = lock_session(&st, &id).await?;
let _guard = _release.arc().lock().await;
let writer = st.writer.clone();
let result = ocserve_core::prompt::run_prompt(&ctx, &writer, &id, &req)
.await
.map_err(prompt_err)?;
drop(_guard);
let (info, parts) = result;
Ok(Json(json!({"info": info, "parts": parts})))
}
async fn post_prompt_async(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<StatusCode, HttpError> {
let req = decode_v1(ocserve_core::wire::PromptRequest::decode(&payload))?;
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")).into());
}
let ctx = build_prompt_context(&st, &req, &id)?;
let release = lock_session(&st, &id).await?;
let writer = st.writer.clone();
let sid = id.clone();
let generation = st
.prompt_gen
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let tasks = st.prompt_tasks.clone();
let tasks_insert = st.prompt_tasks.clone();
let handle = tokio::spawn(async move {
let _release = release;
let guard = _release.arc().lock().await;
match ocserve_core::prompt::run_prompt(&ctx, &writer, &sid, &req).await {
Ok(_) => {}
Err(e) => tracing::error!("prompt_async failed session={sid}: {e:#}"),
}
drop(guard);
let mut t = tasks.lock();
if let Some((g, _)) = t.get(&sid)
&& *g == generation
{
t.remove(&sid);
}
});
tasks_insert.lock().insert(id.clone(), (generation, handle));
Ok(StatusCode::NO_CONTENT)
}
async fn get_children(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")));
}
run_blocking(&st.db, move |db| ocserve_store::load_children(db, &id))
.await
.map(|v| Json(Value::Array(v)))
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})
}
async fn get_todos(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")));
}
run_blocking(&st.db, move |db| ocserve_store::load_todos(db, &id))
.await
.map(|v| Json(Value::Array(v)))
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})
}
async fn patch_session(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")));
}
let mut dirty = false;
if let Some(title) = payload.get("title").and_then(|t| t.as_str()) {
ocserve_store::update_session_title(&st.writer, &id, title).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
dirty = true;
}
if payload.get("model").is_some() {
let raw = payload.get("model").cloned().unwrap_or(Value::Null);
let model_json: Value = match raw {
Value::Null => Value::Null, Value::Object(ref m) => {
let ok = m
.get("id")
.and_then(|v| v.as_str())
.is_some_and(|v| !v.is_empty())
&& m.get("providerID")
.and_then(|v| v.as_str())
.is_some_and(|v| !v.is_empty());
if !ok {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "model must be {id, providerID, variant?} or null".into(),
});
}
Value::String(
serde_json::to_string(&json!({
"id": m["id"],
"providerID": m["providerID"],
"variant": m.get("variant").cloned().unwrap_or(json!("default")),
}))
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?,
)
}
_ => {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "model must be {id, providerID, variant?} or null".into(),
});
}
};
st.writer
.write(vec![ocserve_store::WriteOp::Sql {
sql: "UPDATE session SET model = ?2 WHERE id = ?1".into(),
params: vec![id.clone().into(), model_json],
}])
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
dirty = true;
}
if payload.get("agent").is_some() {
let agent = match payload.get("agent") {
Some(Value::Null) => String::new(),
Some(Value::String(a)) if !a.is_empty() => a.clone(),
_ => {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "agent must be a non-empty string or null".into(),
});
}
};
st.writer
.write(vec![ocserve_store::WriteOp::Sql {
sql: "UPDATE session SET agent = ?2 WHERE id = ?1".into(),
params: vec![id.clone().into(), agent.into()],
}])
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
dirty = true;
}
let info_id = id.clone();
let info = run_blocking(&st.db, move |db| {
ocserve_store::load_session_wire(db, &info_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))?;
if dirty {
st.bus.publish(ocserve_core::event::frame(
st.paths["directory"].as_str().unwrap_or("/"),
"session.updated",
json!({"sessionID": id, "info": info}),
));
}
Ok(Json(info))
}
async fn delete_session_route(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) {
handle.abort();
}
let found = ocserve_store::delete_session(&st.writer, &st.db, &id).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
if !found {
return Err(ApiError::not_found(format!("Session not found: {id}")));
}
st.gate.forget_session(&id);
st.bus.publish(ocserve_core::event::frame(
st.paths["directory"].as_str().unwrap_or("/"),
"session.deleted",
json!({"sessionID": id}),
));
Ok(Json(json!(true)))
}
async fn post_abort(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) {
handle.abort();
tracing::info!("abort: cancelled prompt task for {id}");
}
Ok(Json(json!(true)))
}
async fn delete_message_route(
State(st): State<Arc<AppState>>,
axum::extract::Path((sid, mid)): axum::extract::Path<(String, String)>,
) -> Result<Json<Value>, ApiError> {
let found =
ocserve_store::delete_message(&st.writer, &st.db, &sid, &mid).map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?;
if !found {
return Err(ApiError::not_found(format!("Message not found: {mid}")));
}
st.bus.publish(ocserve_core::event::frame(
st.paths["directory"].as_str().unwrap_or("/"),
"message.removed",
json!({"sessionID": sid, "messageID": mid}),
));
Ok(Json(json!(true)))
}
async fn delete_part_route(
State(st): State<Arc<AppState>>,
axum::extract::Path((_sid, mid, pid)): axum::extract::Path<(String, String, String)>,
) -> Result<Json<Value>, ApiError> {
ocserve_store::delete_part(&st.writer, &st.db, &mid, &pid)
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.then_some(Json(json!(true)))
.ok_or_else(|| ApiError::not_found(format!("Part not found: {pid}")))
}
async fn patch_part_route(
State(st): State<Arc<AppState>>,
axum::extract::Path((sid, mid, pid)): axum::extract::Path<(String, String, String)>,
Json(part): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let body_sid = part.get("sessionID").and_then(|v| v.as_str()).unwrap_or("");
let body_mid = part.get("messageID").and_then(|v| v.as_str()).unwrap_or("");
let body_pid = part.get("id").and_then(|v| v.as_str()).unwrap_or("");
if body_sid != sid || body_mid != mid || body_pid != pid {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "part id/sessionID/messageID must match the path".into(),
});
}
ocserve_store::update_part(&st.writer, &st.blobs, &sid, &mid, &pid, &part)
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})?
.then_some(Json(part))
.ok_or_else(|| ApiError::not_found(format!("Part not found: {pid}")))
}
fn scope_path(base: &std::path::Path, raw: &str) -> Result<std::path::PathBuf, ApiError> {
let joined = if raw.is_empty() || raw == "." {
base.to_path_buf()
} else {
let p = std::path::Path::new(raw);
if p.is_absolute() {
p.to_path_buf()
} else {
base.join(p)
}
};
let canon_base = std::fs::canonicalize(base).unwrap_or_else(|_| base.to_path_buf());
let norm = lexically_normalize(&joined);
if !norm.starts_with(&canon_base) && !norm.starts_with(base) {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: format!("path escapes the session directory: {raw}"),
});
}
let canon = std::fs::canonicalize(&joined).unwrap_or(norm);
if !canon.starts_with(&canon_base) {
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: format!("path escapes the session directory: {raw}"),
});
}
Ok(canon)
}
fn lexically_normalize(p: &std::path::Path) -> std::path::PathBuf {
use std::path::Component;
let mut out = std::path::PathBuf::new();
for c in p.components() {
match c {
Component::ParentDir => {
out.pop();
}
Component::CurDir => {}
other => out.push(other.as_os_str()),
}
}
out
}
async fn list_directory(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, ApiError> {
let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/"));
let raw = q.get("path").map(String::as_str).unwrap_or(".");
let dir = scope_path(base, raw)?;
let dir_for_worker = dir.clone();
let raw_owned = raw.to_string();
let rd = tokio::task::spawn_blocking(move || std::fs::read_dir(&dir_for_worker))
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalServerError",
message: format!("list worker: {e}"),
})?
.map_err(|e| ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: format!("list {}: {e}", dir.display()),
})?;
let mut nodes: Vec<Value> = Vec::new();
for entry in rd.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
let path = entry.path();
let meta = entry.metadata().ok();
let is_dir = meta.as_ref().map(|m| m.is_dir()).unwrap_or(false);
let rel = if raw_owned.is_empty() || raw_owned == "." {
name.clone()
} else {
format!("{}/{}", raw_owned.trim_end_matches('/'), name)
};
let ignored =
name.starts_with('.') || name == "node_modules" || name == "target" || name == ".git";
nodes.push(json!({
"name": name,
"path": rel,
"type": if is_dir { "directory" } else { "file" },
"absolute": path.to_string_lossy(),
"ignored": ignored,
}));
}
nodes.sort_by(|a, b| {
let da = a["type"] == "directory";
let db = b["type"] == "directory";
db.cmp(&da)
.then_with(|| a["name"].as_str().cmp(&b["name"].as_str()))
});
nodes.truncate(5000);
Ok(Json(Value::Array(nodes)))
}
async fn read_file_content(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, HttpError> {
let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/"));
let raw = q
.get("path")
.map(String::as_str)
.ok_or_else(|| HttpError::Query {
message: "Missing key\n at [\"path\"]".into(),
})?;
let path = scope_path(base, raw).map_err(HttpError::Api)?;
let read_path = path.clone();
let bytes = match tokio::task::spawn_blocking(move || std::fs::read(&read_path)).await {
Ok(Ok(b)) => b,
_ => {
return Ok(Json(json!({"type": "text", "content": ""})));
}
};
let truncated = bytes.len() > 1024 * 1024;
let bytes = if truncated {
let mut cut = 1024 * 1024;
while cut > 0 && (bytes[cut] & 0xC0) == 0x80 {
cut -= 1;
}
&bytes[..cut]
} else {
&bytes[..]
};
let (kind, content) = match std::str::from_utf8(bytes) {
Ok(text) => ("text", text.to_string()),
Err(_) => ("binary", String::from_utf8_lossy(bytes).into_owned()),
};
Ok(Json(json!({
"type": kind,
"content": content,
"encoding": "utf-8",
})))
}
async fn find_files(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, HttpError> {
let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/")).to_path_buf();
let query = q
.get("query")
.ok_or_else(|| HttpError::Query {
message: "Missing key\n at [\"query\"]".into(),
})?
.to_lowercase();
if query.is_empty() {
return Ok(Json(json!([])));
}
let want_type = q.get("type").map(String::as_str).unwrap_or("").to_string();
let limit: usize = q
.get("limit")
.and_then(|l| l.parse().ok())
.unwrap_or(10)
.clamp(1, 200);
let glob_mode = query.contains('*') || query.contains('?');
let out = tokio::task::spawn_blocking(move || {
walk_files(&base, &query, &want_type, limit, glob_mode)
})
.await
.map_err(|e| {
HttpError::Api(ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalServerError",
message: format!("find_files worker: {e}"),
})
})?
.map_err(|e| {
HttpError::Api(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: format!("find_files: {e:#}"),
})
})?;
Ok(Json(Value::Array(
out.into_iter().map(Value::String).collect(),
)))
}
static WALK_THREAD: std::sync::Mutex<Option<std::thread::ThreadId>> = std::sync::Mutex::new(None);
pub(crate) fn walk_files(
base: &std::path::Path,
query: &str,
want_type: &str,
limit: usize,
glob_mode: bool,
) -> anyhow::Result<Vec<String>> {
if let Ok(mut g) = WALK_THREAD.lock() {
*g = Some(std::thread::current().id());
}
let mut out: Vec<String> = Vec::new();
let mut stack = vec![(base.to_path_buf(), String::new())];
let mut visited = 0usize;
while let Some((dir, rel)) = stack.pop() {
visited += 1;
if visited > 20_000 || out.len() >= limit * 4 {
break; }
let Ok(rd) = std::fs::read_dir(&dir) else {
continue;
};
for entry in rd.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
if name == ".git" || name == "node_modules" {
continue;
}
let is_dir = entry.file_type().map(|t| t.is_dir()).unwrap_or(false);
let child_rel = if rel.is_empty() {
name.clone()
} else {
format!("{rel}/{name}")
};
let matched = if glob_mode {
ocserve_tools::wildcard_match(&name, query)
} else {
child_rel.to_lowercase().contains(query)
};
if matched
&& (want_type.is_empty()
|| (want_type == "dir" && is_dir)
|| (want_type == "file" && !is_dir))
{
out.push(child_rel.clone());
}
if is_dir {
stack.push((entry.path(), child_rel));
}
}
}
out.sort_by(|a, b| {
let pa = a.to_lowercase().starts_with(query);
let pb = b.to_lowercase().starts_with(query);
pb.cmp(&pa).then_with(|| a.cmp(b))
});
out.truncate(limit);
Ok(out)
}
async fn find_text(
State(st): State<Arc<AppState>>,
axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, HttpError> {
let pattern = q.get("pattern").ok_or_else(|| HttpError::Query {
message: "Missing key\n at [\"pattern\"]".into(),
})?;
if pattern.is_empty() {
return Ok(Json(json!([])));
}
let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/")).to_path_buf();
let pattern = pattern.to_string();
const EXCLUDES: &[&str] = &[
".git",
"node_modules",
"target",
".venv",
"venv",
"__pycache__",
"dist",
"build",
".terraform",
".terragrunt-cache",
];
let base2 = base.clone();
let out = tokio::task::spawn_blocking(move || {
let run = |extra: &[&str]| -> std::io::Result<std::process::Output> {
let mut c = std::process::Command::new("grep");
c.args(["-rnE", "--binary-files=without-match", "-m", "50"]);
c.args(extra);
for d in EXCLUDES {
c.arg(format!("--exclude-dir={d}"));
}
c.arg("--").arg(&pattern).arg(&base2);
c.env("LC_ALL", "C").output()
};
let a = run(&["-n"])?;
let b = run(&["-on", "-b"])?;
Ok::<_, std::io::Error>((a, b))
})
.await
.map_err(|e| {
HttpError::Api(ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("find worker: {e}"),
})
})?;
let (a_out, b_out) = match out {
Ok(o) => o,
Err(e) => {
return Err(HttpError::Api(ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("grep spawn: {e}"),
}));
}
};
if a_out.status.code().unwrap_or(0) > 1 {
return Err(HttpError::Api(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "invalid pattern".into(),
}));
}
let mut subs: std::collections::HashMap<(String, i64), Vec<(usize, String)>> =
std::collections::HashMap::new();
for line in String::from_utf8_lossy(&b_out.stdout).lines() {
if let Some((head, text)) = split3(line) {
let (path, ln, byte) = head;
subs.entry((path, ln)).or_default().push((byte, text));
}
}
let mut matches: Vec<Value> = Vec::new();
for line in String::from_utf8_lossy(&a_out.stdout).lines().take(200) {
let (p1, rest) = match line.split_once(':') {
Some(x) => x,
None => continue,
};
let (ln, text) = match rest.split_once(':') {
Some(x) => x,
None => continue,
};
let ln: i64 = ln.parse().unwrap_or(0);
let line_start = text.as_ptr() as usize; let _ = line_start;
let rel = std::path::Path::new(p1)
.strip_prefix(&base)
.map(|r| r.to_string_lossy().to_string())
.unwrap_or_else(|_| p1.to_string());
let line_subs = subs.get(&(p1.to_string(), ln)).cloned().unwrap_or_default();
let mut submatch_json = Vec::new();
let mut absolute_offset: i64 = 0;
for (i, (byteoff, mtext)) in line_subs.iter().enumerate() {
let from = submatch_json
.last()
.and_then(|v: &Value| v["end"].as_u64())
.map(|e| e as usize + 1)
.unwrap_or(0);
let start = text[from.min(text.len())..]
.find(mtext.as_str())
.map(|p| p + from.min(text.len()))
.unwrap_or(0);
let end = start + mtext.len();
if i == 0 {
absolute_offset = (*byteoff as i64) - (start as i64);
}
submatch_json.push(json!({
"match": {"text": mtext},
"start": start,
"end": end,
}));
}
matches.push(json!({
"path": {"text": rel},
"lines": {"text": format!("{text}\n")},
"line_number": ln,
"absolute_offset": absolute_offset,
"submatches": submatch_json,
}));
}
Ok(Json(Value::Array(matches)))
}
fn split3(line: &str) -> Option<((String, i64, usize), String)> {
let (path, rest) = line.split_once(':')?;
let (ln, rest) = rest.split_once(':')?;
let (byte, text) = rest.split_once(':')?;
Some((
(path.to_string(), ln.parse().ok()?, byte.parse().ok()?),
text.to_string(),
))
}
async fn get_questions(State(st): State<Arc<AppState>>) -> Json<Value> {
Json(Value::Array(st.question_gate.list()))
}
pub(crate) fn validate_request_prefix(id: &str, prefix: &str) -> Result<(), HttpError> {
if id.starts_with(prefix) {
Ok(())
} else {
Err(HttpError::Params {
message: format!(
"Expected a string starting with \"{prefix}\", got \"{id}\"\n at [\"requestID\"]"
),
})
}
}
pub(crate) fn question_not_found(id: &str) -> HttpError {
HttpError::TaggedData {
status: StatusCode::NOT_FOUND,
tag: "QuestionNotFoundError",
fields: json!({"requestID": id, "message": format!("Question request not found: {id}")}),
}
}
async fn post_question_reply(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(body): Json<Value>,
) -> Result<Json<Value>, HttpError> {
validate_request_prefix(&id, "que")?;
let raw = body
.get("answers")
.and_then(|a| a.as_array())
.ok_or_else(|| {
HttpError::Api(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "body must be {answers: [[String]]}".into(),
})
})?;
let mut answers: Vec<Vec<String>> = Vec::with_capacity(raw.len());
for a in raw {
let row = a.as_array().ok_or_else(|| {
HttpError::Api(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: "each answer must be an array of labels".into(),
})
})?;
answers.push(
row.iter()
.map(|l| l.as_str().unwrap_or_default().to_string())
.collect(),
);
}
if st.question_gate.reply(&id, answers) {
Ok(Json(json!(true)))
} else {
Err(question_not_found(&id))
}
}
async fn post_question_reject(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, HttpError> {
validate_request_prefix(&id, "que")?;
if st.question_gate.reject(&id) {
Ok(Json(json!(true)))
} else {
Err(question_not_found(&id))
}
}
pub fn split_command_args(arguments: &str) -> Vec<String> {
let bytes = arguments.as_bytes();
let mut out = Vec::new();
let mut i = 0;
while i < bytes.len() {
let c = bytes[i];
if c.is_ascii_whitespace() {
i += 1;
continue;
}
let (tok, next) = if c == b'"' || c == b'\'' {
let quote = c;
let mut j = i + 1;
while j < bytes.len() && bytes[j] != quote {
j += 1;
}
let end = (j + 1).min(bytes.len());
(
std::str::from_utf8(&bytes[i..end])
.unwrap_or("")
.to_string(),
end,
)
} else {
let mut j = i;
while j < bytes.len() && !bytes[j].is_ascii_whitespace() {
j += 1;
}
(
std::str::from_utf8(&bytes[i..j]).unwrap_or("").to_string(),
j,
)
};
let mut t = tok.as_str();
if t.starts_with(['"', '\'']) {
t = &t[1..];
}
if t.ends_with(['"', '\'']) && !t.is_empty() {
t = &t[..t.len() - 1];
}
out.push(t.to_string());
i = next;
}
out
}
pub fn expand_command(template: &str, arguments: &str) -> String {
let args = split_command_args(arguments);
let mut last = 0usize;
let b = template.as_bytes();
let mut i = 0;
while i + 1 < b.len() {
if b[i] == b'$' && b[i + 1].is_ascii_digit() {
let mut j = i + 1;
while j < b.len() && b[j].is_ascii_digit() {
j += 1;
}
if let Some(n) = std::str::from_utf8(&b[i + 1..j])
.ok()
.and_then(|d| d.parse::<usize>().ok())
{
last = last.max(n);
}
i = j;
} else {
i += 1;
}
}
let mut with_args = String::new();
let mut i = 0;
let mut had_placeholder = false;
while i < b.len() {
if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
let mut j = i + 1;
while j < b.len() && b[j].is_ascii_digit() {
j += 1;
}
let n: usize = std::str::from_utf8(&b[i + 1..j])
.unwrap_or("0")
.parse()
.unwrap_or(0);
had_placeholder = true;
let arg_index = n.saturating_sub(1);
let value = if arg_index >= args.len() {
String::new()
} else if n == last {
args[arg_index..].join(" ")
} else {
args[arg_index].clone()
};
with_args.push_str(&value);
i = j;
} else {
let ch_len = {
let ch = template[i..].chars().next().unwrap_or(' ');
ch.len_utf8()
};
with_args.push_str(&template[i..i + ch_len]);
i += ch_len;
}
}
let uses_arguments = template.contains("$ARGUMENTS");
let mut expanded = with_args.replace("$ARGUMENTS", arguments);
if !had_placeholder && !uses_arguments && !arguments.trim().is_empty() {
expanded = format!("{expanded}\n\n{arguments}");
}
expanded.trim().to_string()
}
fn emit_session_error_event(st: &Arc<AppState>, sid: &str, message: &str) {
let props = json!({
"sessionID": sid,
"error": {"name": "UnknownError", "data": {"message": message}},
});
if let Err(e) = ocserve_store::append_event(&st.writer, Some(sid), "session.error", &props) {
tracing::warn!("session.error persist failed: {e:#}");
}
let seq = ocserve_store::next_event_seq(&st.db, sid).unwrap_or(0);
let dir = st.paths["directory"].as_str().unwrap_or("/");
st.bus.publish(ocserve_core::event::frame(
dir,
"session.error",
props.clone(),
));
st.bus.publish(ocserve_core::event::sync_frame(
dir,
"session.error",
props,
seq as u64,
sid,
));
}
pub(crate) fn session_busy(id: &str) -> ApiError {
ApiError {
status: StatusCode::CONFLICT,
name: "SessionBusyError",
message: format!("Session is busy: {id}"),
}
}
async fn post_command(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, HttpError> {
let cmd = decode_v1(ocserve_core::wire::CommandRequest::decode(&payload))?;
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")).into());
}
let name = cmd.command.as_str();
let arguments = cmd.arguments.as_str();
let (template, available_owned): (Option<String>, Vec<String>) = {
let payloads = st.payloads.read();
let t = payloads
.command
.iter()
.find(|c| c["name"].as_str() == Some(name))
.and_then(|c| c["template"].as_str())
.map(str::to_string);
let names = payloads
.command
.iter()
.filter_map(|c| c["name"].as_str().map(str::to_string))
.collect();
(t, names)
};
let available: Vec<&str> = available_owned.iter().map(String::as_str).collect();
let Some(template) = template.as_deref() else {
let msg = format!(
"Command not found: \"{name}\".{}",
if available.is_empty() {
String::new()
} else {
format!(" Available commands: {}", available.join(", "))
}
);
emit_session_error_event(&st, &id, &msg);
return Err(ApiError {
status: StatusCode::BAD_REQUEST,
name: "BadRequest",
message: msg,
}
.into());
};
let expanded = expand_command(template, arguments);
let mut cmd_parts = json!([{ "type": "text", "text": expanded }]);
let parts_out = ocserve_core::prompt::hook_mutate(
st.plugins.get(),
"command.execute.before",
json!({
"command": name,
"sessionID": id,
"arguments": arguments,
}),
json!({"parts": cmd_parts.clone()}),
)
.await;
if let Some(p) = parts_out.get("parts").and_then(|p| p.as_array()) {
cmd_parts = Value::Array(p.clone());
}
let req = ocserve_core::wire::PromptRequest {
message_id: cmd.message_id.clone(),
model: cmd
.model
.as_deref()
.map(ocserve_core::wire::parse_model_str),
agent: cmd.agent.clone(),
no_reply: false,
tools: None,
format: None,
system: None,
variant: cmd.variant.clone(),
parts: cmd_parts.as_array().cloned().unwrap_or_default(),
};
let ctx = build_prompt_context(&st, &req, &id)?;
let _release = lock_session(&st, &id).await?;
let Ok(_guard) = _release.arc().try_lock() else {
return Err(session_busy(&id).into());
};
let writer = st.writer.clone();
let (info, parts) = ocserve_core::prompt::run_prompt_with(
&ctx,
&writer,
&id,
&req,
ocserve_core::prompt::RunOpts {
auto_title: false,
..Default::default()
},
)
.await
.map_err(prompt_err)?;
drop(_guard);
Ok(Json(json!({"info": info, "parts": parts})))
}
async fn post_shell(
State(st): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, HttpError> {
let sh = decode_v1(ocserve_core::wire::ShellRequest::decode(&payload))?;
let exists_id = id.clone();
if !run_blocking(&st.db, move |db| {
ocserve_store::session_exists(db, &exists_id)
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
})? {
return Err(ApiError::not_found(format!("Session not found: {id}")).into());
}
let command = sh.command.as_str();
let agent = sh.agent.as_str();
let (pid, mid) = resolve_model(&st, sh.model.as_ref(), &id);
let _release = lock_session(&st, &id).await?;
let Ok(_guard) = _release.arc().try_lock() else {
return Err(session_busy(&id).into());
};
let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
let status_frame = |kind: &str| {
ocserve_core::event::frame(
&dir,
"session.status",
json!({"sessionID": id, "status": {"type": kind}}),
)
};
st.bus.publish(status_frame("busy"));
let started = now_ms_local();
let user_msg_id = sh
.message_id
.clone()
.unwrap_or_else(ocserve_core::ids::msg_id);
let user_part = json!({
"type": "text", "id": ocserve_core::ids::prt_id(),
"sessionID": id, "messageID": user_msg_id,
"text": "The following tool was executed by the user",
"synthetic": true,
});
let user_info = json!({
"id": user_msg_id,
"sessionID": id,
"role": "user",
"agent": agent,
"model": {"providerID": pid, "modelID": mid},
"time": {"created": started},
"summary": {"diffs": []},
});
let assistant_id = ocserve_core::ids::msg_id();
let call_id = ocserve_core::ids::evt_id();
let mut tool_part = json!({
"type": "tool", "id": ocserve_core::ids::prt_id(),
"sessionID": id, "messageID": assistant_id,
"callID": call_id, "tool": "bash",
"state": {
"status": "running",
"input": {"command": command},
"time": {"start": started},
},
});
let writer = st.writer.clone();
insert_message_http(
&writer,
&st,
&id,
&user_info,
std::slice::from_ref(&user_part),
)?;
let assistant_info = json!({
"id": assistant_id,
"parentID": user_msg_id,
"role": "assistant",
"mode": agent,
"agent": agent,
"path": {"cwd": dir, "root": "/"},
"cost": 0.0,
"tokens": {"total": 0, "input": 0, "output": 0, "reasoning": 0,
"cache": {"write": 0, "read": 0}},
"modelID": mid,
"providerID": pid,
"time": {"created": started, "completed": started},
"finish": "stop",
"id": assistant_id,
"sessionID": id,
});
insert_message_http(&writer, &st, &id, &assistant_info, &[])?;
emit_durable_http(
&st,
&id,
"message.updated",
json!({"sessionID": id, "info": user_info.clone()}),
)?;
emit_durable_http(
&st,
&id,
"message.updated",
json!({"sessionID": id, "info": assistant_info.clone()}),
)?;
emit_durable_http(
&st,
&id,
"message.part.updated",
json!({"sessionID": id, "part": user_part}),
)?;
emit_durable_http(
&st,
&id,
"message.part.updated",
json!({"sessionID": id, "part": tool_part.clone()}),
)?;
let exec = {
let cmd = json!({"command": command});
let d = dir.clone();
tokio::task::spawn_blocking(move || {
ocserve_tools::execute("bash", &cmd, std::path::Path::new(&d))
})
.await
.map_err(|e| ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("tool join: {e}"),
})?
};
let (status, output) = match exec {
Ok(r) => ("completed", r.output),
Err(e) => ("error", format!("Error: {e:#}")),
};
let completed = now_ms_local();
tool_part["state"] = json!({
"status": status,
"input": {"command": command},
"output": output,
"title": "",
"metadata": {"output": ""},
"time": {"start": started, "end": completed},
});
tool_part["state"]["metadata"]["output"] = tool_part["state"]["output"].clone();
insert_message_http(
&writer,
&st,
&id,
&assistant_info,
std::slice::from_ref(&tool_part),
)?;
emit_durable_http(
&st,
&id,
"message.part.updated",
json!({"sessionID": id, "part": tool_part.clone()}),
)?;
st.bus.publish(status_frame("idle"));
Ok(Json(json!({"info": assistant_info, "parts": [tool_part]})))
}
fn now_ms_local() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
fn insert_message_http(
writer: &ocserve_store::Writer,
st: &Arc<AppState>,
sid: &str,
info: &Value,
parts: &[Value],
) -> Result<(), ApiError> {
ocserve_store::insert_message(writer, Some(&*st.blobs), sid, info, parts).map_err(|e| {
ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
}
})
}
async fn session_wire_or_partial_http(st: &Arc<AppState>, sid: &str, overrides: Value) -> Value {
let sid_owned = sid.to_string();
run_blocking(&st.db, move |db| {
ocserve_store::session_wire_with(db, &sid_owned, overrides)
})
.await
.ok()
.flatten()
.unwrap_or_else(|| json!({"id": sid}))
}
fn emit_durable_http(
st: &Arc<AppState>,
sid: &str,
event_type: &str,
props: Value,
) -> Result<(), ApiError> {
let props = ocserve_core::event::normalize_props(event_type, props);
ocserve_store::append_event(&st.writer, Some(sid), event_type, &props).map_err(|e| {
ApiError {
status: StatusCode::INTERNAL_SERVER_ERROR,
name: "InternalError",
message: format!("{e:#}"),
}
})?;
let seq = ocserve_store::next_event_seq(&st.db, sid).unwrap_or(0);
let dir = st.paths["directory"].as_str().unwrap_or("/");
st.bus
.publish(ocserve_core::event::frame(dir, event_type, props.clone()));
st.bus.publish(ocserve_core::event::sync_frame(
dir, event_type, props, seq as u64, sid,
));
Ok(())
}
fn evt_id(kind: &str) -> String {
use sha2::{Digest, Sha256};
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let mut h = Sha256::new();
h.update(nanos.to_le_bytes());
h.update(kind.as_bytes());
let hex = hex::encode(h.finalize());
format!("evt_{}", &hex[..26])
}
pub(crate) async fn global_event(State(st): State<Arc<AppState>>) -> Response {
event_stream(st, EventFormat::Global).await
}
async fn bare_event(State(st): State<Arc<AppState>>) -> Response {
event_stream(st, EventFormat::Bare).await
}
async fn v2_event(State(st): State<Arc<AppState>>) -> Response {
event_stream(st, EventFormat::V2).await
}
#[derive(Clone, Copy)]
enum EventFormat {
Global,
Bare,
V2,
}
fn encode_event_frame(stored: &str, fmt: EventFormat) -> Option<String> {
if matches!(fmt, EventFormat::Global) {
return Some(stored.to_string());
}
let v: Value = serde_json::from_str(stored).ok()?;
let payload = v.get("payload")?;
if payload.get("type").and_then(Value::as_str) == Some("sync") {
return None;
}
let id = payload.get("id").cloned().unwrap_or(Value::Null);
let ty = payload.get("type").cloned().unwrap_or(Value::Null);
let props = payload.get("properties").cloned().unwrap_or(json!({}));
Some(match fmt {
EventFormat::Bare => json!({"id": id, "type": ty, "properties": props}).to_string(),
EventFormat::V2 => json!({"id": id, "type": ty, "data": props}).to_string(),
EventFormat::Global => unreachable!(),
})
}
fn synthetic_frame(fmt: EventFormat, event_type: &str) -> String {
let id = evt_id(if event_type == "server.connected" {
"connected"
} else {
"heartbeat"
});
match fmt {
EventFormat::Global => json!({"payload": {
"id": id, "type": event_type, "properties": {}}})
.to_string(),
EventFormat::Bare => json!({
"id": id, "type": event_type, "properties": {}})
.to_string(),
EventFormat::V2 => json!({
"id": id, "type": event_type, "data": {}})
.to_string(),
}
}
async fn event_stream(st: Arc<AppState>, fmt: EventFormat) -> Response {
let connected = Event::default().data(synthetic_frame(fmt, "server.connected"));
let rx = st.bus.subscribe();
ocserve_metrics::gauge(
"ocserve_sse_clients",
st.bus.subscriber_count().max(1) as i64,
);
let bus_keepalive = st.bus.clone();
struct SseClientGuard;
impl Drop for SseClientGuard {
fn drop(&mut self) {
ocserve_metrics::gauge_delta("ocserve_sse_clients", -1);
}
}
let sse_guard = SseClientGuard;
let events = stream::once(async move { Ok::<_, std::convert::Infallible>(connected) }).chain(
stream::unfold(
(
rx,
tokio::time::Instant::now() + Duration::from_secs(10),
bus_keepalive,
sse_guard,
fmt,
),
|(mut rx, mut next_hb, bus, guard, fmt)| async move {
loop {
let dur = next_hb.saturating_duration_since(tokio::time::Instant::now());
match tokio::time::timeout(dur.max(Duration::from_millis(1)), rx.recv()).await {
Ok(Ok(frame)) => {
ocserve_metrics::counter("ocserve_sse_events_total", 1);
if let Some(out) = encode_event_frame(&frame, fmt) {
return Some((
Ok(Event::default().data(&out)),
(rx, next_hb, bus, guard, fmt),
));
}
}
Ok(Err(_lagged_or_closed)) => {
ocserve_metrics::labeled_counter(
"ocserve_event_ring_lag_total",
"reason=\"lagged_or_closed\"",
1,
);
return None;
}
Err(_elapsed) => {
let hb =
Event::default().data(synthetic_frame(fmt, "server.heartbeat"));
next_hb += Duration::from_secs(10);
return Some((Ok(hb), (rx, next_hb, bus, guard, fmt)));
}
}
}
},
),
);
let mut resp = Sse::new(events).into_response();
let h = resp.headers_mut();
for (k, v) in sse_expected_headers() {
if let (Ok(name), Ok(val)) = (
axum::http::HeaderName::try_from(k),
HeaderValue::from_str(v),
) {
h.insert(name, val);
}
}
resp
}
fn req_log() -> bool {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ON.get_or_init(|| {
std::env::var("OCSERVE_REQ_LOG")
.map(|v| v == "1")
.unwrap_or(false)
})
}
async fn auth_gate(
State(st): State<Arc<AppState>>,
req: axum::extract::Request,
next: axum::middleware::Next,
) -> Result<Response, ApiError> {
st.requests
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
ocserve_metrics::counter("ocserve_requests_total", 1);
let metric_t0 = std::time::Instant::now();
let metric_label = format!(
"route=\"{}\",method=\"{}\"",
ocserve_metrics::route_label(req.uri().path()),
req.method()
);
let record = move |t0: std::time::Instant| {
ocserve_metrics::observe(
"ocserve_http_request_duration_seconds",
&metric_label,
t0.elapsed().as_micros() as u64,
);
};
if req_log() {
tracing::info!("req {} {}", req.method(), req.uri().path());
}
if let Some((user, pass)) = &st.auth {
let authorized = req
.headers()
.get(axum::http::header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Basic "))
.and_then(|b64| {
use base64::Engine;
base64::engine::general_purpose::STANDARD.decode(b64).ok()
})
.map(|raw| {
let text = String::from_utf8_lossy(&raw);
text == format!("{user}:{pass}")
})
.unwrap_or(false);
if !authorized {
record(metric_t0);
return Err(ApiError {
status: StatusCode::UNAUTHORIZED,
name: "UnauthorizedError",
message: "unauthorized".into(),
});
}
}
let resp = next.run(req).await;
record(metric_t0);
Ok(resp)
}
pub fn router(state: Arc<AppState>) -> Router {
Router::new()
.route("/global/health", get(health))
.route("/doc", get(get_doc))
.route("/config", get(get_config))
.route("/config/providers", get(get_config_providers))
.route("/provider", get(get_provider))
.route("/agent", get(get_agent))
.route("/command", get(get_command))
.route("/experimental/console", get(get_console))
.route("/experimental/tool", get(get_experimental_tool))
.route("/experimental/tool/ids", get(get_experimental_tool_ids))
.route("/experimental/capabilities", get(get_capabilities))
.route("/path", get(get_path))
.route("/project", get(get_projects))
.route("/project/current", get(get_project_current))
.route("/session", get(get_sessions).post(post_session))
.route("/session/{id}/fork", axum::routing::post(post_fork))
.route("/session/status", get(session_status))
.route("/session/search", axum::routing::post(search_messages))
.route("/mcp/{name}/{action}", axum::routing::post(mcp_action))
.route("/mcp/{name}/auth", axum::routing::delete(mcp_auth_remove))
.route(
"/auth/{providerID}",
axum::routing::put(auth_put).delete(auth_delete),
)
.route("/provider/auth", axum::routing::get(provider_auth_methods))
.route("/global/dispose", axum::routing::post(global_dispose))
.route("/config", axum::routing::patch(patch_config))
.route("/global/config", axum::routing::patch(patch_config))
.route(
"/session/{id}/summarize",
axum::routing::post(post_summarize),
)
.route(
"/session/{id}",
get(get_session)
.patch(patch_session)
.delete(delete_session_route),
)
.route(
"/session/{id}/message",
get(get_messages).post(post_message),
)
.route("/global/event", get(global_event))
.route("/event", get(bare_event))
.route("/api/event", get(v2_event))
.route("/metrics", get(metrics))
.route("/experimental/resource", get(experimental_resource))
.route(
"/experimental/workspace",
get(experimental_workspace).post(v2::workspace_create),
)
.route(
"/experimental/workspace/status",
get(experimental_workspace_status),
)
.route("/formatter", get(formatter_list))
.route("/lsp", get(lsp_list))
.route("/project/{id}/directories", get(project_directories))
.route("/vcs", get(vcs_info))
.route("/vcs/status", get(vcs_status))
.route("/vcs/diff", get(vcs_diff))
.route("/vcs/diff/raw", get(vcs_diff_raw))
.route("/file/status", get(file_status))
.route("/find/symbol", get(find_symbol))
.route("/global/config", get(global_config_get))
.route(
"/session/{id}/prompt_async",
axum::routing::post(post_prompt_async),
)
.route("/session/{id}/children", get(get_children))
.route("/file", get(list_directory))
.route("/file/content", get(read_file_content))
.route("/find/file", get(find_files))
.route("/find", get(find_text))
.route("/question", get(get_questions))
.route(
"/question/{id}/reply",
axum::routing::post(post_question_reply),
)
.route(
"/question/{id}/reject",
axum::routing::post(post_question_reject),
)
.route("/session/{id}/todo", get(get_todos))
.route("/session/{id}/abort", axum::routing::post(post_abort))
.route("/session/{id}/command", axum::routing::post(post_command))
.route("/session/{id}/shell", axum::routing::post(post_shell))
.route(
"/session/{id}/message/{mid}",
get(get_message_by_id).delete(delete_message_route),
)
.route(
"/session/{id}/permissions/{pid}",
axum::routing::post(post_session_permission_respond),
)
.route("/log", axum::routing::post(post_log))
.route("/session/{id}/revert", axum::routing::post(post_revert))
.route("/session/{id}/unrevert", axum::routing::post(post_unrevert))
.route("/instance/dispose", axum::routing::post(instance_dispose))
.route("/session/{id}/init", axum::routing::post(post_init))
.route(
"/provider/{id}/oauth/authorize",
axum::routing::post(post_provider_oauth),
)
.route(
"/provider/{id}/oauth/callback",
axum::routing::post(post_provider_oauth_callback),
)
.route(
"/mcp/{name}/auth/authenticate",
axum::routing::post(post_mcp_auth_authenticate),
)
.route(
"/mcp/{name}/auth/callback",
axum::routing::post(post_mcp_auth_callback),
)
.route(
"/session/{id}/message/{mid}/part/{pid}",
axum::routing::delete(delete_part_route).patch(patch_part_route),
)
.route("/experimental/session", get(get_experimental_sessions))
.route("/pty/shells", get(pty::get_shells))
.route("/pty", get(pty::list).post(pty::create))
.route(
"/pty/{id}",
get(pty::get).put(pty::update).delete(pty::remove),
)
.route(
"/pty/{id}/connect-token",
axum::routing::post(pty::connect_token),
)
.route("/pty/{id}/connect", get(pty::connect))
.route(
"/tui/append-prompt",
axum::routing::post(tui::append_prompt),
)
.route("/tui/open-help", axum::routing::post(tui::open_help))
.route(
"/tui/open-sessions",
axum::routing::post(tui::open_sessions),
)
.route("/tui/open-themes", axum::routing::post(tui::open_themes))
.route("/tui/open-models", axum::routing::post(tui::open_models))
.route(
"/tui/submit-prompt",
axum::routing::post(tui::submit_prompt),
)
.route("/tui/clear-prompt", axum::routing::post(tui::clear_prompt))
.route(
"/tui/execute-command",
axum::routing::post(tui::execute_command),
)
.route("/tui/show-toast", axum::routing::post(tui::show_toast))
.route("/tui/publish", axum::routing::post(tui::publish_event))
.route(
"/tui/select-session",
axum::routing::post(tui::select_session),
)
.route("/tui/control/next", get(tui::control_next))
.route(
"/tui/control/response",
axum::routing::post(tui::control_response),
)
.route("/mcp", get(get_mcp).post(post_mcp))
.route("/permission", get(get_permissions))
.route(
"/permission/{id}/reply",
axum::routing::post(post_permission_reply),
)
.route("/api/health", get(v2::health))
.route("/api/location", get(v2::location))
.route("/api/agent", get(get_api_agent))
.route("/api/model", get(v2::list_models))
.route("/api/provider", get(v2::list_providers))
.route("/api/provider/{providerID}", get(v2::get_provider))
.route("/api/skill", get(v2::list_skills))
.route("/skill", get(v2::list_skills_raw))
.route("/vcs/apply", axum::routing::post(v2::vcs_apply))
.route("/api/command", get(get_api_command))
.route("/api/integration", get(v2::list_integrations))
.route("/api/integration/{integrationID}", get(v2::get_integration))
.route("/api/reference", get(v2::list_references))
.route("/api/fs/find", get(v2::fs_find))
.route("/api/fs/list", get(v2::fs_list))
.route("/api/fs/read/{*path}", get(v2::fs_read))
.route(
"/api/session",
get(v2::list_sessions).post(v2::create_session),
)
.route("/api/session/active", get(v2::active_sessions))
.route("/api/session/{sessionID}", get(v2::get_session))
.route("/api/session/{sessionID}/message", get(v2::list_messages))
.route(
"/api/session/{sessionID}/message/{messageID}",
get(v2::get_message),
)
.route("/api/session/{sessionID}/context", get(v2::session_context))
.route("/api/session/{sessionID}/history", get(v2::session_history))
.route(
"/api/session/{sessionID}/agent",
axum::routing::post(v2::switch_agent),
)
.route(
"/api/session/{sessionID}/model",
axum::routing::post(v2::switch_model),
)
.route(
"/api/session/{sessionID}/prompt",
axum::routing::post(v2::prompt),
)
.route(
"/api/session/{sessionID}/interrupt",
axum::routing::post(v2::interrupt),
)
.route(
"/api/session/{sessionID}/compact",
axum::routing::post(v2::compact),
)
.route(
"/api/session/{sessionID}/wait",
axum::routing::post(v2::wait),
)
.route(
"/api/session/{sessionID}/revert/stage",
axum::routing::post(v2::revert_stage),
)
.route(
"/api/session/{sessionID}/revert/clear",
axum::routing::post(v2::revert_clear),
)
.route(
"/api/session/{sessionID}/revert/commit",
axum::routing::post(v2::revert_commit),
)
.route(
"/api/session/{sessionID}/permission/{requestID}/reply",
axum::routing::post(v2::permission_reply),
)
.route(
"/api/session/{sessionID}/question/{requestID}/reply",
axum::routing::post(v2::question_reply),
)
.route(
"/api/session/{sessionID}/question/{requestID}/reject",
axum::routing::post(v2::question_reject),
)
.route("/api/permission/saved", get(v2::permission_saved))
.route(
"/api/permission/saved/{id}",
axum::routing::delete(v2::permission_saved_delete),
)
.route("/api/permission/request", get(v2::permission_request))
.route("/api/question/request", get(v2::question_request))
.route(
"/api/session/{sessionID}/question",
get(v2::session_question),
)
.route(
"/api/session/{sessionID}/permission/{requestID}",
get(v2::session_permission_one),
)
.route(
"/api/session/{sessionID}/permission",
get(v2::session_permission).post(v2::permission_create),
)
.route("/api/pty", get(v2::pty_list).post(v2::pty_create))
.route(
"/api/pty/{ptyID}",
get(v2::pty_get).put(v2::pty_update).delete(v2::pty_remove),
)
.route(
"/api/pty/{ptyID}/connect-token",
axum::routing::post(v2::pty_connect_token),
)
.route("/api/pty/{ptyID}/connect", get(pty::connect))
.route("/session/{sessionID}/diff", get(v2::session_diff))
.route(
"/session/{sessionID}/share",
axum::routing::post(v2::session_share).delete(v2::session_unshare),
)
.route("/experimental/console/orgs", get(v2::console_orgs))
.route(
"/experimental/console/switch",
axum::routing::post(v2::console_switch),
)
.route(
"/experimental/workspace/adapter",
get(v2::workspace_adapter),
)
.route(
"/experimental/workspace/sync-list",
axum::routing::post(v2::workspace_sync_list),
)
.route(
"/experimental/workspace/warp",
axum::routing::post(v2::workspace_warp),
)
.route(
"/experimental/worktree",
get(v2::worktree_list).post(v2::worktree_create),
)
.route(
"/experimental/worktree",
axum::routing::delete(v2::worktree_remove),
)
.route(
"/experimental/worktree/reset",
axum::routing::post(v2::worktree_reset),
)
.route("/api/session/{sessionID}/event", get(v2::session_event))
.route(
"/project/{projectID}",
axum::routing::patch(v2::patch_project),
)
.route("/sync/start", axum::routing::post(v2::sync_start))
.route("/sync/replay", axum::routing::post(v2::sync_replay))
.route("/sync/history", axum::routing::post(v2::sync_history))
.route("/sync/steal", axum::routing::post(v2::sync_steal))
.route("/project/git/init", axum::routing::post(v2::git_init))
.route("/global/upgrade", axum::routing::post(v2::global_upgrade))
.route(
"/api/integration/{integrationID}/connect/key",
axum::routing::post(v2::integration_connect_key),
)
.route(
"/api/integration/{integrationID}/connect/oauth",
axum::routing::post(v2::integration_connect_oauth),
)
.route(
"/api/integration/attempt/{attemptID}",
get(v2::integration_attempt_get).delete(v2::integration_attempt_delete),
)
.route(
"/api/integration/attempt/{attemptID}/complete",
axum::routing::post(v2::integration_attempt_complete),
)
.route(
"/api/credential/{credentialID}",
axum::routing::patch(v2::credential_update).delete(v2::credential_remove),
)
.route(
"/experimental/control-plane/move-session",
axum::routing::post(v2::control_move_session),
)
.route(
"/experimental/project/{projectID}/copy",
axum::routing::post(v2::project_copy_create).delete(v2::project_copy_remove),
)
.route(
"/experimental/project/{projectID}/copy/generate-name",
axum::routing::post(v2::project_copy_generate_name),
)
.route(
"/experimental/project/{projectID}/copy/refresh",
axum::routing::post(v2::project_copy_refresh),
)
.route(
"/experimental/session/{sessionID}/background",
axum::routing::post(v2::session_background),
)
.route(
"/experimental/workspace/{id}",
axum::routing::delete(v2::workspace_delete),
)
.layer(axum::middleware::from_fn_with_state(
state.clone(),
auth_gate,
))
.layer(axum::middleware::from_fn_with_state(
state.clone(),
crate::ui::cors_gate,
))
.fallback(crate::ui::ui_fallback)
.method_not_allowed_fallback(crate::ui::ui_fallback)
.with_state(state)
}
fn read_cors_env() -> Vec<String> {
std::env::var("OCSERVE_CORS")
.ok()
.map(|v| {
v.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect()
})
.unwrap_or_default()
}
pub fn keypaths(v: &serde_json::Value, prefix: &str, out: &mut std::collections::BTreeSet<String>) {
match v {
serde_json::Value::Object(m) => {
for (k, val) in m {
let p = format!("{prefix}.{k}");
out.insert(p.clone());
keypaths(val, &p, out);
}
}
serde_json::Value::Array(a) => {
let p = format!("{prefix}[]");
if let Some(first) = a.first() {
keypaths(first, &p, out);
} else {
out.insert(p);
}
}
_ => {}
}
}
pub fn sse_expected_headers() -> Vec<(&'static str, &'static str)> {
vec![
("cache-control", "no-cache, no-transform"),
("x-content-type-options", "nosniff"),
("content-type", "text/event-stream"),
("x-accel-buffering", "no"),
]
}
#[allow(dead_code)]
fn _assert_header_value(v: &str) -> HeaderValue {
HeaderValue::from_str(v).expect("static header value")
}
#[cfg(test)]
mod limit_matrix {
use super::parse_limit;
#[test]
fn matches_probe_envelopes() {
assert_eq!(
parse_limit("abc").unwrap_err(),
"Expected an integer, got NaN\n at [\"limit\"]"
);
assert_eq!(
parse_limit("5.5").unwrap_err(),
"Expected an integer, got 5.5\n at [\"limit\"]"
);
assert_eq!(
parse_limit("-0.5").unwrap_err(),
"Expected an integer, got -0.5\n at [\"limit\"]"
);
assert_eq!(
parse_limit("-1").unwrap_err(),
"Expected a value greater than or equal to 0, got -1\n at [\"limit\"]"
);
assert_eq!(
parse_limit("999999999999999999999").unwrap_err(),
"Expected an integer, got 1e+21\n at [\"limit\"]"
);
assert_eq!(
parse_limit("Infinity").unwrap_err(),
"Expected an integer, got Infinity\n at [\"limit\"]"
);
}
#[test]
fn accepts_integers_like_js_number() {
assert_eq!(parse_limit("+5"), Ok(5));
assert_eq!(parse_limit("1e3"), Ok(1000)); assert_eq!(parse_limit("0"), Ok(0));
assert_eq!(parse_limit("9007199254740991"), Ok(9007199254740991)); }
}
#[cfg(test)]
mod f1_blocking_tests {
use super::{AppState, Payloads, router, run_blocking};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
#[tokio::test]
async fn find_files_walk_runs_off_the_worker() {
use tower::ServiceExt as _;
let dir = std::env::temp_dir().join(format!(
"ocserve-m3-walk-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&dir).unwrap();
for i in 0..200 {
std::fs::write(dir.join(format!("f{i}.txt")), b"x").unwrap();
}
let mut payloads = Payloads::default();
payloads.config["directory"] = serde_json::Value::String(dir.to_string_lossy().to_string());
let st = AppState::with_payloads(None, payloads);
let handler_thread = std::thread::current().id();
let req = axum::http::Request::builder()
.uri("/find/file?query=f1&limit=10")
.body(axum::body::Body::empty())
.unwrap();
let resp = tokio::spawn(async move { router(st).oneshot(req).await })
.await
.expect("route task")
.expect("infallible");
assert_eq!(resp.status(), 200);
let walked_on = super::WALK_THREAD.lock().ok().and_then(|g| *g);
let walked_on = walked_on.expect("walk_files must record its thread");
assert_ne!(
walked_on, handler_thread,
"the walk ran on the async worker — blocking syscalls must go \
through spawn_blocking (M3)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn run_blocking_frees_the_worker() {
let ticks = Arc::new(AtomicU64::new(0));
let t = ticks.clone();
let ticker = tokio::spawn(async move {
let mut i = 0u64;
loop {
tokio::time::sleep(Duration::from_millis(10)).await;
i += 1;
t.store(i, Ordering::Relaxed);
}
});
tokio::time::sleep(Duration::from_millis(30)).await;
let before = ticks.load(Ordering::Relaxed);
assert!(before > 0, "ticker must be running before the probe");
let probe = tokio::spawn(async move {
run_blocking(std::path::Path::new("/nonexistent/f1-probe"), |_db| {
std::thread::sleep(Duration::from_millis(250));
Ok::<(), anyhow::Error>(())
})
.await
.expect("probe op");
});
probe.await.expect("probe join");
let after = ticks.load(Ordering::Relaxed);
ticker.abort();
assert!(
after > before,
"blocking call parked the worker: before={before} after={after}"
);
}
}
#[cfg(test)]
mod f5_wire {
use super::*;
#[test]
fn wire_bytes_are_serde_identical() {
let st = AppState::new();
let p = st.payloads.read();
let w = st.wire.read();
for (k, v) in [
("config", &p.config),
("config_providers", &p.config_providers),
("provider", &p.provider),
("console", &p.console),
("capabilities", &p.capabilities),
] {
let cached = w.get(k).unwrap_or_else(|| panic!("missing wire key {k}"));
let want = serde_json::to_vec(v).unwrap();
assert_eq!(
&cached.identity[..],
&want[..],
"wire bytes for {k} must equal Value serialization"
);
}
for (k, v) in [("agent", &p.agent), ("command", &p.command)] {
let cached = w.get(k).unwrap_or_else(|| panic!("missing wire key {k}"));
let want = serde_json::to_vec(v).unwrap();
assert_eq!(&cached.identity[..], &want[..]);
}
}
#[test]
fn wire_off_disables_cache() {
let empty = rebuild_wire(
&Payloads::default(),
true,
&json!({"directory": "/", "project": {"id": "global", "directory": "/"}}),
);
assert!(empty.is_empty(), "off => nothing serialized");
let st = AppState::new();
st.wire_off
.store(true, std::sync::atomic::Ordering::Relaxed);
assert!(wire_json(&st, "config").is_none(), "off => Value path");
}
#[test]
fn v2_metadata_wire_matches_builders() {
for (catalog, name) in [
(
serde_json::json!({"deepseek": {"name": "DeepSeek"}}),
"with-catalog",
),
(serde_json::json!({}), "empty-catalog"),
] {
let payloads = Payloads {
provider: serde_json::json!({
"connected": ["opencode", "deepseek"],
"all": [
{"id": "opencode", "name": "opencode", "models": {"m": {"id": "m", "providerID": "opencode", "name": "M"}}},
{"id": "deepseek", "name": "DeepSeek", "models": {"d": {"id": "d", "providerID": "deepseek", "name": "D"}}},
]
}),
config: serde_json::json!({"provider": {"entrim": {"name": "Entrim"}}}),
catalog: std::sync::Arc::new(catalog),
..Default::default()
};
let location = json!({"directory": "/", "project": {"id": "global", "directory": "/"}});
let wire = rebuild_wire(&payloads, false, &location);
for (key, want) in [
("api_model", crate::v2::models_body(&payloads, &location)),
(
"api_provider",
crate::v2::providers_body(&payloads, &location),
),
(
"api_integration",
crate::v2::integrations_body(&payloads, &location),
),
] {
let e = wire
.get(key)
.unwrap_or_else(|| panic!("[{name}] missing wire key {key}"));
assert_eq!(
&e.identity[..],
&serde_json::to_vec(&want).unwrap()[..],
"[{name}] {key} wire bytes must equal the pure builder"
);
}
}
}
}
#[cfg(test)]
mod doc_contract {
#[test]
fn spec_is_valid_and_complete() {
let d: serde_json::Value =
serde_json::from_str(crate::doc::SPEC_JSON).expect("vendored spec must be valid JSON");
assert_eq!(d["openapi"], "3.1.0");
let paths = d["paths"].as_object().expect("paths object");
assert_eq!(paths.len(), 162, "frozen 1.18.31 has 162 paths");
let ops: usize = paths
.values()
.map(|v| {
v.as_object()
.unwrap()
.keys()
.filter(|k| matches!(k.as_str(), "get" | "post" | "put" | "delete" | "patch"))
.count()
})
.sum();
assert_eq!(ops, 188, "frozen 1.18.31 has 188 operations");
assert!(d["components"]["schemas"].as_object().unwrap().len() >= 400);
}
}