chore: move 3-14 WS push design to done/

2026-05-17 16:15:52 +08:00
parent 3f43020603
commit 2ea559beaa
22 changed files with 5283 additions and 24 deletions
@@ -1,9 +1,11 @@
+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 tokio::sync::broadcast;
 use axum::extract::{Extension, Path, Query, State};
 use axum::http::{header, HeaderMap, HeaderName, HeaderValue, StatusCode};
 use axum::response::Response;
@@ -15,10 +17,10 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::fs;
 use std::path::{Path as FsPath, PathBuf};
 use std::process::Command;
-use std::sync::{LazyLock, Mutex};
+use std::sync::{Arc, LazyLock, Mutex};
 use std::time::Duration;
 use std::time::{SystemTime, UNIX_EPOCH};
-use tracing::warn;
+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";
@@ -27,6 +29,9 @@ static HERMES_RUNTIME_REGISTRY: LazyLock<Mutex<HashMap<String, HermesRuntimeStat
    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()));

 #[derive(Debug, Clone)]
 struct HermesRuntimeState {
@@ -522,6 +527,27 @@ pub async fn create_run(
    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 is_acp_profile(&registration.profile) {
+        let run_id = registration.session_id.clone(); // session_id serves as run_id
+        let runtime_state = register_acp_runtime(&registration);
+        // 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)));
@@ -567,12 +593,148 @@ pub async fn cancel_queued_run(
    ))
 }

+/// 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,
+) -> 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 prompt_blocks = vec![ContentBlock::Text {
+        text: input.to_string(),
+    }];
+
+    let runtime_name = crate::acp_bridge::runtime_name_for_profile(profile);
+
+    // Ensure runtime is active; switch_to either activates it or returns existing
+    let client = 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);
+        }
+    });
+
+    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)
+        })?;
+
+    // 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() }),
+                });
+            }
+        }
+        update_runtime_by_run_id(&run_id_owned, "completed", Some("acp.prompt.done"), None);
+    });
+
+    // Build SSE response from event channel
+    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,
+            }
+        }
+    });
+
+    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 is_acp_profile(&profile) {
+        return acp_stream_events(state, context, &run_id, &profile).await;
+    }
+
    let Some(upstream) = configured_upstream_for_profile(&profile) else {
        return Err(hermes_unconfigured_error(&context));
    };
@@ -856,9 +1018,28 @@ fn list_profiles_payload() -> Value {
        .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
+        "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 协议直连 Reasonix（DeepSeek 缓存优先）",
+                "model": reasonix_model,
+                "preset": reasonix_preset,
+                "apiKeyConfigured": reasonix_has_key,
+                "version": "0.43.0"
+            }
+        ])
    })
 }

@@ -1472,6 +1653,41 @@ fn profile_env_key(prefix: &str, profile: &str, suffix: &str) -> 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)
+}
+
+/// 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" {
@@ -2050,6 +2266,33 @@ fn run_registration_from_payload(
    }
 }

+/// Register a runtime state from a registration (without upstream response).
+/// Used by the ACP path where no Hermes HTTP upstream exists.
+fn register_acp_runtime(registration: &HermesRunRegistration) -> Value {
+    let now = now_ms();
+    let run_id = registration.session_id.clone(); // use session_id as run_id for ACP
+    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: "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,