Files
mnote/rust/crates/mnote-web/src/routes/hermes_client.rs
T
lix-2026 bb2f190f50 feat: stabilize page AI ACP runtimes
实现并稳定页面 AI 的 ACP Hermes / ACP Reasonix 运行路径。

主要内容:

- 分离 profile 与 acpRuntime,ACP Hermes 按所选 Hermes profile 启动并注入 provider key。

- 修复 Reasonix ACP wrapper 的 API key 读取、ToolRegistry 注册、LoopEvent role 映射和 reasoning/final 分流。

- 修复 ACP agent_thought_chunk 被 untagged enum 误解析为 message.delta 的问题,补充 thought 相关单测。

- 补充页面 AI 浏览器验证 skill 证据到 7-15 设计稿,并记录严格验收标准。

- 同步提交当前仓库中已存在的 rust-web / Hermes tools / SSE / bug 文档相关改动。

验证:

- node --check scripts/reasonix-acp-wrapper.mjs

- cargo test -p mnote-web acp -- --nocapture

- 页面 AI ACP 浏览器验证:tmp/page-ai-acp-browser-UAYwyM/
2026-05-17 20:11:39 +08:00

4422 lines
157 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use crate::acp_types::ContentBlock;
use crate::app::AppState;
use crate::context::RequestContext;
use crate::error::WebError;
use crate::hermes_tools::manifest;
use crate::transport::convex::execute_convex_query_by_name;
use axum::body::Body;
use axum::extract::{Extension, Path, Query, State};
use axum::http::{header, HeaderMap, HeaderName, HeaderValue, StatusCode};
use axum::response::Response;
use axum::Json;
use futures_util::TryStreamExt;
use serde::Deserialize;
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet, VecDeque};
use std::fs;
use std::path::{Path as FsPath, PathBuf};
use std::process::Command;
use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast;
use tracing::{info, warn};
const HEADER_MNOTE_WEB_OWNER: &str = "x-mnote-web-owner";
const HEADER_HERMES_CLIENT_OWNER: &str = "x-mnote-hermes-client-owner";
static HERMES_RUNTIME_REGISTRY: LazyLock<Mutex<HashMap<String, HermesRuntimeState>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
static HERMES_RUN_QUEUE: LazyLock<Mutex<HashMap<String, VecDeque<HermesQueuedRun>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
/// Store ACP run payloads keyed by run_id, so stream_events can read them.
static ACP_RUN_PAYLOADS: LazyLock<Mutex<HashMap<String, Value>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
static ACP_ACTIVE_RUNS: LazyLock<Mutex<HashMap<String, AcpActiveRun>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
#[derive(Debug, Clone)]
struct HermesRuntimeState {
session_id: String,
run_id: String,
profile: String,
document_id: String,
trace_id: String,
status: String,
started_at: u128,
last_event_at: u128,
last_event: Option<String>,
last_tool_name: Option<String>,
last_tool_call_id: Option<String>,
last_audit_id: Option<String>,
}
#[derive(Debug, Clone)]
struct HermesRunRegistration {
session_id: String,
profile: String,
document_id: String,
trace_id: String,
}
#[derive(Debug, Clone)]
struct HermesQueuedRun {
queue_id: String,
session_id: String,
profile: String,
document_id: String,
trace_id: String,
actor_id: String,
actor_type: String,
input: String,
context_summary: Value,
queued_at: u128,
}
#[derive(Clone)]
struct AcpActiveRun {
manager: Arc<crate::acp_session_manager::AcpSessionManager>,
mnote_session_id: String,
acp_session_id: String,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateSessionRequest {
workspace_id: Option<String>,
document_id: Option<String>,
trace_id: Option<String>,
title: Option<String>,
profile: Option<String>,
}
pub async fn list_sessions(
Extension(context): Extension<RequestContext>,
Query(query): Query<HashMap<String, String>>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let profile = query
.get("profile")
.map(String::as_str)
.unwrap_or("default");
let Some(upstream) = configured_upstream_for_profile(profile) else {
return hermes_unconfigured(&context);
};
let mut path = "/api/hermes/sessions".to_string();
if !query.is_empty() {
let params = query
.iter()
.map(|(key, value)| format!("{}={}", url_escape(key), url_escape(value)))
.collect::<Vec<_>>()
.join("&");
path.push('?');
path.push_str(&params);
}
proxy_json(
&context,
reqwest::Method::GET,
&upstream,
&path,
None,
Some(profile),
)
.await
}
pub async fn create_session(
Extension(context): Extension<RequestContext>,
Json(payload): Json<CreateSessionRequest>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let trace_id = payload
.trace_id
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| context.trace.trace_id.clone());
let document_id = payload
.document_id
.as_deref()
.filter(|value| !value.trim().is_empty())
.unwrap_or("current");
let session_id = stable_session_id(document_id, &trace_id);
let profile = payload
.profile
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("default");
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"sessionId": session_id,
"workspaceId": payload.workspace_id,
"documentId": payload.document_id,
"profile": profile,
"title": payload.title.unwrap_or_else(|| "当前页问答".into()),
"traceId": trace_id,
"persistence": "hermes_on_first_run"
})),
))
}
pub async fn list_profiles(
Extension(context): Extension<RequestContext>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(list_profiles_payload()),
))
}
pub async fn gateway_health(
Extension(context): Extension<RequestContext>,
Query(query): Query<HashMap<String, String>>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let profile = query
.get("profile")
.map(String::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(active_profile_name)
.unwrap_or_else(|| "default".into());
let upstream = configured_upstream_for_profile(&profile);
let profile_status = profile_gateway_status(&profile);
let mut suggestions = profile_status
.get("suggestions")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
if upstream.is_none() {
suggestions.push(Value::String(
"未配置 MNOTE_WEB_HERMES_UPSTREAM_URL;请先启动 Hermes gateway,并让 mnote-web 指向该 gateway。"
.into(),
));
}
let mut gateway = json!({
"configured": upstream.is_some(),
"upstream": upstream,
"ok": false,
"status": if upstream.is_some() { "checking" } else { "unconfigured" }
});
if let Some(upstream_url) = gateway["upstream"].as_str().map(ToOwned::to_owned) {
let probe =
probe_gateway_health(&upstream_url, configured_api_key_for_profile(&profile)).await;
gateway["ok"] = Value::Bool(probe.ok);
gateway["status"] = Value::String(probe.status);
gateway["httpStatus"] = probe.http_status.map(Value::from).unwrap_or(Value::Null);
gateway["path"] = probe.path.map(Value::from).unwrap_or(Value::Null);
gateway["message"] = probe.message.map(Value::from).unwrap_or(Value::Null);
if !probe.ok {
suggestions.extend(
hermes_settings_suggestions(
gateway["httpStatus"].as_u64(),
gateway["message"].as_str(),
)
.into_iter()
.map(Value::String),
);
}
}
let ok = gateway["ok"].as_bool().unwrap_or(false)
&& profile_status["modelConfigured"].as_bool().unwrap_or(false)
&& profile_status["apiKeyConfigured"]
.as_bool()
.unwrap_or(false);
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": ok,
"traceId": context.trace.trace_id,
"profile": profile_status,
"gateway": gateway,
"suggestions": unique_string_values(suggestions)
})),
))
}
pub async fn get_profile(
Extension(context): Extension<RequestContext>,
Path(profile_name): Path<String>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"profile": profile_detail_payload(&profile_name)
})),
))
}
pub async fn switch_active_profile(
Extension(context): Extension<RequestContext>,
Json(payload): Json<Value>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let name = payload
.get("name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 profile name")
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
switch_active_profile_local(name).map_err(|error| {
WebError::bad_gateway_code(
"hermes_client_profile_switch_failed",
format!("切换 Hermes profile 失败: {error}"),
)
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"profile": profile_detail_payload(name)
})),
))
}
pub async fn get_profile_memory(
Extension(context): Extension<RequestContext>,
Query(query): Query<HashMap<String, String>>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let fallback_profile = active_profile_name().unwrap_or_else(|| "default".into());
let profile = query
.get("profile")
.map(String::as_str)
.unwrap_or(fallback_profile.as_str());
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(profile_memory_payload(profile)),
))
}
pub async fn save_profile_memory(
Extension(context): Extension<RequestContext>,
Json(payload): Json<Value>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let fallback_profile = active_profile_name().unwrap_or_else(|| "default".into());
let profile = payload
.get("profile")
.and_then(Value::as_str)
.unwrap_or(fallback_profile.as_str());
let section = payload
.get("section")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| matches!(*value, "memory" | "user" | "soul"))
.ok_or_else(|| {
WebError::bad_request_code(
"hermes_client_bad_request",
"section 必须是 memory、user 或 soul",
)
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let content = payload
.get("content")
.and_then(Value::as_str)
.unwrap_or_default();
save_profile_memory_local(profile, section, content).map_err(|error| {
WebError::bad_gateway_code(
"hermes_client_memory_save_failed",
format!("保存 Hermes profile memory 失败: {error}"),
)
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({"ok": true})),
))
}
pub async fn list_skills(
Extension(context): Extension<RequestContext>,
Query(query): Query<HashMap<String, String>>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let fallback_profile = active_profile_name().unwrap_or_else(|| "default".into());
let profile = query
.get("profile")
.map(String::as_str)
.unwrap_or(fallback_profile.as_str());
let runtime = query.get("runtime").map(String::as_str).unwrap_or(profile);
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(if runtime == "reasonix" {
reasonix_skills_payload()
} else {
skills_payload(profile)
}),
))
}
pub async fn toggle_skill(
Extension(context): Extension<RequestContext>,
Json(payload): Json<Value>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let name = payload
.get("name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 skill name")
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let enabled = payload
.get("enabled")
.and_then(Value::as_bool)
.ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 enabled")
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let fallback_profile = active_profile_name().unwrap_or_else(|| "default".into());
let profile = payload
.get("profile")
.and_then(Value::as_str)
.unwrap_or(fallback_profile.as_str());
set_skill_enabled(profile, name, enabled).map_err(|error| {
WebError::bad_gateway_code(
"hermes_client_skill_toggle_failed",
format!("更新 Hermes skill 设置失败: {error}"),
)
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({"ok": true})),
))
}
pub async fn toggle_tool(
Extension(context): Extension<RequestContext>,
Json(payload): Json<Value>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let name = payload
.get("name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 tool name")
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let enabled = payload
.get("enabled")
.and_then(Value::as_bool)
.ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 enabled")
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let fallback_profile = active_profile_name().unwrap_or_else(|| "default".into());
let profile = payload
.get("profile")
.and_then(Value::as_str)
.unwrap_or(fallback_profile.as_str());
set_mnote_tool_enabled(profile, name, enabled).map_err(|error| {
WebError::bad_gateway_code(
"hermes_client_tool_toggle_failed",
format!("更新 mnote tool 设置失败: {error}"),
)
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({"ok": true})),
))
}
pub async fn get_session(
Extension(context): Extension<RequestContext>,
Path(session_id): Path<String>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let runtime = runtime_state_for_session(&session_id);
if let Some(session) = load_session_from_hermes_cli(&session_id).await {
return Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"traceId": context.trace.trace_id,
"sessionId": session_id,
"session": session,
"runtime": runtime
})),
));
}
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"traceId": context.trace.trace_id,
"sessionId": session_id,
"session": {
"sessionId": session_id,
"messages": []
},
"runtime": runtime
})),
))
}
pub async fn resume_session(
Extension(context): Extension<RequestContext>,
Path(session_id): Path<String>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
get_session(Extension(context), Path(session_id)).await
}
async fn load_session_from_hermes_cli(session_id: &str) -> Option<Value> {
let session_id = session_id.to_string();
tokio::task::spawn_blocking(move || {
let hermes_bin = std::env::var("MNOTE_WEB_HERMES_BIN")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "hermes".into());
let output = Command::new(hermes_bin)
.args(["sessions", "export", "--session-id", &session_id, "-"])
.output()
.ok()?;
if !output.status.success() {
return None;
}
let stdout = String::from_utf8(output.stdout).ok()?;
stdout
.lines()
.find_map(|line| serde_json::from_str::<Value>(line).ok())
})
.await
.ok()
.flatten()
}
pub async fn create_run(
State(state): State<AppState>,
Extension(context): Extension<RequestContext>,
Json(mut payload): Json<Value>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let (actor_id, actor_type) = resolve_run_actor(&state, &context).await;
stamp_run_actor(&mut payload, &actor_id, &actor_type);
let registration = run_registration_from_payload(&context, &payload);
// ACP path: skip the HTTP proxy, just register and return run info
if acp_runtime_for_payload(&payload, &registration.profile).is_some() {
let run_id = payload
.get("runId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(ToOwned::to_owned)
.unwrap_or_else(|| new_acp_run_id(&registration));
payload["runId"] = Value::String(run_id.clone());
let runtime_state = register_acp_runtime(&registration, &run_id);
// Store payload for stream_events to use
ACP_RUN_PAYLOADS
.lock()
.expect("acp run payloads")
.insert(run_id.clone(), payload.clone());
let response = json!({
"ok": true,
"runId": run_id,
"sessionId": registration.session_id,
"profile": registration.profile,
"traceId": context.trace.trace_id,
"runtime": runtime_state,
});
return Ok((StatusCode::OK, stamp_client_headers(), Json(response)));
}
if session_has_active_run(&registration.session_id) {
let queued = enqueue_run(&context, &registration, &payload)?;
return Ok((StatusCode::ACCEPTED, stamp_client_headers(), Json(queued)));
}
let Some(upstream) = configured_upstream_for_profile(&registration.profile) else {
return hermes_unconfigured(&context);
};
let upstream_body = build_run_upstream_body(&context, payload)?;
let result = proxy_json(
&context,
reqwest::Method::POST,
&upstream,
"/v1/runs",
Some(upstream_body),
Some(&registration.profile),
)
.await?;
if let Some(runtime) = register_runtime_from_create_run_response(&registration, &result.2 .0) {
let mut payload = result.2 .0;
payload["runtime"] = runtime;
Ok((result.0, result.1, Json(payload)))
} else {
Ok(result)
}
}
pub async fn cancel_queued_run(
Extension(context): Extension<RequestContext>,
Path((session_id, queue_id)): Path<(String, String)>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let cancelled = remove_queued_run(&session_id, &queue_id);
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"sessionId": session_id,
"queueId": queue_id,
"cancelled": cancelled,
"queueLength": queue_len_for_session(&session_id)
})),
))
}
/// ACP variant of stream_events: creates an ACP session, runs the prompt,
/// and returns an SSE stream of events.
async fn acp_stream_events(
state: AppState,
context: RequestContext,
run_id: &str,
profile: &str,
acp_runtime_name: &str,
) -> Result<Response, WebError> {
// Get the stored payload from create_run
let payload = ACP_RUN_PAYLOADS
.lock()
.expect("acp run payloads")
.remove(run_id)
.ok_or_else(|| {
WebError::bad_gateway_code(
"acp_run_payload_not_found",
format!("ACP run payload not found for run_id={run_id}"),
)
.with_context(&context)
})?;
// Build prompt from the message field (frontend sends "message", not "input")
let input = payload
.get("message")
.or_else(|| payload.get("input"))
.and_then(Value::as_str)
.unwrap_or("请读取当前文档内容");
let mut prompt_blocks = Vec::new();
if let Ok(upstream_body) = build_run_upstream_body(&context, payload.clone()) {
if let Some(instructions) = upstream_body.get("instructions").and_then(Value::as_str) {
prompt_blocks.push(ContentBlock::Text {
text: format!(
"以下是 mnote 页面 AI 的冻结上下文与工具约束,请在本轮回答中遵守:\n{instructions}"
),
});
}
}
prompt_blocks.push(ContentBlock::Text {
text: input.to_string(),
});
let runtime_name = acp_runtime_name;
// Ensure runtime is active; switch_to either activates it or returns existing
let client = if runtime_name == "hermes" {
let hermes_bin = std::env::var("MNOTE_WEB_HERMES_BIN")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "hermes".into());
let mut config =
crate::acp_runtime::AcpRuntimeConfig::hermes(Some(&hermes_bin), Some(profile));
config.env = acp_hermes_env_for_profile(profile);
state.acp_runtime.switch_to_config(config).await
} else {
state.acp_runtime.switch_to(runtime_name).await
}
.map_err(|e| {
WebError::bad_gateway_code(
"acp_runtime_switch_failed",
format!("Failed to activate ACP runtime '{runtime_name}': {e}"),
)
.with_context(&context)
})?;
// Create session manager and start the prompt
let mgr = Arc::new(crate::acp_session_manager::AcpSessionManager::new(client));
let (event_tx, _event_rx) = broadcast::channel(256);
let event_tx_clone = event_tx.clone();
mgr.on_event(move |event| {
if let Some(sse) = crate::acp_bridge::acp_event_to_sse(event) {
let _ = event_tx_clone.send(sse);
}
});
let acp_session_id = mgr.create_session(None, None).await.map_err(|e| {
WebError::bad_gateway_code(
"acp_session_create_failed",
format!("ACP session creation failed: {e}"),
)
.with_context(&context)
})?;
let mnote_session_id = session_id_for_run(run_id).unwrap_or_else(|| run_id.to_string());
ACP_ACTIVE_RUNS.lock().expect("acp active runs").insert(
run_id.to_string(),
AcpActiveRun {
manager: Arc::clone(&mgr),
mnote_session_id,
acp_session_id,
},
);
// Build SSE response from event channel FIRST (before running prompt),
// so that if the prompt fails quickly, events are not lost.
use tokio::sync::mpsc;
let (tx, rx) = mpsc::channel::<Result<axum::body::Bytes, std::convert::Infallible>>(256);
let mut event_rx = event_tx.subscribe();
tokio::spawn(async move {
loop {
match event_rx.recv().await {
Ok(event) => {
let json_str =
serde_json::to_string(&event.data).unwrap_or_else(|_| "{}".into());
let bytes = axum::body::Bytes::from(format!(
"event: {}\ndata: {}\n\n",
event.event, json_str
));
if tx.send(Ok(bytes)).await.is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
warn!("ACP SSE lagged: {n} events dropped");
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
// Run prompt in background
let run_id_owned = run_id.to_string();
let mgr_clone = Arc::clone(&mgr);
let event_tx_prompt = event_tx.clone();
update_runtime_by_run_id(&run_id_owned, "running", Some("acp.prompt.started"), None);
tokio::spawn(async move {
match mgr_clone.run_prompt(prompt_blocks).await {
Ok(result) => {
info!("ACP prompt completed: stop_reason={:?}", result.stop_reason);
let _ = event_tx_prompt.send(crate::acp_bridge::SseEvent {
event: "run.completed".into(),
data: json!({
"stopReason": format!("{:?}", result.stop_reason),
}),
});
}
Err(e) => {
warn!("ACP prompt failed: {e}");
let _ = event_tx_prompt.send(crate::acp_bridge::SseEvent {
event: "run.failed".into(),
data: json!({ "error": e.to_string() }),
});
}
}
mgr_clone.close().await;
ACP_ACTIVE_RUNS
.lock()
.expect("acp active runs")
.remove(&run_id_owned);
if !matches!(
runtime_status_for_run(&run_id_owned).as_deref(),
Some("aborting" | "aborted")
) {
update_runtime_by_run_id(&run_id_owned, "completed", Some("acp.prompt.done"), None);
}
});
let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
let mut response = Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "text/event-stream; charset=utf-8")
.header(header::CACHE_CONTROL, "no-cache, no-transform")
.header("x-accel-buffering", "no")
.body(Body::from_stream(stream))
.map_err(|e| {
WebError::internal(format!("SSE response build failed: {e}")).with_context(&context)
})?;
stamp_client_headers_into(response.headers_mut());
Ok(response)
}
pub async fn stream_events(
State(state): State<AppState>,
Extension(context): Extension<RequestContext>,
Path(run_id): Path<String>,
) -> Result<Response, WebError> {
ensure_authenticated(&context)?;
let profile = profile_for_run(&run_id).unwrap_or_else(|| "default".into());
// ACP path: start AcpRunBridge and return SSE stream
if let Some(acp_runtime_name) = acp_runtime_for_run(&run_id, &profile) {
return acp_stream_events(state, context, &run_id, &profile, &acp_runtime_name).await;
}
let Some(upstream) = configured_upstream_for_profile(&profile) else {
return Err(hermes_unconfigured_error(&context));
};
let url = upstream_url(
&upstream,
&format!("/v1/runs/{}/events", url_escape(&run_id)),
)?;
let mut request = reqwest::Client::builder()
.timeout(Duration::from_secs(1800))
.build()
.map_err(|error| {
WebError::internal(format!("Hermes client 构造失败: {error}")).with_context(&context)
})?
.get(url);
if let Some(api_key) = configured_api_key_for_profile(&profile) {
request = request.bearer_auth(api_key);
}
let upstream_response = request.send().await.map_err(|error| {
WebError::bad_gateway_code(
"hermes_client_upstream_unavailable",
format!("Hermes events upstream 连接失败: {error}"),
)
.with_context(&context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
if !upstream_response.status().is_success() {
let status = upstream_response.status();
let text = upstream_response.text().await.unwrap_or_default();
return Err(upstream_error(&context, status, text));
}
let run_id_for_stream = run_id.clone();
let context_for_queue = context.clone();
update_runtime_by_run_id(&run_id, "running", Some("events.connected"), None);
let stream = upstream_response
.bytes_stream()
.map_ok(move |chunk| {
let text = String::from_utf8_lossy(&chunk);
let (normalized, terminal_sessions) =
normalize_sse_chunk(&run_id_for_stream, text.as_ref());
for session_id in terminal_sessions {
let context = context_for_queue.clone();
tokio::spawn(async move {
start_next_queued_run(context, session_id).await;
});
}
axum::body::Bytes::from(normalized)
})
.map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Hermes events stream 读取失败: {error}"),
)
});
let mut response = Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "text/event-stream; charset=utf-8")
.header(header::CACHE_CONTROL, "no-cache, no-transform")
.header("x-accel-buffering", "no")
.body(Body::from_stream(stream))
.map_err(|error| WebError::internal(format!("Hermes events 响应构造失败: {error}")))?;
stamp_client_headers_into(response.headers_mut());
Ok(response)
}
pub async fn abort_run(
Extension(context): Extension<RequestContext>,
Path(run_id): Path<String>,
Json(payload): Json<Value>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let profile = profile_for_run(&run_id).unwrap_or_else(|| "default".into());
if acp_runtime_for_run(&run_id, &profile).is_some() {
let active = ACP_ACTIVE_RUNS
.lock()
.expect("acp active runs")
.remove(&run_id);
update_runtime_by_run_id(&run_id, "aborting", Some("abort.started"), None);
let Some(active) = active else {
update_runtime_by_run_id(&run_id, "aborted", Some("abort.missing_active_run"), None);
return Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"runId": run_id,
"status": "aborted",
"runtime": runtime_state_for_run(&run_id).unwrap_or(Value::Null),
"events": [
{"event": "abort.started", "runId": run_id},
{"event": "abort.completed", "runId": run_id, "note": "active ACP run was already finished or missing"}
]
})),
));
};
active.manager.cancel().await.map_err(|error| {
WebError::bad_gateway_code("acp_abort_failed", format!("ACP abort failed: {error}"))
.with_context(&context)
})?;
update_runtime_by_run_id(&run_id, "aborted", Some("abort.completed"), None);
return Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"runId": run_id,
"sessionId": active.mnote_session_id,
"acpSessionId": active.acp_session_id,
"status": "aborted",
"runtime": runtime_state_for_run(&run_id).unwrap_or(Value::Null),
"events": [
{"event": "abort.started", "runId": run_id},
{"event": "abort.completed", "runId": run_id}
]
})),
));
}
let Some(upstream) = configured_upstream_for_profile(&profile) else {
return hermes_unconfigured(&context);
};
let queued_session_id = session_id_for_run(&run_id);
update_runtime_by_run_id(&run_id, "aborting", Some("abort.started"), None);
let result = proxy_json(
&context,
reqwest::Method::POST,
&upstream,
&format!("/v1/runs/{}/stop", url_escape(&run_id)),
Some(payload),
Some(&profile),
)
.await;
match result {
Ok((status, headers, Json(mut payload))) => {
update_runtime_by_run_id(&run_id, "aborted", Some("abort.completed"), None);
payload["runtime"] = runtime_state_for_run(&run_id).unwrap_or(Value::Null);
payload["events"] = json!([
{"event": "abort.started", "runId": run_id},
{"event": "abort.completed", "runId": run_id}
]);
if let Some(session_id) = queued_session_id {
start_next_queued_run(context.clone(), session_id).await;
}
Ok((status, headers, Json(payload)))
}
Err(error) => {
update_runtime_by_run_id(&run_id, "failed", Some("abort.failed"), None);
Err(error)
}
}
}
pub async fn list_models(
Extension(context): Extension<RequestContext>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let Some(upstream) = configured_upstream() else {
return hermes_unconfigured(&context);
};
proxy_json(
&context,
reqwest::Method::GET,
&upstream,
"/v1/models",
None,
None,
)
.await
}
pub async fn list_tools(
Extension(context): Extension<RequestContext>,
Query(query): Query<HashMap<String, String>>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
ensure_authenticated(&context)?;
let fallback_profile = active_profile_name().unwrap_or_else(|| "default".into());
let profile = query
.get("profile")
.map(String::as_str)
.unwrap_or(fallback_profile.as_str());
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(json!({
"ok": true,
"traceId": context.trace.trace_id,
"profile": profile,
"tools": mnote_tools_payload(profile)
})),
))
}
fn hermes_bin() -> String {
std::env::var("MNOTE_WEB_HERMES_BIN")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "hermes".into())
}
fn hermes_home() -> PathBuf {
std::env::var("HERMES_HOME")
.ok()
.filter(|value| !value.trim().is_empty())
.map(PathBuf::from)
.or_else(|| {
std::env::var("HOME")
.ok()
.map(|home| PathBuf::from(home).join(".hermes"))
})
.unwrap_or_else(|| PathBuf::from(".hermes"))
}
pub(crate) fn active_profile_name() -> Option<String> {
fs::read_to_string(hermes_home().join("active_profile"))
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
fn profile_home(profile: &str) -> PathBuf {
let home = hermes_home();
let profile = profile.trim();
if profile.is_empty() || profile == "default" {
return home;
}
let candidate = home.join("profiles").join(profile);
if candidate.exists() {
candidate
} else {
home
}
}
fn profile_config_path(profile: &str) -> PathBuf {
profile_home(profile).join("config.yaml")
}
fn parse_profile_list(stdout: &str) -> Vec<Value> {
let active = active_profile_name().unwrap_or_else(|| "default".into());
stdout
.replace("\r\n", "\n")
.replace('\r', "\n")
.lines()
.filter_map(|line| {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with("Profile") || trimmed.starts_with('─') {
return None;
}
let is_active = trimmed.starts_with('◆');
let without_marker = trimmed.trim_start_matches('◆').trim();
let parts = without_marker
.split_whitespace()
.map(str::to_string)
.collect::<Vec<_>>();
if parts.is_empty() {
return None;
}
let name = parts[0].clone();
let model = parts
.get(1)
.filter(|value| value.as_str() != "—")
.cloned()
.unwrap_or_default();
let gateway = parts
.get(2)
.filter(|value| value.as_str() != "—")
.cloned()
.unwrap_or_default();
let alias = parts
.get(3)
.filter(|value| value.as_str() != "—")
.cloned()
.unwrap_or_default();
Some(json!({
"name": name,
"active": name == active || (active == "default" && is_active),
"model": model,
"gateway": gateway,
"alias": alias
}))
})
.collect()
}
fn fallback_profiles() -> Vec<Value> {
let active = active_profile_name().unwrap_or_else(|| "default".into());
let default_status = profile_gateway_status("default");
let mut profiles = vec![json!({
"name": "default",
"active": active == "default",
"model": default_status["modelDefault"].as_str().unwrap_or_default(),
"gateway": default_status["gateway"].as_str().unwrap_or_default(),
"provider": default_status["provider"].as_str().unwrap_or_default(),
"alias": ""
})];
let profiles_dir = hermes_home().join("profiles");
if let Ok(entries) = fs::read_dir(profiles_dir) {
let mut names = entries
.filter_map(Result::ok)
.filter(|entry| entry.file_type().map(|kind| kind.is_dir()).unwrap_or(false))
.filter_map(|entry| entry.file_name().into_string().ok())
.collect::<Vec<_>>();
names.sort();
profiles.extend(names.into_iter().map(|name| {
let status = profile_gateway_status(&name);
json!({
"name": name,
"active": name == active,
"model": status["modelDefault"].as_str().unwrap_or_default(),
"gateway": status["gateway"].as_str().unwrap_or_default(),
"provider": status["provider"].as_str().unwrap_or_default(),
"alias": ""
})
}));
}
profiles
}
fn list_profiles_payload() -> Value {
let profiles = Command::new(hermes_bin())
.args(["profile", "list"])
.output()
.ok()
.filter(|output| output.status.success())
.and_then(|output| String::from_utf8(output.stdout).ok())
.map(|stdout| parse_profile_list(&stdout))
.filter(|profiles| !profiles.is_empty())
.unwrap_or_else(fallback_profiles);
let reasonix_model = std::env::var("REASONIX_MODEL").unwrap_or_else(|_| "deepseek-chat".into());
let reasonix_preset = std::env::var("REASONIX_PRESET").unwrap_or_else(|_| "auto".into());
let reasonix_has_key = std::env::var("DEEPSEEK_API_KEY").is_ok();
json!({
"ok": true,
"profiles": profiles,
"acpRuntimes": json!([
{
"name": "hermes",
"title": "ACP · Hermes",
"description": "通过 ACP 协议直连 Hermes agent runtime · model/default 来自 Hermes profile"
},
{
"name": "reasonix",
"title": "ACP · Reasonix",
"description": "通过 ACP 协议直连 ReasonixDeepSeek 缓存优先)",
"model": reasonix_model,
"preset": reasonix_preset,
"apiKeyConfigured": reasonix_has_key,
"version": "0.43.0"
}
])
})
}
fn profile_detail_payload(profile: &str) -> Value {
let dir = profile_home(profile);
let gateway_status = profile_gateway_status(profile);
json!({
"name": if profile.trim().is_empty() { "default" } else { profile.trim() },
"path": dir.to_string_lossy(),
"model": gateway_status["modelDefault"].as_str().unwrap_or_default(),
"provider": gateway_status["provider"].as_str().unwrap_or_default(),
"gateway": gateway_status["gateway"].as_str().unwrap_or_default(),
"modelConfigured": gateway_status["modelConfigured"].as_bool().unwrap_or(false),
"apiKeyConfigured": gateway_status["apiKeyConfigured"].as_bool().unwrap_or(false),
"suggestions": gateway_status["suggestions"].clone(),
"skills": skills_payload(profile)["categories"].as_array().map(Vec::len).unwrap_or_default(),
"hasEnv": dir.join(".env").exists(),
"hasSoulMd": dir.join("SOUL.md").exists()
})
}
fn switch_active_profile_local(profile: &str) -> std::io::Result<()> {
let command_result = Command::new(hermes_bin())
.args(["profile", "use", profile])
.output();
if command_result
.as_ref()
.map(|output| output.status.success())
.unwrap_or(false)
{
return Ok(());
}
let home = hermes_home();
fs::create_dir_all(&home)?;
if profile != "default" {
fs::create_dir_all(home.join("profiles").join(profile))?;
}
fs::write(home.join("active_profile"), profile.as_bytes())
}
fn file_text(path: &FsPath) -> String {
fs::read_to_string(path).unwrap_or_default()
}
fn file_mtime_ms(path: &FsPath) -> Option<i128> {
let modified = fs::metadata(path).ok()?.modified().ok()?;
let duration = modified.duration_since(std::time::UNIX_EPOCH).ok()?;
Some(duration.as_millis() as i128)
}
fn profile_memory_payload(profile: &str) -> Value {
let dir = profile_home(profile);
let memory_path = dir.join("memories").join("MEMORY.md");
let user_path = dir.join("memories").join("USER.md");
let soul_path = dir.join("SOUL.md");
json!({
"ok": true,
"memory": file_text(&memory_path),
"user": file_text(&user_path),
"soul": file_text(&soul_path),
"memory_mtime": file_mtime_ms(&memory_path),
"user_mtime": file_mtime_ms(&user_path),
"soul_mtime": file_mtime_ms(&soul_path)
})
}
fn save_profile_memory_local(profile: &str, section: &str, content: &str) -> std::io::Result<()> {
let dir = profile_home(profile);
let path = match section {
"soul" => dir.join("SOUL.md"),
"user" => dir.join("memories").join("USER.md"),
_ => dir.join("memories").join("MEMORY.md"),
};
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
fs::write(path, content)
}
fn disabled_skills(profile: &str) -> Vec<String> {
let content = fs::read_to_string(profile_config_path(profile)).unwrap_or_default();
let mut disabled = Vec::new();
let mut in_skills = false;
let mut in_disabled = false;
for line in content.lines() {
let trimmed = line.trim();
if !line.starts_with(' ') && !trimmed.is_empty() {
in_skills = trimmed == "skills:";
in_disabled = false;
continue;
}
if in_skills && trimmed == "disabled:" {
in_disabled = true;
continue;
}
if in_disabled {
if let Some(value) = trimmed.strip_prefix("- ") {
disabled.push(
value
.trim()
.trim_matches('"')
.trim_matches('\'')
.to_string(),
);
} else if !trimmed.is_empty() && !trimmed.starts_with('#') {
in_disabled = false;
}
}
}
disabled
}
fn yaml_disabled_list(content: &str, path: &[&str]) -> Vec<String> {
let mut stack: Vec<(usize, String)> = Vec::new();
let mut values = Vec::new();
for raw_line in content.lines() {
let line = raw_line.trim_end_matches('\r');
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
let indent = line.chars().take_while(|ch| ch.is_whitespace()).count();
if let Some(value) = trimmed.strip_prefix("- ") {
let keys = stack
.iter()
.map(|(_, key)| key.as_str())
.collect::<Vec<_>>();
if keys == path {
let normalized = value.trim().trim_matches('"').trim_matches('\'');
if !normalized.is_empty() {
values.push(normalized.to_string());
}
}
continue;
}
while stack
.last()
.map(|(level, _)| *level >= indent)
.unwrap_or(false)
{
stack.pop();
}
let Some((key, _value)) = trimmed.split_once(':') else {
continue;
};
let key = key.trim().trim_matches('"').trim_matches('\'').to_string();
stack.push((indent, key));
}
values
}
pub(crate) fn disabled_mnote_tools(profile: &str) -> Vec<String> {
let content = fs::read_to_string(profile_config_path(profile)).unwrap_or_default();
yaml_disabled_list(&content, &["mnote", "tools", "disabled"])
}
pub(crate) fn is_mnote_tool_disabled(profile: &str, name: &str) -> bool {
disabled_mnote_tools(profile)
.iter()
.any(|item| item == name)
}
fn mnote_tools_payload(profile: &str) -> Vec<Value> {
let disabled = disabled_mnote_tools(profile);
manifest::manifest()
.get("tools")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(|tool| mnote_tool_entry(tool, &disabled))
.collect()
}
fn mnote_tool_entry(tool: &Value, disabled: &[String]) -> Option<Value> {
let name = tool.get("name").and_then(Value::as_str)?.trim();
if name.is_empty() {
return None;
}
let capability_scope = tool
.get("capabilityScope")
.and_then(Value::as_array)
.map(|items| {
items
.iter()
.filter_map(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(str::to_string)
.collect::<Vec<_>>()
})
.unwrap_or_default();
let disabled = disabled.iter().any(|item| item == name);
let status = if disabled {
"disabled"
} else {
tool.get("status")
.and_then(Value::as_str)
.unwrap_or("available")
};
Some(json!({
"name": name,
"description": tool.get("description").and_then(Value::as_str).unwrap_or_default(),
"scope": capability_scope.first().cloned().unwrap_or_else(|| "mnote".into()),
"capabilityScope": capability_scope,
"kind": if capability_scope.iter().any(|scope| scope.ends_with(".write")) { "write" } else { "read" },
"schemaVersion": tool.get("schemaVersion").and_then(Value::as_str).unwrap_or(manifest::TOOL_SCHEMA_VERSION),
"status": status,
"enabled": !disabled,
"unavailableReason": if disabled { "当前 Hermes profile 已关闭该 mnote tool" } else { "" }
}))
}
fn extract_skill_description(markdown: &str) -> String {
markdown
.lines()
.find_map(|line| {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
return None;
}
Some(
trimmed
.trim_start_matches("Description:")
.trim()
.to_string(),
)
})
.unwrap_or_default()
}
struct SkillCatalogMeta {
bundled: HashSet<String>,
hub_installed: HashSet<String>,
usage: HashMap<String, Value>,
}
impl SkillCatalogMeta {
fn read(skills_dir: &FsPath) -> Self {
Self {
bundled: read_bundled_skill_names(&skills_dir.join(".bundled_manifest")),
hub_installed: read_hub_installed_skill_names(
&skills_dir.join(".hub").join("lock.json"),
),
usage: read_skill_usage(&skills_dir.join(".usage.json")),
}
}
fn usage_for(&self, name: &str) -> Option<&Value> {
self.usage.get(name)
}
}
fn read_bundled_skill_names(path: &FsPath) -> HashSet<String> {
fs::read_to_string(path)
.ok()
.map(|content| {
content
.lines()
.filter_map(|line| {
line.split_once(':')
.map(|(name, _)| name.trim().to_string())
})
.filter(|name| !name.is_empty())
.collect()
})
.unwrap_or_default()
}
fn read_hub_installed_skill_names(path: &FsPath) -> HashSet<String> {
let Ok(content) = fs::read_to_string(path) else {
return HashSet::new();
};
let Ok(payload) = serde_json::from_str::<Value>(&content) else {
return HashSet::new();
};
payload
.get("installed")
.and_then(Value::as_object)
.map(|installed| {
installed
.iter()
.flat_map(|(name, entry)| {
let mut names = vec![name.to_string()];
if let Some(path) = entry.get("install_path").and_then(Value::as_str) {
if let Some(last) = path.rsplit('/').find(|part| !part.trim().is_empty()) {
names.push(last.to_string());
}
}
names
})
.collect()
})
.unwrap_or_default()
}
fn read_skill_usage(path: &FsPath) -> HashMap<String, Value> {
let Ok(content) = fs::read_to_string(path) else {
return HashMap::new();
};
serde_json::from_str::<Value>(&content)
.ok()
.and_then(|payload| payload.as_object().cloned())
.map(|items| items.into_iter().collect())
.unwrap_or_default()
}
fn skill_source(name: &str, meta: &SkillCatalogMeta) -> &'static str {
if meta.bundled.contains(name) {
"builtin"
} else if meta.hub_installed.contains(name) {
"hub"
} else {
"local"
}
}
fn skill_is_agent_generated(usage: Option<&Value>) -> bool {
usage
.and_then(Value::as_object)
.map(|record| {
record
.get("created_by")
.and_then(Value::as_str)
.map(|value| value == "agent")
.unwrap_or(false)
|| record
.get("agent_created")
.and_then(Value::as_bool)
.unwrap_or(false)
})
.unwrap_or(false)
}
fn skill_origin(source: &str, usage: Option<&Value>) -> &'static str {
if skill_is_agent_generated(usage) {
"generated"
} else if source == "builtin" {
"builtin"
} else if source == "hub" {
"installed"
} else {
"copied"
}
}
fn skill_entry(
name: String,
category: String,
path: PathBuf,
disabled: &[String],
meta: &SkillCatalogMeta,
) -> Option<Value> {
let content = fs::read_to_string(path).ok()?;
let source = skill_source(&name, meta);
let usage = meta.usage_for(&name);
Some(json!({
"name": name,
"description": extract_skill_description(&content),
"enabled": !disabled.iter().any(|item| item == &name),
"source": source,
"origin": skill_origin(source, usage),
"createdBy": usage.and_then(|record| record.get("created_by")).and_then(Value::as_str),
"createdAt": usage.and_then(|record| record.get("created_at")).and_then(Value::as_str),
"patchCount": usage.and_then(|record| record.get("patch_count")).and_then(Value::as_i64),
"category": category
}))
}
fn skills_payload(profile: &str) -> Value {
let skills_dir = hermes_home().join("skills");
let disabled = disabled_skills(profile);
let meta = SkillCatalogMeta::read(&skills_dir);
let mut categories = Vec::new();
let Ok(entries) = fs::read_dir(&skills_dir) else {
return json!({"ok": true, "categories": [], "archived": []});
};
let mut dirs = entries
.filter_map(Result::ok)
.filter(|entry| entry.file_type().map(|kind| kind.is_dir()).unwrap_or(false))
.filter(|entry| !entry.file_name().to_string_lossy().starts_with('.'))
.collect::<Vec<_>>();
dirs.sort_by_key(|entry| entry.file_name());
for entry in dirs {
let name = entry.file_name().to_string_lossy().to_string();
let dir = entry.path();
let direct_skill = dir.join("SKILL.md");
if direct_skill.exists() {
if let Some(skill) = skill_entry(name, "misc".into(), direct_skill, &disabled, &meta) {
categories.push(json!({
"name": "misc",
"description": "misc",
"skills": [skill]
}));
}
continue;
}
let mut skills = fs::read_dir(&dir)
.ok()
.into_iter()
.flat_map(|entries| entries.filter_map(Result::ok))
.filter(|child| child.file_type().map(|kind| kind.is_dir()).unwrap_or(false))
.filter_map(|child| {
let skill_name = child.file_name().to_string_lossy().to_string();
skill_entry(
skill_name,
name.clone(),
child.path().join("SKILL.md"),
&disabled,
&meta,
)
})
.collect::<Vec<_>>();
if !skills.is_empty() {
skills.sort_by_key(|skill| skill["name"].as_str().unwrap_or_default().to_string());
categories.push(json!({
"name": name,
"description": "",
"skills": skills
}));
}
}
json!({
"ok": true,
"categories": merge_misc_categories(categories),
"archived": []
})
}
fn reasonix_skill_roots() -> Vec<(PathBuf, &'static str)> {
let mut roots = Vec::new();
let project_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.ancestors()
.nth(3)
.unwrap_or(std::path::Path::new("/mnt/Data1T/mnote"))
.to_path_buf();
roots.push((project_root.join(".reasonix").join("skills"), "project"));
roots.push((project_root.join(".agents").join("skills"), "project"));
if let Ok(home) = std::env::var("HOME") {
let home = PathBuf::from(home);
roots.push((home.join(".reasonix").join("skills"), "global"));
roots.push((home.join(".agents").join("skills"), "global"));
}
roots
}
fn reasonix_skill_entry(path: PathBuf, stem: String, scope: &str) -> Option<Value> {
let content = fs::read_to_string(&path).ok()?;
let metadata = parse_reasonix_skill_frontmatter(&content);
let name = metadata
.get("name")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or(&stem)
.to_string();
Some(json!({
"name": name,
"description": metadata
.get("description")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(ToOwned::to_owned)
.unwrap_or_else(|| extract_skill_description(&content)),
"enabled": true,
"toggleable": false,
"source": "reasonix",
"origin": scope,
"category": scope,
"path": path.to_string_lossy(),
"scope": scope,
"runAs": metadata.get("runAs").cloned().unwrap_or(Value::Null),
"model": metadata.get("model").cloned().unwrap_or(Value::Null)
}))
}
fn parse_reasonix_skill_frontmatter(markdown: &str) -> serde_json::Map<String, Value> {
let mut metadata = serde_json::Map::new();
let mut lines = markdown.lines();
if lines.next().map(str::trim) != Some("---") {
return metadata;
}
for line in lines {
let trimmed = line.trim();
if trimmed == "---" {
break;
}
let Some((key, value)) = trimmed.split_once(':') else {
continue;
};
let normalized = value.trim().trim_matches('"').trim_matches('\'');
if !key.trim().is_empty() && !normalized.is_empty() {
metadata.insert(
key.trim().to_string(),
Value::String(normalized.to_string()),
);
}
}
metadata
}
fn reasonix_skills_payload() -> Value {
let mut seen = HashSet::new();
let mut by_scope: HashMap<String, Vec<Value>> = HashMap::new();
for (root, scope) in reasonix_skill_roots() {
let Ok(entries) = fs::read_dir(&root) else {
continue;
};
let mut entries = entries.filter_map(Result::ok).collect::<Vec<_>>();
entries.sort_by_key(|entry| entry.file_name());
for entry in entries {
let file_name = entry.file_name().to_string_lossy().to_string();
let path = entry.path();
let skill = if entry.file_type().map(|kind| kind.is_dir()).unwrap_or(false) {
reasonix_skill_entry(path.join("SKILL.md"), file_name, scope)
} else if entry
.file_type()
.map(|kind| kind.is_file())
.unwrap_or(false)
&& file_name.ends_with(".md")
{
let stem = file_name.trim_end_matches(".md").to_string();
reasonix_skill_entry(path, stem, scope)
} else {
None
};
let Some(skill) = skill else {
continue;
};
let Some(name) = skill.get("name").and_then(Value::as_str) else {
continue;
};
if !seen.insert(name.to_string()) {
continue;
}
by_scope.entry(scope.to_string()).or_default().push(skill);
}
}
let mut categories = by_scope
.into_iter()
.map(|(scope, mut skills)| {
skills.sort_by_key(|skill| skill["name"].as_str().unwrap_or_default().to_string());
json!({
"name": scope,
"description": format!("Reasonix {scope} skills"),
"skills": skills
})
})
.collect::<Vec<_>>();
categories.sort_by_key(
|category| match category["name"].as_str().unwrap_or_default() {
"project" => 0,
"custom" => 1,
"global" => 2,
_ => 3,
},
);
json!({
"ok": true,
"runtime": "reasonix",
"categories": categories,
"archived": []
})
}
fn merge_misc_categories(categories: Vec<Value>) -> Vec<Value> {
let mut merged = Vec::new();
let mut misc = Vec::new();
for category in categories {
if category["name"].as_str() == Some("misc") {
if let Some(skills) = category["skills"].as_array() {
misc.extend(skills.iter().cloned());
}
} else {
merged.push(category);
}
}
if !misc.is_empty() {
misc.sort_by_key(|skill| skill["name"].as_str().unwrap_or_default().to_string());
merged.insert(
0,
json!({
"name": "misc",
"description": "misc",
"skills": misc
}),
);
}
merged
}
fn set_skill_enabled(profile: &str, name: &str, enabled: bool) -> std::io::Result<()> {
let path = profile_config_path(profile);
let mut disabled = disabled_skills(profile);
if enabled {
disabled.retain(|item| item != name);
} else if !disabled.iter().any(|item| item == name) {
disabled.push(name.to_string());
}
disabled.sort();
let existing = fs::read_to_string(&path).unwrap_or_default();
let mut kept = Vec::new();
let mut in_skills = false;
let mut in_disabled = false;
for line in existing.lines() {
let trimmed = line.trim();
if !line.starts_with(' ') && !trimmed.is_empty() {
in_skills = trimmed == "skills:";
in_disabled = false;
}
if in_skills && trimmed == "disabled:" {
in_disabled = true;
continue;
}
if in_disabled {
if trimmed.starts_with("- ") || trimmed.is_empty() {
continue;
}
in_disabled = false;
}
kept.push(line.to_string());
}
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let mut output = kept.join("\n");
if !output.ends_with('\n') && !output.is_empty() {
output.push('\n');
}
output.push_str("skills:\n disabled:\n");
for skill in disabled {
output.push_str(" - ");
output.push_str(&skill);
output.push('\n');
}
fs::write(path, output)
}
fn set_mnote_tool_enabled(profile: &str, name: &str, enabled: bool) -> std::io::Result<()> {
let path = profile_config_path(profile);
let mut disabled = disabled_mnote_tools(profile);
if enabled {
disabled.retain(|item| item != name);
} else if !disabled.iter().any(|item| item == name) {
disabled.push(name.to_string());
}
disabled.sort();
let existing = fs::read_to_string(&path).unwrap_or_default();
let mut kept = Vec::new();
let mut stack: Vec<(usize, String)> = Vec::new();
let mut skipping_disabled_list = false;
let mut disabled_indent = 0usize;
for line in existing.lines() {
let trimmed = line.trim();
let indent = line.chars().take_while(|ch| ch.is_whitespace()).count();
if skipping_disabled_list {
if trimmed.starts_with("- ") && indent > disabled_indent {
continue;
}
skipping_disabled_list = false;
}
if trimmed.is_empty() || trimmed.starts_with('#') {
kept.push(line.to_string());
continue;
}
if !trimmed.starts_with("- ") {
while stack
.last()
.map(|(level, _)| *level >= indent)
.unwrap_or(false)
{
stack.pop();
}
if let Some((key, _value)) = trimmed.split_once(':') {
let key = key.trim().trim_matches('"').trim_matches('\'').to_string();
stack.push((indent, key));
let keys = stack
.iter()
.map(|(_, key)| key.as_str())
.collect::<Vec<_>>();
if keys == ["mnote", "tools", "disabled"] {
skipping_disabled_list = true;
disabled_indent = indent;
continue;
}
}
}
kept.push(line.to_string());
}
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let mut output = kept.join("\n");
if !output.ends_with('\n') && !output.is_empty() {
output.push('\n');
}
output.push_str("mnote:\n tools:\n disabled:\n");
for tool in disabled {
output.push_str(" - ");
output.push_str(&tool);
output.push('\n');
}
fs::write(path, output)
}
fn ensure_authenticated(context: &RequestContext) -> Result<(), WebError> {
let has_actor = context.auth.actor_id.trim() != "anonymous";
if has_actor || context.auth.authorization.is_some() || context.auth.cookie_header.is_some() {
return Ok(());
}
Err(WebError::new(
StatusCode::UNAUTHORIZED,
"hermes_client_unauthorized",
"页面 AI Hermes client 需要登录后访问",
)
.with_context(context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client"))
}
fn configured_upstream() -> Option<String> {
[
"MNOTE_WEB_HERMES_UPSTREAM_URL",
"MNOTE_HERMES_UPSTREAM_URL",
"MNOTE_HERMES_API_BASE_URL",
]
.into_iter()
.find_map(env_or_dotenv)
.map(|value| value.trim().trim_end_matches('/').to_string())
.filter(|value| !value.is_empty())
}
fn profile_config_value(profile: &str, path: &[&str]) -> Option<String> {
let content = fs::read_to_string(profile_config_path(profile)).ok()?;
yaml_path_value(&content, path)
}
fn profile_env_key(prefix: &str, profile: &str, suffix: &str) -> String {
let normalized = profile
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() {
ch.to_ascii_uppercase()
} else {
'_'
}
})
.collect::<String>();
format!("{prefix}_{normalized}_{suffix}")
}
/// Returns true if the given profile should use ACP instead of Hermes HTTP proxy.
///
/// Profile names `reasonix` always use ACP. Other profiles can be configured
/// via `MNOTE_WEB_<PROFILE>_RUNTIME_TYPE=acp`.
/// Default profiles (`default`, `hermes`) use the traditional Hermes HTTP proxy.
fn is_acp_profile(profile: &str) -> bool {
// ACP runtimes: "reasonix" and "hermes" both use ACP protocol when selected from the UI.
// The "reasonix" name is hardcoded; "hermes" as ACP is triggered by env var or UI selection.
if profile == "reasonix" || profile == "hermes" {
return true;
}
let env_key = profile_env_key("MNOTE_WEB", profile, "RUNTIME_TYPE");
env_or_dotenv(&env_key)
.map(|v| v.trim().to_lowercase() == "acp")
.unwrap_or(false)
}
fn acp_runtime_for_payload(payload: &Value, profile: &str) -> Option<String> {
if let Some(runtime) = payload
.get("acpRuntime")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
return Some(runtime.to_lowercase());
}
if is_acp_profile(profile) {
return Some(crate::acp_bridge::runtime_name_for_profile(profile).to_string());
}
None
}
fn acp_runtime_for_run(run_id: &str, profile: &str) -> Option<String> {
if let Some(payload) = ACP_RUN_PAYLOADS
.lock()
.expect("acp run payloads")
.get(run_id)
.cloned()
{
return acp_runtime_for_payload(&payload, profile);
}
if ACP_ACTIVE_RUNS
.lock()
.expect("acp active runs")
.contains_key(run_id)
{
return Some(crate::acp_bridge::runtime_name_for_profile(profile).to_string());
}
if is_acp_profile(profile) {
return Some(crate::acp_bridge::runtime_name_for_profile(profile).to_string());
}
None
}
fn provider_default_key_env(provider: &str) -> Option<&'static str> {
match provider.trim().to_ascii_lowercase().as_str() {
"deepseek" => Some("DEEPSEEK_API_KEY"),
"openrouter" => Some("OPENROUTER_API_KEY"),
"omniroute" => Some("OMNIROUTE_API_KEY"),
"openai" | "custom" => Some("OPENAI_API_KEY"),
_ => None,
}
}
fn acp_hermes_env_for_profile(profile: &str) -> Option<HashMap<String, String>> {
let config = fs::read_to_string(profile_config_path(profile)).ok()?;
let provider = yaml_path_value(&config, &["model", "provider"]).unwrap_or_default();
let key_env = yaml_path_value(&config, &["model", "key_env"])
.or_else(|| {
if provider.is_empty() {
None
} else {
yaml_path_value(&config, &["providers", &provider, "key_env"])
}
})
.or_else(|| provider_default_key_env(&provider).map(str::to_string))?;
let key = env_or_dotenv(&key_env)
.or_else(|| yaml_path_value(&config, &["model", "api_key"]))
.or_else(|| {
if provider.is_empty() {
None
} else {
yaml_path_value(&config, &["providers", &provider, "api_key"])
}
})
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())?;
let mut env = HashMap::new();
env.insert(key_env, key);
Some(env)
}
/// Returns the ACP runtime name for a profile.
/// For ACP profiles, returns the runtime backend name ("hermes" or "reasonix").
/// The profile name is used as the runtime name unless overridden by env var.
#[allow(dead_code)]
fn configured_runtime_for_profile(profile: &str) -> Option<String> {
if !is_acp_profile(profile) {
return None;
}
if profile == "reasonix" {
return Some("reasonix".into());
}
let env_key = profile_env_key("MNOTE_WEB", profile, "RUNTIME_NAME");
let name = env_or_dotenv(&env_key)
.filter(|v| !v.trim().is_empty())
.map(|v| v.trim().to_lowercase());
Some(name.unwrap_or_else(|| profile.to_lowercase()))
}
fn configured_upstream_for_profile(profile: &str) -> Option<String> {
let profile = profile.trim();
if !profile.is_empty() && profile != "default" {
let env_key = profile_env_key("MNOTE_WEB_HERMES", profile, "UPSTREAM_URL");
if let Some(value) = env_or_dotenv(&env_key) {
return Some(value.trim().trim_end_matches('/').to_string())
.filter(|value| !value.is_empty());
}
let enabled = profile_config_value(profile, &["API_SERVER_ENABLED"])
.map(|value| {
matches!(
value.trim().to_ascii_lowercase().as_str(),
"true" | "1" | "yes"
)
})
.unwrap_or(false);
let port = profile_config_value(profile, &["API_SERVER_PORT"]);
if enabled {
if let Some(port) = port
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
let host = profile_config_value(profile, &["API_SERVER_HOST"])
.unwrap_or_else(|| "127.0.0.1".into());
return Some(format!("http://{}:{}", host.trim(), port));
}
}
}
configured_upstream()
}
fn configured_api_key() -> Option<String> {
[
"MNOTE_WEB_HERMES_API_KEY",
"MNOTE_HERMES_API_KEY",
"HERMES_API_SERVER_KEY",
"API_SERVER_KEY",
]
.into_iter()
.find_map(env_or_dotenv)
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
fn configured_api_key_for_profile(profile: &str) -> Option<String> {
let profile = profile.trim();
if !profile.is_empty() && profile != "default" {
let env_key = profile_env_key("MNOTE_WEB_HERMES", profile, "API_KEY");
if let Some(value) = env_or_dotenv(&env_key) {
return Some(value);
}
if let Some(value) = profile_config_value(profile, &["API_SERVER_KEY"]) {
let trimmed = value.trim().to_string();
if !trimmed.is_empty() {
return Some(trimmed);
}
}
}
configured_api_key()
}
#[derive(Debug)]
struct GatewayProbe {
ok: bool,
status: String,
http_status: Option<u64>,
path: Option<String>,
message: Option<String>,
}
async fn probe_gateway_health(upstream: &str, api_key: Option<String>) -> GatewayProbe {
let client = match reqwest::Client::builder()
.timeout(Duration::from_secs(3))
.build()
{
Ok(client) => client,
Err(error) => {
return GatewayProbe {
ok: false,
status: "client_error".into(),
http_status: None,
path: None,
message: Some(format!("Hermes gateway health client 构造失败: {error}")),
};
}
};
for path in ["/health", "/v1/models"] {
let Ok(url) = upstream_url(upstream, path) else {
continue;
};
let mut request = client.get(url);
if let Some(api_key) = api_key.as_deref() {
request = request.bearer_auth(api_key);
}
match request.send().await {
Ok(response) if response.status().is_success() => {
return GatewayProbe {
ok: true,
status: "ok".into(),
http_status: Some(response.status().as_u16() as u64),
path: Some(path.into()),
message: None,
};
}
Ok(response) => {
let status = response.status();
let text = response.text().await.unwrap_or_default();
if path == "/v1/models" || status != reqwest::StatusCode::NOT_FOUND {
return GatewayProbe {
ok: false,
status: "upstream_error".into(),
http_status: Some(status.as_u16() as u64),
path: Some(path.into()),
message: Some(text.chars().take(600).collect()),
};
}
}
Err(error) => {
return GatewayProbe {
ok: false,
status: "unreachable".into(),
http_status: None,
path: Some(path.into()),
message: Some(error.to_string()),
};
}
}
}
GatewayProbe {
ok: false,
status: "unknown".into(),
http_status: None,
path: None,
message: Some("Hermes gateway health 探测未返回有效结果".into()),
}
}
fn env_or_dotenv(key: &str) -> Option<String> {
if let Ok(value) = std::env::var(key) {
let trimmed = value.trim().trim_matches('"').to_string();
if !trimmed.is_empty() {
return Some(trimmed);
}
}
if cfg!(test) {
return None;
}
let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../..")
.join(".env.all");
let content = fs::read_to_string(root).ok()?;
for line in content.lines() {
let line = line.trim_end_matches('\r');
if line.starts_with('#') || line.trim().is_empty() {
continue;
}
let Some((candidate_key, value)) = line.split_once('=') else {
continue;
};
if candidate_key.trim() != key {
continue;
}
let trimmed = value.trim().trim_matches('"').to_string();
if !trimmed.is_empty() {
return Some(trimmed);
}
}
None
}
fn yaml_path_value(content: &str, path: &[&str]) -> Option<String> {
let mut stack: Vec<(usize, String)> = Vec::new();
for raw_line in content.lines() {
let line = raw_line.trim_end_matches('\r');
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') || !line.contains(':') {
continue;
}
let indent = line.chars().take_while(|ch| ch.is_whitespace()).count();
while stack
.last()
.map(|(level, _)| *level >= indent)
.unwrap_or(false)
{
stack.pop();
}
let Some((key, value)) = trimmed.split_once(':') else {
continue;
};
let key = key.trim().trim_matches('"').trim_matches('\'').to_string();
let value = value
.trim()
.trim_matches('"')
.trim_matches('\'')
.to_string();
stack.push((indent, key));
let keys = stack
.iter()
.map(|(_, key)| key.as_str())
.collect::<Vec<_>>();
if keys == path && !value.is_empty() {
return Some(value);
}
}
None
}
fn profile_gateway_status(profile: &str) -> Value {
let normalized_profile = if profile.trim().is_empty() {
"default"
} else {
profile.trim()
};
let config_path = profile_config_path(normalized_profile);
let config = fs::read_to_string(&config_path).unwrap_or_default();
let model_default = yaml_path_value(&config, &["model", "default"]).unwrap_or_default();
let provider = yaml_path_value(&config, &["model", "provider"]).unwrap_or_default();
let gateway = yaml_path_value(&config, &["model", "gateway"])
.or_else(|| yaml_path_value(&config, &["model", "base_url"]))
.unwrap_or_default();
let model_api_key = yaml_path_value(&config, &["model", "api_key"]);
let model_key_env = yaml_path_value(&config, &["model", "key_env"]);
let provider_api_key = if provider.is_empty() {
None
} else {
yaml_path_value(&config, &["providers", &provider, "api_key"])
};
let provider_key_env = if provider.is_empty() {
None
} else {
yaml_path_value(&config, &["providers", &provider, "key_env"])
};
let key_env_configured = model_key_env
.as_deref()
.or(provider_key_env.as_deref())
.and_then(env_or_dotenv)
.is_some();
let api_key_configured = model_api_key
.as_deref()
.or(provider_api_key.as_deref())
.map(|value| !value.trim().is_empty())
.unwrap_or(false)
|| key_env_configured;
let mut suggestions = Vec::new();
if model_default.is_empty() {
suggestions
.push("当前 Hermes profile 缺少 model.default,请在 Hermes 设置中选择默认模型。");
}
if provider.is_empty() {
suggestions
.push("当前 Hermes profile 缺少 model.provider,请在 Hermes 设置中选择 provider。");
}
if !api_key_configured {
suggestions.push("当前 Hermes profile 未检测到 provider API key 或 key_env,请在 Hermes 设置中补齐 API key。");
}
json!({
"name": normalized_profile,
"path": profile_home(normalized_profile).to_string_lossy(),
"configPath": config_path.to_string_lossy(),
"configExists": config_path.exists(),
"modelDefault": model_default,
"provider": provider,
"gateway": gateway,
"modelConfigured": !model_default.is_empty(),
"apiKeyConfigured": api_key_configured,
"suggestions": suggestions
})
}
fn unique_string_values(values: Vec<Value>) -> Vec<Value> {
let mut seen = HashSet::new();
values
.into_iter()
.filter_map(|value| value.as_str().map(str::trim).map(ToOwned::to_owned))
.filter(|value| !value.is_empty())
.filter(|value| seen.insert(value.clone()))
.map(Value::String)
.collect()
}
fn hermes_settings_suggestions(status: Option<u64>, message: Option<&str>) -> Vec<String> {
let lower = message.unwrap_or_default().to_lowercase();
let mut suggestions = Vec::new();
if lower.contains("model.default")
|| (lower.contains("model") && (lower.contains("default") || lower.contains("missing")))
{
suggestions.push(
"当前 Hermes profile 可能缺少 model.default,请在 Hermes 设置中选择默认模型。".into(),
);
}
if lower.contains("api key")
|| lower.contains("apikey")
|| lower.contains("unauthorized")
|| lower.contains("forbidden")
|| matches!(status, Some(401 | 403))
{
suggestions.push("当前 Hermes provider 可能缺少 API key 或 key_env 未生效,请在 Hermes 设置中补齐 API key。".into());
}
if lower.contains("profile") || lower.contains("provider") {
suggestions
.push("请检查当前页面 AI 选择的 Hermes profile、provider 与 gateway 是否匹配。".into());
}
if suggestions.is_empty() {
suggestions.push("请检查 Hermes gateway 是否正在运行,并确认当前 profile 的 model.default、provider 与 API key 已配置。".into());
}
suggestions
}
fn build_run_upstream_body(context: &RequestContext, payload: Value) -> Result<Value, WebError> {
let message = extract_run_message(&payload).ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 message")
.with_context(context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let document_id = payload
.get("documentId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or("current");
let trace_id = payload
.get("traceId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or(&context.trace.trace_id);
let session_id = payload
.get("sessionId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(ToOwned::to_owned)
.unwrap_or_else(|| stable_session_id(document_id, trace_id));
let page_context = payload.get("pageContext").cloned().unwrap_or(Value::Null);
let workspace_id = payload.get("workspaceId").cloned().unwrap_or(Value::Null);
let profile = payload
.get("profile")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty());
let tool_guidance = concat!(
"每次调用 mnote 工具都必须显式透传 workspaceId、documentId、actorId、sessionId、runId、traceId",
"actorId 必须使用本 instructions 中的 actorId,不允许省略或填 hermes/anonymous。",
"凡用户要求新增、删除、修改、替换、移动正文段落或块,必须走块级链路。",
"本 instructions 的 pageContext.aiContext 是本次 run 开始时冻结的 mnote.page_ai_context.v1",
"其中 contextBlocks/pageXml/pageText 可直接作为 tiptapRead 风格的已读上下文使用;",
"如果 aiContext 已包含足够唯一的目标文本或 blockId,应直接调用 mnote_doc_apply_block_ops",
"不要为了同一上下文再先调用 mnote_doc_fetch。",
"scope=selection 时只能修改 allowedTargetBlockIds 内的块;",
"简单小段落编辑或同一页内多个普通叶子块操作,应优先用 mnote_doc_apply_block_ops 一次提交 operations",
"该快路径等价于 Tiptap tiptapEdit 风格的批量编辑工具,可用唯一 matchText/anchorText 或已知 blockId/anchorBlockId 定位,",
"避免 fetch、plan、多个单步写入造成多轮模型往返。",
"只有当文本不唯一、目标不明确、涉及复杂块/子块/表格/资源块,或 apply_block_ops 返回歧义/不支持时,",
"再降级为 mnote_doc_fetch(scope=full 或 keyword, detail=with_ids) 定位块,",
"mnote_block_fetch 读取目标块 revisionRef/context",
"mnote_doc_plan_update(dryRun=true) 生成并检查 diff",
"最后调用 mnote_block_replace、mnote_block_insert_after、mnote_block_delete 或 mnote_block_move_after。",
"写入后再 mnote_doc_fetch 或 mnote_block_fetch 回读验证。",
"mnote_page_get 只用于读取页面标题、页面设置或粗略摘要;",
"mnote_page_save 只允许在用户明确要求整页覆盖/整页追加且块级工具无法表达时作为高风险兜底,",
"不能作为正文块新增、删除、修改、移动的首选工具。",
"不要依据非 aiContext 的页面摘要猜测正文内容或块 id。"
);
let instructions = json!({
"role": "mnote_page_ai_context",
"workspaceId": workspace_id,
"documentId": document_id,
"profile": profile.unwrap_or("default"),
"actorId": payload.get("actorId").and_then(Value::as_str).unwrap_or(&context.auth.actor_id),
"actorType": payload.get("actorType").and_then(Value::as_str).unwrap_or(&context.auth.actor_type),
"sessionId": session_id,
"runId": payload
.get("runId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or(&session_id),
"traceId": trace_id,
"toolGuidance": tool_guidance,
"blockEditingToolOrder": [
"mnote_doc_apply_block_ops for simple small paragraph edits or multiple leaf-block operations",
"mnote_doc_fetch | mnote_doc_find only when target text is ambiguous or structure is complex",
"mnote_doc_fetch",
"mnote_block_fetch",
"mnote_doc_plan_update(dryRun=true)",
"mnote_block_replace | mnote_block_insert_after | mnote_block_delete | mnote_block_move_after",
"mnote_doc_fetch | mnote_block_fetch"
],
"pageSavePolicy": {
"mnote_page_save": "fallback_only_for_explicit_whole_page_write",
"forBlockEditing": "forbidden_as_first_choice"
},
"selectedBlockId": payload.get("selectedBlockId").cloned().unwrap_or(Value::Null),
"selectedText": payload.get("selectedText").cloned().unwrap_or(Value::Null),
"pageContext": sanitize_run_page_context(page_context)
})
.to_string();
let mut body = json!({
"input": message,
"session_id": session_id,
"instructions": instructions
});
if let Some(model) = payload
.get("model")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
{
body["model"] = Value::String(model.to_string());
}
if let Some(profile) = profile {
body["profile"] = Value::String(profile.to_string());
}
Ok(body)
}
fn effective_run_actor(state: &AppState, context: &RequestContext) -> (String, String) {
let actor_id = context.auth.actor_id.trim();
if !actor_id.is_empty() && actor_id != "anonymous" {
return (actor_id.to_string(), context.auth.actor_type.clone());
}
if context.auth.authorization.is_some() || context.auth.cookie_header.is_some() {
return (state.config().dev_user_id.clone(), "devFallback".into());
}
("anonymous".into(), "anonymous".into())
}
async fn resolve_run_actor(state: &AppState, context: &RequestContext) -> (String, String) {
let actor_id = context.auth.actor_id.trim();
if !actor_id.is_empty() && actor_id != "anonymous" {
return (actor_id.to_string(), context.auth.actor_type.clone());
}
if context.auth.authorization.is_none() && context.auth.cookie_header.is_none() {
return ("anonymous".into(), "anonymous".into());
}
if let Ok(Some(user_id)) = resolve_current_convex_user_id(state, context).await {
return (user_id, "user".into());
}
effective_run_actor(state, context)
}
async fn resolve_current_convex_user_id(
state: &AppState,
context: &RequestContext,
) -> Result<Option<String>, WebError> {
let payload = execute_convex_query_by_name(
state.config(),
context,
"users:currentUser",
json!({}),
None,
"hermes_run_current_user",
)
.await?;
Ok(payload
.get("_id")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned))
}
fn stamp_run_actor(payload: &mut Value, actor_id: &str, actor_type: &str) {
let Value::Object(map) = payload else {
return;
};
map.insert("actorId".into(), Value::String(actor_id.to_string()));
map.insert("actorType".into(), Value::String(actor_type.to_string()));
}
fn sanitize_run_page_context(page_context: Value) -> Value {
let Some(source) = page_context.as_object() else {
return Value::Null;
};
let mut sanitized = serde_json::Map::new();
for key in [
"contextScope",
"node",
"pageSubtreeSource",
"evidence",
"pageOptions",
"contentAccess",
"aiContext",
] {
if let Some(value) = source.get(key) {
sanitized.insert(key.to_string(), value.clone());
}
}
sanitized.insert(
"contentAccess".to_string(),
sanitized
.get("contentAccess")
.cloned()
.unwrap_or_else(|| Value::String("mnote.doc.fetch".into())),
);
Value::Object(sanitized)
}
fn now_ms() -> u128 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis())
.unwrap_or_default()
}
fn runtime_state_to_json(state: &HermesRuntimeState) -> Value {
json!({
"sessionId": state.session_id,
"runId": state.run_id,
"profile": state.profile,
"documentId": state.document_id,
"traceId": state.trace_id,
"status": state.status,
"startedAt": state.started_at,
"lastEventAt": state.last_event_at,
"lastEvent": state.last_event,
"lastToolName": state.last_tool_name,
"lastToolCallId": state.last_tool_call_id,
"lastAuditId": state.last_audit_id
})
}
fn runtime_state_for_session(session_id: &str) -> Value {
let registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
registry
.get(session_id)
.map(runtime_state_to_json)
.unwrap_or(Value::Null)
}
fn runtime_state_for_run(run_id: &str) -> Option<Value> {
let registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
registry
.values()
.find(|state| state.run_id == run_id)
.map(runtime_state_to_json)
}
fn runtime_status_for_run(run_id: &str) -> Option<String> {
let registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
registry
.values()
.find(|state| state.run_id == run_id)
.map(|state| state.status.clone())
}
fn run_registration_from_payload(
context: &RequestContext,
payload: &Value,
) -> HermesRunRegistration {
let document_id = payload
.get("documentId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or("current")
.to_string();
let trace_id = payload
.get("traceId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or(&context.trace.trace_id)
.to_string();
let session_id = payload
.get("sessionId")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(ToOwned::to_owned)
.unwrap_or_else(|| stable_session_id(&document_id, &trace_id));
let profile = payload
.get("profile")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("default")
.to_string();
HermesRunRegistration {
session_id,
profile,
document_id,
trace_id,
}
}
/// Register a runtime state from a registration (without upstream response).
/// Used by the ACP path where no Hermes HTTP upstream exists.
fn new_acp_run_id(registration: &HermesRunRegistration) -> String {
format!("run_{}_{}", registration.trace_id, now_ms())
}
fn register_acp_runtime(registration: &HermesRunRegistration, run_id: &str) -> Value {
let now = now_ms();
let state = HermesRuntimeState {
session_id: registration.session_id.clone(),
run_id: run_id.to_string(),
profile: registration.profile.clone(),
document_id: registration.document_id.clone(),
trace_id: registration.trace_id.clone(),
status: "acp_pending".into(),
started_at: now,
last_event_at: now,
last_event: Some("run.started".into()),
last_tool_name: None,
last_tool_call_id: None,
last_audit_id: None,
};
let json = runtime_state_to_json(&state);
HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry")
.insert(registration.session_id.clone(), state);
json
}
fn register_runtime_from_create_run_response(
registration: &HermesRunRegistration,
payload: &Value,
) -> Option<Value> {
let upstream = payload.get("upstream").unwrap_or(payload);
let run_id = upstream
.get("run_id")
.or_else(|| upstream.get("runId"))
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())?
.to_string();
let now = now_ms();
let state = HermesRuntimeState {
session_id: registration.session_id.clone(),
run_id,
profile: registration.profile.clone(),
document_id: registration.document_id.clone(),
trace_id: registration.trace_id.clone(),
status: "running".into(),
started_at: now,
last_event_at: now,
last_event: Some("run.started".into()),
last_tool_name: None,
last_tool_call_id: None,
last_audit_id: None,
};
let json = runtime_state_to_json(&state);
HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry")
.insert(registration.session_id.clone(), state);
Some(json)
}
fn update_runtime_by_run_id(
run_id: &str,
status: &str,
event: Option<&str>,
tool: Option<(Option<String>, Option<String>, Option<String>)>,
) {
let mut registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
let Some(state) = registry.values_mut().find(|state| state.run_id == run_id) else {
return;
};
state.status = status.to_string();
state.last_event_at = now_ms();
if let Some(event) = event {
state.last_event = Some(event.to_string());
}
if let Some((tool_name, tool_call_id, audit_id)) = tool {
if let Some(tool_name) = tool_name {
state.last_tool_name = Some(tool_name);
}
if let Some(tool_call_id) = tool_call_id {
state.last_tool_call_id = Some(tool_call_id);
}
if let Some(audit_id) = audit_id {
state.last_audit_id = Some(audit_id);
}
}
}
fn normalize_sse_chunk(run_id: &str, chunk: &str) -> (String, Vec<String>) {
let events = sse_json_events(chunk);
let terminal_sessions = update_runtime_from_events(run_id, &events);
let mut output = String::new();
for event in events {
let normalized = normalize_runtime_event(run_id, event);
let event_name = normalized
.get("event")
.and_then(Value::as_str)
.unwrap_or("runtime.event");
output.push_str("event: ");
output.push_str(event_name);
output.push_str("\ndata: ");
output.push_str(&normalized.to_string());
output.push_str("\n\n");
}
if output.is_empty() {
output.push_str(chunk);
}
(output, terminal_sessions)
}
fn update_runtime_from_events(run_id: &str, events: &[Value]) -> Vec<String> {
let mut terminal_sessions = Vec::new();
for event in events {
let event_name = canonical_event_name(event);
if event_name.is_empty() {
continue;
}
let next_status = match event_name.as_str() {
"tool.started" => "tool_calling",
"tool.completed" | "tool.failed" => "running",
"run.completed" => "completed",
"run.failed" => "failed",
"run.aborted" => "aborted",
_ => "running",
};
let tool = if event_name.starts_with("tool.") {
Some((
event
.get("name")
.or_else(|| event.get("tool"))
.or_else(|| event.get("toolName"))
.and_then(Value::as_str)
.map(ToOwned::to_owned),
event
.get("tool_call_id")
.or_else(|| event.get("toolCallId"))
.or_else(|| event.get("id"))
.and_then(Value::as_str)
.map(ToOwned::to_owned),
event
.get("audit_id")
.or_else(|| event.get("auditId"))
.and_then(Value::as_str)
.map(ToOwned::to_owned),
))
} else {
None
};
update_runtime_by_run_id(run_id, next_status, Some(&event_name), tool);
if matches!(next_status, "completed" | "failed" | "aborted") {
if let Some(session_id) = session_id_for_run(run_id) {
terminal_sessions.push(session_id);
}
}
}
terminal_sessions
}
fn canonical_event_name(event: &Value) -> String {
let raw = event
.get("event")
.and_then(Value::as_str)
.or_else(|| event.get("type").and_then(Value::as_str))
.unwrap_or_default();
match raw {
"assistant_message" | "response.output_text.delta" => "message.delta",
"response.completed" => "run.completed",
"response.failed" => "run.failed",
"run.cancelled" | "run.canceled" | "abort.completed" => "run.aborted",
value => value,
}
.to_string()
}
fn normalize_runtime_event(run_id: &str, event: Value) -> Value {
let event_name = canonical_event_name(&event);
let session_id = event
.get("session_id")
.or_else(|| event.get("sessionId"))
.and_then(Value::as_str)
.map(ToOwned::to_owned)
.or_else(|| session_id_for_run(run_id));
let trace_id = event
.get("trace_id")
.or_else(|| event.get("traceId"))
.and_then(Value::as_str)
.unwrap_or_default();
let mut normalized = json!({
"schemaVersion": "mnote.hermes_runtime_event.v1",
"event": event_name,
"runId": event.get("run_id").or_else(|| event.get("runId")).and_then(Value::as_str).unwrap_or(run_id),
"sessionId": session_id,
"traceId": trace_id,
"upstream": event
});
if event_name == "message.delta" {
normalized["delta"] = event
.get("delta")
.or_else(|| event.get("text"))
.or_else(|| event.get("output_text"))
.cloned()
.unwrap_or(Value::Null);
}
if event_name.starts_with("tool.") {
normalized["toolName"] = event
.get("name")
.or_else(|| event.get("tool"))
.or_else(|| event.get("toolName"))
.cloned()
.unwrap_or(Value::Null);
normalized["toolCallId"] = event
.get("tool_call_id")
.or_else(|| event.get("toolCallId"))
.or_else(|| event.get("id"))
.cloned()
.unwrap_or(Value::Null);
normalized["args"] = event
.get("arguments")
.or_else(|| event.get("args"))
.or_else(|| event.get("input"))
.cloned()
.unwrap_or(Value::Null);
normalized["summary"] = event
.get("summary")
.or_else(|| event.get("result"))
.or_else(|| event.get("output"))
.cloned()
.unwrap_or(Value::Null);
normalized["code"] = event.get("code").cloned().unwrap_or(Value::Null);
normalized["error"] = event.get("error").cloned().unwrap_or(Value::Null);
normalized["auditId"] = event
.get("audit_id")
.or_else(|| event.get("auditId"))
.cloned()
.unwrap_or(Value::Null);
}
if event_name == "run.completed" {
normalized["output"] = event.get("output").cloned().unwrap_or(Value::Null);
normalized["usage"] = event.get("usage").cloned().unwrap_or(Value::Null);
}
if event_name == "run.failed" {
normalized["code"] = event.get("code").cloned().unwrap_or(Value::Null);
normalized["message"] = event
.get("message")
.or_else(|| event.get("error"))
.cloned()
.unwrap_or(Value::Null);
}
normalized
}
fn session_id_for_run(run_id: &str) -> Option<String> {
let registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
registry
.values()
.find(|state| state.run_id == run_id)
.map(|state| state.session_id.clone())
}
fn profile_for_run(run_id: &str) -> Option<String> {
let registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
registry
.values()
.find(|state| state.run_id == run_id)
.map(|state| state.profile.clone())
}
fn sse_json_events(chunk: &str) -> Vec<Value> {
chunk
.split("\n\n")
.filter_map(|block| {
let data = block
.lines()
.filter_map(|line| line.strip_prefix("data:"))
.map(str::trim)
.collect::<Vec<_>>()
.join("\n");
if data.is_empty() || data == "[DONE]" {
return None;
}
serde_json::from_str::<Value>(&data).ok()
})
.collect()
}
async fn proxy_json(
context: &RequestContext,
method: reqwest::Method,
upstream: &str,
path: &str,
body: Option<Value>,
profile: Option<&str>,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
let url = upstream_url(upstream, path)?;
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(1800))
.build()
.map_err(|error| {
WebError::internal(format!("Hermes client 构造失败: {error}")).with_context(context)
})?;
let mut request = client.request(method, url);
if let Some(api_key) = profile
.and_then(configured_api_key_for_profile)
.or_else(configured_api_key)
{
request = request.bearer_auth(api_key);
}
if let Some(body) = body {
request = request.json(&body);
}
let response = request.send().await.map_err(|error| {
WebError::bad_gateway_code(
"hermes_client_upstream_unavailable",
format!("Hermes upstream 连接失败: {error}"),
)
.with_context(context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let status = response.status();
let text = response.text().await.unwrap_or_default();
if !status.is_success() {
return Err(upstream_error(context, status, text));
}
let payload = serde_json::from_str::<Value>(&text).unwrap_or_else(|_| {
json!({
"ok": true,
"traceId": context.trace.trace_id,
"raw": text
})
});
Ok((
StatusCode::OK,
stamp_client_headers(),
Json(normalize_success_payload(context, payload)),
))
}
fn session_has_active_run(session_id: &str) -> bool {
let registry = HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry");
registry
.get(session_id)
.map(|state| {
matches!(
state.status.as_str(),
"queued" | "running" | "tool_calling" | "aborting"
)
})
.unwrap_or(false)
}
fn queue_len_for_session(session_id: &str) -> usize {
HERMES_RUN_QUEUE
.lock()
.expect("hermes run queue")
.get(session_id)
.map(VecDeque::len)
.unwrap_or_default()
}
fn queued_run_to_json(queued: &HermesQueuedRun, queue_length: usize) -> Value {
json!({
"ok": true,
"queued": true,
"queueId": queued.queue_id,
"sessionId": queued.session_id,
"profile": queued.profile,
"documentId": queued.document_id,
"traceId": queued.trace_id,
"status": "queued",
"queueLength": queue_length,
"queuedAt": queued.queued_at,
"contextSummary": queued.context_summary
})
}
fn enqueue_run(
context: &RequestContext,
registration: &HermesRunRegistration,
payload: &Value,
) -> Result<Value, WebError> {
let input = extract_run_message(payload).ok_or_else(|| {
WebError::bad_request_code("hermes_client_bad_request", "缺少 message")
.with_context(context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
})?;
let queued = HermesQueuedRun {
queue_id: format!(
"queue_{}_{}",
sanitize_id_part(&registration.trace_id),
now_ms()
),
session_id: registration.session_id.clone(),
profile: registration.profile.clone(),
document_id: registration.document_id.clone(),
trace_id: registration.trace_id.clone(),
actor_id: payload
.get("actorId")
.and_then(Value::as_str)
.unwrap_or("anonymous")
.to_string(),
actor_type: payload
.get("actorType")
.and_then(Value::as_str)
.unwrap_or("anonymous")
.to_string(),
input,
context_summary: queue_context_summary(payload),
queued_at: now_ms(),
};
let queue_length = {
let mut queue = HERMES_RUN_QUEUE.lock().expect("hermes run queue");
let entries = queue.entry(registration.session_id.clone()).or_default();
entries.push_back(queued.clone());
entries.len()
};
Ok(queued_run_to_json(&queued, queue_length))
}
async fn start_next_queued_run(context: RequestContext, session_id: String) {
if session_has_active_run(&session_id) {
return;
}
let Some(queued) = pop_next_queued_run(&session_id) else {
return;
};
let Some(upstream) = configured_upstream_for_profile(&queued.profile) else {
warn!(session_id = %session_id, queue_id = %queued.queue_id, "Hermes queue 无 upstream,无法自动启动下一条 run");
return;
};
let payload = queued_run_payload(&queued);
let upstream_body = match build_run_upstream_body(&context, payload) {
Ok(body) => body,
Err(error) => {
warn!(session_id = %session_id, queue_id = %queued.queue_id, error = ?error, "Hermes queue 构造 run body 失败");
return;
}
};
let registration = HermesRunRegistration {
session_id: queued.session_id.clone(),
profile: queued.profile.clone(),
document_id: queued.document_id.clone(),
trace_id: queued.trace_id.clone(),
};
match proxy_json(
&context,
reqwest::Method::POST,
&upstream,
"/v1/runs",
Some(upstream_body),
Some(&queued.profile),
)
.await
{
Ok((_, _, Json(payload))) => {
let _ = register_runtime_from_create_run_response(&registration, &payload);
}
Err(error) => {
warn!(session_id = %session_id, queue_id = %queued.queue_id, error = ?error, "Hermes queue 自动启动下一条 run 失败");
}
}
}
fn queued_run_payload(queued: &HermesQueuedRun) -> Value {
json!({
"documentId": queued.document_id,
"sessionId": queued.session_id,
"profile": queued.profile,
"message": queued.input,
"traceId": queued.trace_id,
"actorId": queued.actor_id,
"actorType": queued.actor_type,
"contextScope": queued.context_summary.get("contextScope").cloned().unwrap_or(Value::Null)
})
}
fn remove_queued_run(session_id: &str, queue_id: &str) -> bool {
let mut queue = HERMES_RUN_QUEUE.lock().expect("hermes run queue");
let Some(entries) = queue.get_mut(session_id) else {
return false;
};
let before = entries.len();
entries.retain(|item| item.queue_id != queue_id);
let removed = entries.len() != before;
if entries.is_empty() {
queue.remove(session_id);
}
removed
}
fn pop_next_queued_run(session_id: &str) -> Option<HermesQueuedRun> {
let mut queue = HERMES_RUN_QUEUE.lock().expect("hermes run queue");
let entries = queue.get_mut(session_id)?;
let next = entries.pop_front();
if entries.is_empty() {
queue.remove(session_id);
}
next
}
fn extract_run_message(payload: &Value) -> Option<String> {
payload
.get("message")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
payload
.get("messages")
.and_then(Value::as_array)
.and_then(|messages| messages.last())
.and_then(|message| message.get("content"))
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
}
fn queue_context_summary(payload: &Value) -> Value {
let page_context = payload.get("pageContext").unwrap_or(&Value::Null);
json!({
"contextScope": payload.get("contextScope").cloned().unwrap_or(Value::Null),
"selectedBlockId": payload.get("selectedBlockId").cloned().unwrap_or(Value::Null),
"hasSelectedText": payload.get("selectedText").and_then(Value::as_str).map(|value| !value.trim().is_empty()).unwrap_or(false),
"hasPageContext": !page_context.is_null(),
"pageContextKeys": page_context.as_object().map(|object| object.keys().take(12).cloned().collect::<Vec<_>>()).unwrap_or_default()
})
}
fn normalize_success_payload(context: &RequestContext, payload: Value) -> Value {
if payload.get("ok").is_some() {
payload
} else {
json!({
"ok": true,
"traceId": context.trace.trace_id,
"upstream": payload
})
}
}
fn upstream_error(context: &RequestContext, status: reqwest::StatusCode, text: String) -> WebError {
let (response_status, code) = match status {
reqwest::StatusCode::UNAUTHORIZED | reqwest::StatusCode::FORBIDDEN => (
StatusCode::BAD_GATEWAY,
"hermes_client_upstream_unauthorized",
),
reqwest::StatusCode::TOO_MANY_REQUESTS => (
StatusCode::TOO_MANY_REQUESTS,
"hermes_client_upstream_rate_limited",
),
status if status.is_server_error() => (
StatusCode::BAD_GATEWAY,
"hermes_client_upstream_unavailable",
),
_ => (StatusCode::BAD_GATEWAY, "hermes_client_upstream_error"),
};
let suggestions = hermes_settings_suggestions(Some(status.as_u16() as u64), Some(&text));
WebError::new(
response_status,
code,
format!(
"Hermes upstream 返回 HTTP {}: {}。建议:{}",
status.as_u16(),
text.chars().take(600).collect::<String>(),
suggestions.join("")
),
)
.with_context(context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
.with_header("x-hermes-client-suggestion", suggestions.join(" | "))
}
fn hermes_unconfigured(
context: &RequestContext,
) -> Result<(StatusCode, HeaderMap, Json<Value>), WebError> {
Err(hermes_unconfigured_error(context))
}
fn hermes_unconfigured_error(context: &RequestContext) -> WebError {
WebError::service_unavailable_code(
"hermes_client_unconfigured",
"Hermes client proxy 未配置 MNOTE_WEB_HERMES_UPSTREAM_URL",
)
.with_context(context)
.with_header(HEADER_MNOTE_WEB_OWNER, "mnote-web")
.with_header(HEADER_HERMES_CLIENT_OWNER, "mnote-web-hermes-client")
}
fn upstream_url(upstream: &str, path: &str) -> Result<String, WebError> {
let url = format!(
"{}/{}",
upstream.trim_end_matches('/'),
path.trim_start_matches('/')
);
reqwest::Url::parse(&url)
.map(|url| url.to_string())
.map_err(|error| WebError::internal(format!("Hermes upstream URL 无效: {error}")))
}
fn stable_session_id(document_id: &str, trace_id: &str) -> String {
format!(
"mnote_{}_{}",
sanitize_id_part(document_id),
sanitize_id_part(trace_id)
)
}
fn sanitize_id_part(value: &str) -> String {
let sanitized = value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
ch
} else {
'_'
}
})
.collect::<String>();
if sanitized.is_empty() {
"current".into()
} else {
sanitized
}
}
fn url_escape(value: &str) -> String {
value
.bytes()
.flat_map(|byte| match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
vec![byte as char]
}
_ => format!("%{byte:02X}").chars().collect(),
})
.collect()
}
fn stamp_client_headers() -> HeaderMap {
let mut headers = HeaderMap::new();
stamp_client_headers_into(&mut headers);
headers
}
fn stamp_client_headers_into(headers: &mut HeaderMap) {
if let Ok(name) = HeaderName::from_lowercase(HEADER_MNOTE_WEB_OWNER.as_bytes()) {
headers.insert(name, HeaderValue::from_static("mnote-web"));
}
if let Ok(name) = HeaderName::from_lowercase(HEADER_HERMES_CLIENT_OWNER.as_bytes()) {
headers.insert(name, HeaderValue::from_static("mnote-web-hermes-client"));
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::app::{build_app, AppConfig, AppState};
use axum::body::{to_bytes, Body};
use axum::http::Request;
use axum::routing::{get, post};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use tower::util::ServiceExt;
fn env_lock() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}
fn runtime_lock() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}
fn clear_runtime_registry() {
HERMES_RUNTIME_REGISTRY
.lock()
.expect("hermes runtime registry")
.clear();
}
fn clear_run_queue() {
HERMES_RUN_QUEUE.lock().expect("hermes run queue").clear();
}
fn app() -> axum::Router {
build_app(AppState::new(AppConfig {
service_name: "mnote-web".into(),
service_version: "0.1.0".into(),
bind_addr: "127.0.0.1:0".into(),
public_bind_addr: "127.0.0.1:3000".into(),
legacy_next_base_url: Some("http://127.0.0.1:3100".into()),
enable_legacy_next_compat: true,
enable_debug_shell_routes: false,
enable_editor_actor: true,
hermes_base_path: "/api/hermes".into(),
compat_next_base_path: "/api/compat/next".into(),
convex_url: None,
convex_admin_key: None,
allow_dev_fixtures: true,
query_fixtures_json: None,
mutation_fixtures_json: None,
dev_user_id: "dev-user".into(),
dev_user_name: "开发用户".into(),
dev_user_email: "dev@mnote.local".into(),
}))
}
fn test_state() -> AppState {
AppState::new(AppConfig {
service_name: "mnote-web".into(),
service_version: "0.1.0".into(),
bind_addr: "127.0.0.1:0".into(),
public_bind_addr: "127.0.0.1:3000".into(),
legacy_next_base_url: Some("http://127.0.0.1:3100".into()),
enable_legacy_next_compat: true,
enable_debug_shell_routes: false,
enable_editor_actor: true,
hermes_base_path: "/api/hermes".into(),
compat_next_base_path: "/api/compat/next".into(),
convex_url: None,
convex_admin_key: None,
allow_dev_fixtures: true,
query_fixtures_json: None,
mutation_fixtures_json: None,
dev_user_id: "dev-user".into(),
dev_user_name: "开发用户".into(),
dev_user_email: "dev@mnote.local".into(),
})
}
#[tokio::test]
async fn hermes_client_gateway_health_reports_unconfigured_profile_settings() {
let _env_guard = env_lock().lock().expect("env lock");
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
std::env::remove_var("MNOTE_HERMES_UPSTREAM_URL");
std::env::remove_var("MNOTE_HERMES_API_BASE_URL");
let hermes_home = std::env::temp_dir().join(format!(
"mnote-web-hermes-health-unconfigured-{}",
std::process::id()
));
let _ = fs::remove_dir_all(&hermes_home);
fs::create_dir_all(&hermes_home).expect("hermes home");
std::env::set_var("HERMES_HOME", &hermes_home);
let response = app()
.oneshot(
Request::builder()
.method("GET")
.uri("/api/hermes/client/gateway/health?profile=default")
.header("x-mnote-actor-id", "user_1")
.body(Body::empty())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["ok"], false);
assert_eq!(payload["gateway"]["status"], "unconfigured");
assert_eq!(payload["profile"]["modelConfigured"], false);
let suggestions = payload["suggestions"].to_string();
assert!(suggestions.contains("MNOTE_WEB_HERMES_UPSTREAM_URL"));
assert!(suggestions.contains("model.default"));
std::env::remove_var("HERMES_HOME");
let _ = fs::remove_dir_all(&hermes_home);
}
#[tokio::test]
async fn hermes_client_gateway_health_reports_profile_model_and_api_key_gaps() {
let _env_guard = env_lock().lock().expect("env lock");
let hermes_home = std::env::temp_dir().join(format!(
"mnote-web-hermes-health-profile-{}",
std::process::id()
));
let _ = fs::remove_dir_all(&hermes_home);
let profile_dir = hermes_home.join("profiles").join("chemist");
fs::create_dir_all(&profile_dir).expect("profile dir");
fs::write(
profile_dir.join("config.yaml"),
"model:\n provider: openai\nproviders:\n openai:\n base_url: https://api.openai.com/v1\n",
)
.expect("profile config");
std::env::set_var("HERMES_HOME", &hermes_home);
let mock = axum::Router::new().route(
"/v1/models",
get(|| async { (StatusCode::BAD_REQUEST, "missing model.default and API key") }),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("listener");
let upstream = format!("http://{}", listener.local_addr().expect("addr"));
tokio::spawn(async move {
axum::serve(listener, mock).await.expect("mock upstream");
});
std::env::set_var("MNOTE_WEB_HERMES_UPSTREAM_URL", &upstream);
let response = app()
.oneshot(
Request::builder()
.method("GET")
.uri("/api/hermes/client/gateway/health?profile=chemist")
.header("x-mnote-actor-id", "user_1")
.body(Body::empty())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["ok"], false);
assert_eq!(payload["gateway"]["status"], "upstream_error");
assert_eq!(payload["profile"]["name"], "chemist");
assert_eq!(payload["profile"]["modelConfigured"], false);
assert_eq!(payload["profile"]["apiKeyConfigured"], false);
let suggestions = payload["suggestions"].to_string();
assert!(suggestions.contains("model.default"));
assert!(suggestions.contains("API key"));
std::env::remove_var("HERMES_HOME");
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
let _ = fs::remove_dir_all(&hermes_home);
}
#[tokio::test]
async fn hermes_client_upstream_error_message_points_to_hermes_settings() {
let _env_guard = env_lock().lock().expect("env lock");
let _runtime_guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let mock = axum::Router::new().route(
"/v1/runs",
post(|| async { (StatusCode::BAD_REQUEST, "missing model.default and API key") }),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("listener");
let upstream = format!("http://{}", listener.local_addr().expect("addr"));
tokio::spawn(async move {
axum::serve(listener, mock).await.expect("mock upstream");
});
std::env::set_var("MNOTE_WEB_HERMES_UPSTREAM_URL", &upstream);
std::env::remove_var("MNOTE_HERMES_UPSTREAM_URL");
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/runs")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(
json!({
"documentId": "doc_error",
"sessionId": "mnote_doc_error_trace_1",
"profile": "chemist",
"message": "测试错误提示",
"traceId": "trace_error_1"
})
.to_string(),
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["code"], "hermes_client_upstream_error");
let message = payload["message"].as_str().unwrap_or_default();
assert!(message.contains("model.default"));
assert!(message.contains("API key"));
assert!(message.contains("Hermes 设置"));
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
}
#[tokio::test]
async fn hermes_client_unauthenticated_requests_return_401() {
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/sessions")
.header("content-type", "application/json")
.body(Body::from(json!({"documentId":"doc_1"}).to_string()))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
assert_eq!(
response
.headers()
.get("x-error-code")
.and_then(|value| value.to_str().ok()),
Some("hermes_client_unauthorized")
);
}
#[tokio::test]
async fn hermes_client_unconfigured_run_returns_stable_error() {
let _guard = env_lock().lock().expect("env lock");
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
std::env::remove_var("MNOTE_HERMES_UPSTREAM_URL");
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/runs")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(
json!({
"workspaceId": "ws_1",
"documentId": "doc_1",
"sessionId": "sess_1",
"message": "ping",
"traceId": "trace_1"
})
.to_string(),
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(
response
.headers()
.get("x-error-code")
.and_then(|value| value.to_str().ok()),
Some("hermes_client_unconfigured")
);
}
#[tokio::test]
async fn hermes_client_session_create_does_not_require_upstream_or_store_chat() {
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/sessions")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(
json!({
"workspaceId": "ws_1",
"documentId": "doc_1",
"traceId": "trace_1",
"title": "当前页问答"
})
.to_string(),
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["ok"], true);
assert_eq!(payload["sessionId"], "mnote_doc_1_trace_1");
assert_eq!(payload["persistence"], "hermes_on_first_run");
assert!(payload.get("messages").is_none());
}
#[tokio::test]
async fn hermes_client_session_create_carries_agent_profile_without_storing_chat() {
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/sessions")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(
json!({
"workspaceId": "ws_1",
"documentId": "doc_1",
"traceId": "trace_1",
"title": "当前页问答",
"profile": "chemist"
})
.to_string(),
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["sessionId"], "mnote_doc_1_trace_1");
assert_eq!(payload["profile"], "chemist");
assert!(payload.get("messages").is_none());
}
#[tokio::test]
async fn hermes_client_resume_returns_runtime_state_without_chat_store() {
let _guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let registration = HermesRunRegistration {
session_id: "mnote_doc_1_trace_1".into(),
profile: "chemist".into(),
document_id: "doc_1".into(),
trace_id: "trace_1".into(),
};
let runtime = register_runtime_from_create_run_response(
&registration,
&json!({
"ok": true,
"upstream": {"run_id": "run_1"}
}),
)
.expect("runtime");
assert_eq!(runtime["sessionId"], "mnote_doc_1_trace_1");
assert_eq!(runtime["runId"], "run_1");
assert!(runtime.get("messages").is_none());
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/sessions/mnote_doc_1_trace_1/resume")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::empty())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["ok"], true);
assert_eq!(
payload["session"]["messages"].as_array().map(Vec::len),
Some(0)
);
assert_eq!(payload["runtime"]["sessionId"], "mnote_doc_1_trace_1");
assert_eq!(payload["runtime"]["runId"], "run_1");
assert_eq!(payload["runtime"]["status"], "running");
}
#[test]
fn hermes_client_active_session_queues_second_run_without_chat_truth() {
let _guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs".parse().expect("uri"),
&HeaderMap::new(),
);
let registration = HermesRunRegistration {
session_id: "mnote_doc_queue_trace_1".into(),
profile: "chemist".into(),
document_id: "doc_queue".into(),
trace_id: "trace_queued_1".into(),
};
register_runtime_from_create_run_response(
&registration,
&json!({
"ok": true,
"upstream": {"run_id": "run_active_1"}
}),
)
.expect("runtime");
assert!(session_has_active_run("mnote_doc_queue_trace_1"));
let queued = enqueue_run(
&context,
&registration,
&json!({
"workspaceId": "ws_1",
"documentId": "doc_queue",
"sessionId": "mnote_doc_queue_trace_1",
"profile": "chemist",
"message": "第二条问题",
"traceId": "trace_queued_1",
"selectedText": "选区",
"pageContext": {
"title": "页面标题",
"body": "正文不应进入 queue 响应",
"blocks": [{"id": "block_1"}]
},
"messages": [
{"role": "assistant", "content": "历史消息不应保存为 mnote 真相"}
]
}),
)
.expect("queued");
assert_eq!(queued["queued"], true);
assert_eq!(queued["queueLength"], 1);
assert_eq!(queued["sessionId"], "mnote_doc_queue_trace_1");
assert!(queued.get("messages").is_none());
assert!(queued.get("pageContext").is_none());
assert_eq!(queued["contextSummary"]["hasPageContext"], true);
assert_eq!(queued["contextSummary"]["hasSelectedText"], true);
let popped = pop_next_queued_run("mnote_doc_queue_trace_1").expect("queued run");
assert_eq!(popped.input, "第二条问题");
assert_eq!(popped.profile, "chemist");
assert!(popped.context_summary.get("pageContext").is_none());
assert_eq!(queue_len_for_session("mnote_doc_queue_trace_1"), 0);
}
#[test]
fn hermes_client_cancel_queued_run_removes_pending_item() {
let _guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs".parse().expect("uri"),
&HeaderMap::new(),
);
let registration = HermesRunRegistration {
session_id: "mnote_doc_cancel_trace_1".into(),
profile: "default".into(),
document_id: "doc_cancel".into(),
trace_id: "trace_cancel_1".into(),
};
let queued = enqueue_run(
&context,
&registration,
&json!({
"documentId": "doc_cancel",
"sessionId": "mnote_doc_cancel_trace_1",
"message": "等待取消",
"traceId": "trace_cancel_1"
}),
)
.expect("queued");
let queue_id = queued["queueId"].as_str().expect("queue id");
assert_eq!(queue_len_for_session("mnote_doc_cancel_trace_1"), 1);
assert!(remove_queued_run("mnote_doc_cancel_trace_1", queue_id));
assert_eq!(queue_len_for_session("mnote_doc_cancel_trace_1"), 0);
assert!(pop_next_queued_run("mnote_doc_cancel_trace_1").is_none());
}
#[tokio::test]
async fn hermes_client_terminal_run_can_start_next_queued_run() {
let _env_guard = env_lock().lock().expect("env lock");
let _runtime_guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let hits = Arc::new(AtomicUsize::new(0));
let hits_for_route = Arc::clone(&hits);
let mock = axum::Router::new().route(
"/v1/runs",
post(move |Json(_body): Json<Value>| {
let hits = Arc::clone(&hits_for_route);
async move {
hits.fetch_add(1, Ordering::SeqCst);
Json(json!({
"run_id": "run_queued_next",
"trace_id": "trace_queued_next"
}))
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("listener");
let upstream = format!("http://{}", listener.local_addr().expect("addr"));
tokio::spawn(async move {
axum::serve(listener, mock).await.expect("mock upstream");
});
std::env::set_var("MNOTE_WEB_HERMES_UPSTREAM_URL", &upstream);
std::env::remove_var("MNOTE_HERMES_UPSTREAM_URL");
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs".parse().expect("uri"),
&HeaderMap::new(),
);
let registration = HermesRunRegistration {
session_id: "mnote_doc_auto_trace_1".into(),
profile: "chemist".into(),
document_id: "doc_auto".into(),
trace_id: "trace_auto_1".into(),
};
register_runtime_from_create_run_response(
&registration,
&json!({
"ok": true,
"upstream": {"run_id": "run_active_auto"}
}),
)
.expect("runtime");
enqueue_run(
&context,
&registration,
&json!({
"workspaceId": "ws_1",
"documentId": "doc_auto",
"sessionId": "mnote_doc_auto_trace_1",
"profile": "chemist",
"message": "自动出队问题",
"traceId": "trace_auto_1",
"pageContext": {"title": "自动出队"}
}),
)
.expect("queued");
let (_, terminal_sessions) = normalize_sse_chunk(
"run_active_auto",
&format!(
"data: {}\n\n",
json!({
"event": "run.completed",
"run_id": "run_active_auto",
"session_id": "mnote_doc_auto_trace_1"
})
),
);
assert_eq!(
terminal_sessions,
vec!["mnote_doc_auto_trace_1".to_string()]
);
start_next_queued_run(context, "mnote_doc_auto_trace_1".into()).await;
assert_eq!(hits.load(Ordering::SeqCst), 1);
assert_eq!(queue_len_for_session("mnote_doc_auto_trace_1"), 0);
let runtime = runtime_state_for_session("mnote_doc_auto_trace_1");
assert_eq!(runtime["runId"], "run_queued_next");
assert_eq!(runtime["status"], "running");
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
}
#[tokio::test]
async fn hermes_client_abort_returns_events_and_continues_queue() {
let _env_guard = env_lock().lock().expect("env lock");
let _runtime_guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let stop_hits = Arc::new(AtomicUsize::new(0));
let run_hits = Arc::new(AtomicUsize::new(0));
let stop_hits_for_route = Arc::clone(&stop_hits);
let run_hits_for_route = Arc::clone(&run_hits);
let mock = axum::Router::new()
.route(
"/v1/runs/{run_id}/stop",
post(move || {
let stop_hits = Arc::clone(&stop_hits_for_route);
async move {
stop_hits.fetch_add(1, Ordering::SeqCst);
Json(json!({"ok": true, "status": "aborted"}))
}
}),
)
.route(
"/v1/runs",
post(move |Json(_body): Json<Value>| {
let run_hits = Arc::clone(&run_hits_for_route);
async move {
run_hits.fetch_add(1, Ordering::SeqCst);
Json(json!({
"run_id": "run_after_abort",
"trace_id": "trace_after_abort"
}))
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("listener");
let upstream = format!("http://{}", listener.local_addr().expect("addr"));
tokio::spawn(async move {
axum::serve(listener, mock).await.expect("mock upstream");
});
std::env::set_var("MNOTE_WEB_HERMES_UPSTREAM_URL", &upstream);
std::env::remove_var("MNOTE_HERMES_UPSTREAM_URL");
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs/run_abort_1/abort"
.parse()
.expect("uri"),
&HeaderMap::new(),
);
let registration = HermesRunRegistration {
session_id: "mnote_doc_abort_trace_1".into(),
profile: "default".into(),
document_id: "doc_abort".into(),
trace_id: "trace_abort_1".into(),
};
register_runtime_from_create_run_response(
&registration,
&json!({
"ok": true,
"upstream": {"run_id": "run_abort_1"}
}),
)
.expect("runtime");
enqueue_run(
&context,
&registration,
&json!({
"documentId": "doc_abort",
"sessionId": "mnote_doc_abort_trace_1",
"message": "abort 后继续",
"traceId": "trace_abort_1"
}),
)
.expect("queued");
let response = app()
.oneshot(
Request::builder()
.method("POST")
.uri("/api/hermes/client/runs/run_abort_1/abort")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(json!({"reason":"test_abort"}).to_string()))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("json");
assert_eq!(payload["events"][0]["event"], "abort.started");
assert_eq!(payload["events"][1]["event"], "abort.completed");
assert_eq!(stop_hits.load(Ordering::SeqCst), 1);
assert_eq!(run_hits.load(Ordering::SeqCst), 1);
assert_eq!(queue_len_for_session("mnote_doc_abort_trace_1"), 0);
let runtime = runtime_state_for_session("mnote_doc_abort_trace_1");
assert_eq!(runtime["runId"], "run_after_abort");
assert_eq!(runtime["status"], "running");
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
}
#[test]
fn hermes_client_runtime_registry_tracks_tool_event_summary_only() {
let _guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let registration = HermesRunRegistration {
session_id: "mnote_doc_2_trace_2".into(),
profile: "default".into(),
document_id: "doc_2".into(),
trace_id: "trace_2".into(),
};
register_runtime_from_create_run_response(
&registration,
&json!({
"ok": true,
"runId": "run_2"
}),
)
.expect("runtime");
let _ = normalize_sse_chunk(
"run_2",
&format!(
"data: {}\n\n",
json!({
"event": "tool.started",
"run_id": "run_2",
"session_id": "mnote_doc_2_trace_2",
"name": "mnote.page.get",
"toolCallId": "call_1",
"arguments": {"includeBody": false}
})
),
);
let runtime = runtime_state_for_session("mnote_doc_2_trace_2");
assert_eq!(runtime["status"], "tool_calling");
assert_eq!(runtime["lastToolName"], "mnote.page.get");
assert_eq!(runtime["lastToolCallId"], "call_1");
assert!(runtime.get("arguments").is_none());
assert!(runtime.get("messages").is_none());
}
#[test]
fn hermes_client_normalizes_upstream_sse_events_to_runtime_contract() {
let _guard = runtime_lock().lock().expect("runtime lock");
clear_runtime_registry();
clear_run_queue();
let registration = HermesRunRegistration {
session_id: "mnote_doc_norm_trace_1".into(),
profile: "default".into(),
document_id: "doc_norm".into(),
trace_id: "trace_norm_1".into(),
};
register_runtime_from_create_run_response(
&registration,
&json!({
"ok": true,
"runId": "run_norm_1"
}),
)
.expect("runtime");
let (normalized, terminal_sessions) = normalize_sse_chunk(
"run_norm_1",
&format!(
"data: {}\n\ndata: {}\n\n",
json!({
"event": "assistant_message",
"run_id": "run_norm_1",
"session_id": "mnote_doc_norm_trace_1",
"text": "hello"
}),
json!({
"event": "tool.failed",
"run_id": "run_norm_1",
"session_id": "mnote_doc_norm_trace_1",
"toolCallId": "call_failed_1",
"name": "mnote.page.save",
"code": "permission_denied",
"error": "写入被拒绝"
})
),
);
assert!(normalized.contains("event: message.delta"));
assert!(normalized.contains("\"schemaVersion\":\"mnote.hermes_runtime_event.v1\""));
assert!(normalized.contains("\"event\":\"tool.failed\""));
assert!(normalized.contains("\"toolCallId\":\"call_failed_1\""));
assert!(normalized.contains("\"code\":\"permission_denied\""));
assert!(terminal_sessions.is_empty());
}
#[test]
fn hermes_client_run_body_carries_minimal_page_context_into_run_input() {
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs".parse().expect("uri"),
&HeaderMap::new(),
);
let body = build_run_upstream_body(
&context,
json!({
"workspaceId": "ws_1",
"documentId": "doc_1",
"sessionId": "sess_1",
"message": "概括当前页面",
"pageContext": {
"contextScope": "page",
"node": {"documentId": "doc_1", "title": "页面标题"},
"documentBlocks": [{"type": "paragraph", "text": "不应进入 Hermes instructions"}],
"subtree": {"children": [{"title": "不应进入 Hermes instructions"}]},
"outline": [{"title": "不应进入 Hermes instructions"}],
"contentAccess": "mnote.page.get",
"aiContext": {
"schema": "mnote.page_ai_context.v1",
"contextBlocks": [{"blockId": "block_1", "text": "允许进入 Hermes instructions"}],
"pageXml": "<page><block id=\"block_1\">允许进入 Hermes instructions</block></page>",
"allowedTargetBlockIds": ["block_1"]
}
},
"selectedBlockId": "block_1",
"selectedText": "选中文本",
"traceId": "trace_1"
}),
)
.expect("body");
assert_eq!(body["input"], "概括当前页面");
assert_eq!(body["session_id"], "sess_1");
let instructions = body["instructions"].as_str().expect("instructions");
assert!(instructions.contains("\"workspaceId\":\"ws_1\""));
assert!(instructions.contains("\"documentId\":\"doc_1\""));
assert!(instructions.contains("\"title\":\"页面标题\""));
assert!(instructions.contains("\"selectedBlockId\":\"block_1\""));
assert!(instructions.contains("\"contentAccess\":\"mnote.page.get\""));
assert!(instructions.contains("\"schema\":\"mnote.page_ai_context.v1\""));
assert!(instructions.contains("允许进入 Hermes instructions"));
assert!(!instructions.contains("不应进入 Hermes instructions"));
assert!(!instructions.contains("\"documentBlocks\""));
assert!(!instructions.contains("\"subtree\""));
assert!(!instructions.contains("\"outline\""));
}
#[test]
fn hermes_client_run_actor_falls_back_to_dev_user_for_cookie_auth() {
let state = test_state();
let mut headers = HeaderMap::new();
headers.insert("cookie", "mnote_web_convex_token=token_1".parse().unwrap());
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs".parse().expect("uri"),
&headers,
);
let (actor_id, actor_type) = effective_run_actor(&state, &context);
assert_eq!(actor_id, "dev-user");
assert_eq!(actor_type, "devFallback");
let mut payload = json!({"message": "读取当前页面"});
stamp_run_actor(&mut payload, &actor_id, &actor_type);
let body = build_run_upstream_body(&context, payload).expect("body");
let instructions = body["instructions"].as_str().expect("instructions");
assert!(instructions.contains("\"actorId\":\"dev-user\""));
assert!(instructions.contains("\"actorType\":\"devFallback\""));
assert!(!instructions.contains("\"actorId\":\"anonymous\""));
}
#[test]
fn hermes_client_run_body_carries_agent_profile() {
let context = RequestContext::from_http_parts(
&axum::http::Method::POST,
&"/api/hermes/client/runs".parse().expect("uri"),
&HeaderMap::new(),
);
let body = build_run_upstream_body(
&context,
json!({
"workspaceId": "ws_1",
"documentId": "doc_1",
"sessionId": "sess_1",
"message": "用当前 agent 回答",
"profile": "chemist",
"traceId": "trace_1"
}),
)
.expect("body");
assert_eq!(body["profile"], "chemist");
let instructions = body["instructions"].as_str().expect("instructions");
assert!(instructions.contains("\"profile\":\"chemist\""));
}
#[tokio::test]
async fn hermes_client_profile_skill_and_memory_routes_use_local_bff_without_upstream() {
let _guard = env_lock().lock().expect("env lock");
std::env::remove_var("MNOTE_WEB_HERMES_UPSTREAM_URL");
std::env::remove_var("MNOTE_HERMES_UPSTREAM_URL");
let hermes_home = std::env::temp_dir().join(format!(
"mnote-web-hermes-client-test-{}",
std::process::id()
));
let _ = fs::remove_dir_all(&hermes_home);
for name in ["draft", "installed-demo", "builtin-demo"] {
fs::create_dir_all(hermes_home.join("skills").join("writing").join(name))
.expect("test skill dir");
fs::write(
hermes_home
.join("skills")
.join("writing")
.join(name)
.join("SKILL.md"),
format!("# {name}\n\n写作助手"),
)
.expect("test skill");
}
fs::create_dir_all(hermes_home.join("skills").join("copied-demo"))
.expect("test copied skill dir");
fs::write(
hermes_home
.join("skills")
.join("copied-demo")
.join("SKILL.md"),
"# copied-demo\n\n复制技能",
)
.expect("test copied skill");
fs::write(
hermes_home.join("skills").join(".bundled_manifest"),
"builtin-demo:abc123\n",
)
.expect("test bundled manifest");
fs::create_dir_all(hermes_home.join("skills").join(".hub")).expect("test hub dir");
fs::write(
hermes_home.join("skills").join(".hub").join("lock.json"),
json!({
"installed": {
"installed-demo": {
"source": "skills.sh",
"install_path": "writing/installed-demo"
}
}
})
.to_string(),
)
.expect("test hub lock");
fs::write(
hermes_home.join("skills").join(".usage.json"),
json!({
"draft": {
"created_by": "agent",
"created_at": "2026-05-14T00:00:00Z",
"patch_count": 2
}
})
.to_string(),
)
.expect("test usage");
std::env::set_var("HERMES_HOME", &hermes_home);
std::env::set_var("MNOTE_WEB_HERMES_BIN", "__missing_hermes_for_test__");
let cases = [
("GET", "/api/hermes/client/profiles", None),
("GET", "/api/hermes/client/profiles/chemist", None),
(
"GET",
"/api/hermes/client/profile-memory?profile=chemist",
None,
),
("GET", "/api/hermes/client/skills?profile=chemist", None),
(
"PUT",
"/api/hermes/client/profiles/active",
Some(json!({"name": "chemist"})),
),
(
"POST",
"/api/hermes/client/profile-memory",
Some(json!({"profile": "chemist", "section": "soul", "content": "test"})),
),
(
"PUT",
"/api/hermes/client/skills/toggle",
Some(json!({"name": "chem", "enabled": true})),
),
];
for (method, uri, body) in cases {
let request = Request::builder()
.method(method)
.uri(uri)
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(body.unwrap_or_else(|| json!({})).to_string()))
.expect("request");
let response = app().oneshot(request).await.expect("response");
assert_eq!(response.status(), StatusCode::OK, "{uri}");
}
assert_eq!(
fs::read_to_string(hermes_home.join("active_profile"))
.expect("active profile")
.trim(),
"chemist"
);
assert_eq!(
fs::read_to_string(hermes_home.join("profiles").join("chemist").join("SOUL.md"))
.expect("soul"),
"test"
);
let request = Request::builder()
.method("GET")
.uri("/api/hermes/client/skills?profile=chemist")
.header("x-mnote-actor-id", "user_1")
.body(Body::empty())
.expect("request");
let response = app().oneshot(request).await.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("skills json");
let skills = payload["categories"]
.as_array()
.expect("categories")
.iter()
.flat_map(|category| category["skills"].as_array().into_iter().flatten())
.map(|skill| {
(
skill["name"].as_str().unwrap_or_default().to_string(),
skill.clone(),
)
})
.collect::<HashMap<_, _>>();
assert_eq!(skills["draft"]["origin"], "generated");
assert_eq!(skills["draft"]["createdBy"], "agent");
assert_eq!(skills["installed-demo"]["source"], "hub");
assert_eq!(skills["installed-demo"]["origin"], "installed");
assert_eq!(skills["builtin-demo"]["source"], "builtin");
assert_eq!(skills["copied-demo"]["origin"], "copied");
std::env::remove_var("HERMES_HOME");
std::env::remove_var("MNOTE_WEB_HERMES_BIN");
let _ = fs::remove_dir_all(&hermes_home);
}
#[tokio::test]
async fn hermes_client_tools_uses_manifest_and_profile_disabled_state() {
let _guard = env_lock().lock().expect("env lock");
let hermes_home = std::env::temp_dir().join(format!(
"mnote-web-hermes-tools-local-{}",
std::process::id()
));
let _ = fs::remove_dir_all(&hermes_home);
let profile_dir = hermes_home.join("profiles").join("chemist");
fs::create_dir_all(&profile_dir).expect("profile dir");
fs::write(
profile_dir.join("config.yaml"),
"mnote:\n tools:\n disabled:\n - mnote.block.replace\n",
)
.expect("profile config");
std::env::set_var("HERMES_HOME", &hermes_home);
let response = app()
.oneshot(
Request::builder()
.method("GET")
.uri("/api/hermes/client/tools?scope=mnote&profile=chemist")
.header("x-mnote-actor-id", "user_1")
.body(Body::empty())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("tools json");
let tools = payload["tools"]
.as_array()
.expect("tools")
.iter()
.map(|tool| {
(
tool["name"].as_str().unwrap_or_default().to_string(),
tool.clone(),
)
})
.collect::<HashMap<_, _>>();
assert!(tools.contains_key("mnote.doc.fetch"));
assert!(tools.contains_key("mnote.block.fetch"));
assert!(tools.contains_key("mnote.block.replace"));
assert!(tools.contains_key("mnote.block.insert_after"));
assert!(tools.contains_key("mnote.block.delete"));
assert!(tools.contains_key("mnote.block.move_after"));
assert!(tools.contains_key("mnote.doc.apply_block_ops"));
assert_eq!(tools["mnote.doc.fetch"]["enabled"], true);
assert_eq!(tools["mnote.block.replace"]["enabled"], false);
assert_eq!(tools["mnote.block.replace"]["status"], "disabled");
assert_eq!(
tools["mnote.block.replace"]["unavailableReason"],
"当前 Hermes profile 已关闭该 mnote tool"
);
let toggle_response = app()
.oneshot(
Request::builder()
.method("PUT")
.uri("/api/hermes/client/tools/toggle")
.header("content-type", "application/json")
.header("x-mnote-actor-id", "user_1")
.body(Body::from(
json!({
"profile": "chemist",
"name": "mnote.block.replace",
"enabled": true
})
.to_string(),
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(toggle_response.status(), StatusCode::OK);
let response = app()
.oneshot(
Request::builder()
.method("GET")
.uri("/api/hermes/client/tools?scope=mnote&profile=chemist")
.header("x-mnote-actor-id", "user_1")
.body(Body::empty())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
let payload: Value = serde_json::from_slice(&body).expect("tools json");
let tools = payload["tools"]
.as_array()
.expect("tools")
.iter()
.map(|tool| {
(
tool["name"].as_str().unwrap_or_default().to_string(),
tool.clone(),
)
})
.collect::<HashMap<_, _>>();
assert_eq!(tools["mnote.block.replace"]["enabled"], true);
assert_eq!(tools["mnote.block.replace"]["status"], "available");
std::env::remove_var("HERMES_HOME");
let _ = fs::remove_dir_all(&hermes_home);
}
}