lix-2026
1882db7681
- 归档 OnlyOffice live bridge、Page AI、mindmap、design governance 与相关 bug 条目 - 补齐 MinerU OCR 后端 runtime 合同与 smoke/test 基线 - 收口 ChatOnly/Doubao、ObjectIdentity、Page Aggregate compat 与 runtime owner 文档口径 验证： - cargo test --manifest-path rust/Cargo.toml -p mnote-web local_ocr -- --test-threads=1 - cargo test --manifest-path rust/Cargo.toml -p mnote-web onlyoffice_bridge -- --test-threads=1 - git diff --check - git diff --cached --check - codegraph index . --force && codegraph status . - codegraph sync . && codegraph status .
459 lines
16 KiB
Rust
459 lines
16 KiB
Rust
/// ACP ↔ SSE bridge for Hermes route integration.
|
|
///
|
|
/// Transforms ACP session events into the SSE event format expected by the
|
|
/// frontend (HermesRunEvent), and manages the background prompt lifecycle.
|
|
///
|
|
/// Reference: `hermes-vscode-main/src/sessionManager.ts` handleUpdate()
|
|
use crate::acp_runtime::AcpRuntimeManager;
|
|
use crate::acp_session_manager::{AcpSessionEvent, AcpSessionManager};
|
|
use crate::acp_types::ContentBlock;
|
|
use axum::body::Body;
|
|
use axum::http::{header, StatusCode};
|
|
use axum::response::Response;
|
|
use serde_json::{json, Value};
|
|
use std::sync::Arc;
|
|
use tokio::sync::broadcast;
|
|
use tracing::{info, warn};
|
|
|
|
/// Errors from the ACP bridge.
|
|
#[derive(Debug)]
|
|
pub enum AcpBridgeError {
|
|
NoActiveRuntime,
|
|
SessionError(String),
|
|
StreamError(String),
|
|
}
|
|
|
|
impl std::fmt::Display for AcpBridgeError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
AcpBridgeError::NoActiveRuntime => write!(f, "no active ACP runtime"),
|
|
AcpBridgeError::SessionError(msg) => write!(f, "ACP session error: {msg}"),
|
|
AcpBridgeError::StreamError(msg) => write!(f, "ACP stream error: {msg}"),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// SSE event types sent to the frontend.
|
|
/// Mirrors HermesRunEvent from bridge.ts.
|
|
#[derive(Debug, Clone)]
|
|
pub struct SseEvent {
|
|
pub event: String,
|
|
pub data: Value,
|
|
}
|
|
|
|
/// Bridge state for one run: holds the broadcast channel for SSE events.
|
|
pub struct AcpRunBridge {
|
|
session_id: String,
|
|
event_tx: broadcast::Sender<SseEvent>,
|
|
}
|
|
|
|
impl AcpRunBridge {
|
|
/// Create a new ACP run: create session + start prompt in background.
|
|
///
|
|
/// Returns a bridge with a broadcast receiver that the SSE endpoint can use.
|
|
pub async fn start(
|
|
runtime_mgr: &AcpRuntimeManager,
|
|
runtime_name: &str,
|
|
prompt_blocks: Vec<ContentBlock>,
|
|
) -> Result<Self, AcpBridgeError> {
|
|
// Get or activate the runtime
|
|
let client = if runtime_mgr.is_active().await {
|
|
runtime_mgr
|
|
.active_client()
|
|
.await
|
|
.ok_or(AcpBridgeError::NoActiveRuntime)?
|
|
} else {
|
|
runtime_mgr
|
|
.switch_to(runtime_name)
|
|
.await
|
|
.map_err(|e| AcpBridgeError::SessionError(e.to_string()))?
|
|
};
|
|
|
|
// Create session manager
|
|
let mgr = Arc::new(AcpSessionManager::new(client));
|
|
|
|
// Create event channel (256 buffered, enough for SSE streaming)
|
|
let (event_tx, _) = broadcast::channel(256);
|
|
let event_tx_clone = event_tx.clone();
|
|
|
|
// Set up event handler
|
|
mgr.on_event(move |event| {
|
|
if let Some(sse) = acp_event_to_sse(event) {
|
|
let _ = event_tx_clone.send(sse);
|
|
}
|
|
});
|
|
|
|
// Create session
|
|
let sid = mgr
|
|
.create_session(None, None)
|
|
.await
|
|
.map_err(|e| AcpBridgeError::SessionError(e.to_string()))?;
|
|
|
|
// Start prompt in background
|
|
let mgr_clone = mgr.clone();
|
|
let event_tx_prompt = event_tx.clone();
|
|
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(SseEvent {
|
|
event: "run.completed".into(),
|
|
data: json!({
|
|
"stopReason": format!("{:?}", result.stop_reason),
|
|
}),
|
|
});
|
|
}
|
|
Err(e) => {
|
|
warn!("ACP prompt failed: {e}");
|
|
let _ = event_tx_prompt.send(SseEvent {
|
|
event: "run.failed".into(),
|
|
data: json!({ "error": e.to_string() }),
|
|
});
|
|
}
|
|
}
|
|
});
|
|
|
|
info!("ACP run started: session={}", sid);
|
|
Ok(Self {
|
|
session_id: sid,
|
|
event_tx,
|
|
})
|
|
}
|
|
|
|
/// Cancel the current run.
|
|
pub async fn abort(&self) {
|
|
// Cancellation is sent via the session manager.
|
|
// For now, we just drop the bridge — the background task will detect this
|
|
// via the broadcast channel being closed.
|
|
info!("ACP run aborted: session={}", self.session_id);
|
|
}
|
|
|
|
/// Create an SSE response body from the event broadcast receiver.
|
|
pub fn into_sse_response(self) -> Response {
|
|
use tokio::sync::mpsc;
|
|
let (tx, rx) = mpsc::channel::<Result<axum::body::Bytes, std::convert::Infallible>>(256);
|
|
let mut broadcast_rx = self.event_tx.subscribe();
|
|
|
|
// Forward events from broadcast to mpsc
|
|
tokio::spawn(async move {
|
|
loop {
|
|
match broadcast_rx.recv().await {
|
|
Ok(event) => {
|
|
let json =
|
|
serde_json::to_string(&event.data).unwrap_or_else(|_| "{}".into());
|
|
let bytes = axum::body::Bytes::from(format!(
|
|
"event: {}\ndata: {}\n\n",
|
|
event.event, json
|
|
));
|
|
if tx.send(Ok(bytes)).await.is_err() {
|
|
break; // receiver dropped
|
|
}
|
|
}
|
|
Err(broadcast::error::RecvError::Lagged(n)) => {
|
|
warn!("ACP SSE lagged: {n} events dropped");
|
|
}
|
|
Err(broadcast::error::RecvError::Closed) => {
|
|
break; // stream ended
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
|
|
|
|
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))
|
|
.unwrap_or_else(|_| {
|
|
Response::builder()
|
|
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
|
.body(Body::empty())
|
|
.unwrap()
|
|
})
|
|
}
|
|
}
|
|
|
|
/// Map an AcpSessionEvent to an SSE event for the frontend.
|
|
///
|
|
/// Reference: hermes-vscode-main protocol.ts extractTextContent/parseToolCall/parseToolCallUpdate
|
|
/// Reference: recycle/wolai-frontend bridge.ts HermesRunEvent type (historic)
|
|
pub fn acp_event_to_sse(event: AcpSessionEvent) -> Option<SseEvent> {
|
|
match event {
|
|
AcpSessionEvent::TextDelta { text } => Some(SseEvent {
|
|
event: "message.delta".into(),
|
|
data: json!({ "delta": text }),
|
|
}),
|
|
AcpSessionEvent::ThoughtDelta { text } => Some(SseEvent {
|
|
event: "thought.delta".into(),
|
|
data: json!({ "delta": text }),
|
|
}),
|
|
AcpSessionEvent::ToolCall {
|
|
tool_call_id,
|
|
title,
|
|
kind,
|
|
status,
|
|
raw_input,
|
|
locations,
|
|
} => Some(SseEvent {
|
|
event: "tool.started".into(),
|
|
data: json!({
|
|
"tool": title,
|
|
"toolCallId": tool_call_id,
|
|
"kind": kind,
|
|
"status": status,
|
|
"input": raw_input,
|
|
"locations": locations,
|
|
}),
|
|
}),
|
|
AcpSessionEvent::ToolCallUpdate {
|
|
tool_call_id,
|
|
status,
|
|
content,
|
|
} => {
|
|
let error = status == crate::acp_types::ToolCallStatus::Failed;
|
|
let event = if error {
|
|
"tool.failed"
|
|
} else if status == crate::acp_types::ToolCallStatus::Completed {
|
|
"tool.completed"
|
|
} else {
|
|
"tool.started"
|
|
};
|
|
Some(SseEvent {
|
|
event: event.into(),
|
|
data: json!({
|
|
"toolCallId": tool_call_id,
|
|
"status": status,
|
|
"error": error,
|
|
"output": content,
|
|
}),
|
|
})
|
|
}
|
|
AcpSessionEvent::UsageUpdate { used, size } => Some(SseEvent {
|
|
event: "usage.updated".into(),
|
|
data: json!({ "used": used, "size": size }),
|
|
}),
|
|
AcpSessionEvent::PermissionRequest {
|
|
permission_id,
|
|
tool_name,
|
|
params,
|
|
decision,
|
|
} => {
|
|
let event = match decision.as_str() {
|
|
"allowed" => "permission.allowed",
|
|
"requested" => "permission.requested",
|
|
_ => "permission.denied",
|
|
};
|
|
Some(SseEvent {
|
|
event: event.into(),
|
|
data: json!({
|
|
"permissionId": permission_id,
|
|
"toolName": tool_name,
|
|
"params": params,
|
|
"decision": decision,
|
|
}),
|
|
})
|
|
}
|
|
AcpSessionEvent::SessionInfoUpdate { title } => Some(SseEvent {
|
|
event: "session.info.updated".into(),
|
|
data: json!({ "title": title }),
|
|
}),
|
|
AcpSessionEvent::ProviderConversationBound {
|
|
provider,
|
|
remote_conversation_id,
|
|
remote_url,
|
|
acp_session_id,
|
|
} => Some(SseEvent {
|
|
event: "provider.conversation.bound".into(),
|
|
data: json!({
|
|
"provider": provider,
|
|
"remoteConversationId": remote_conversation_id,
|
|
"remoteUrl": remote_url,
|
|
"acpSessionId": acp_session_id,
|
|
}),
|
|
}),
|
|
AcpSessionEvent::PlanUpdate { entries } => Some(SseEvent {
|
|
event: "plan.updated".into(),
|
|
data: json!({ "entries": entries }),
|
|
}),
|
|
AcpSessionEvent::Disconnected { reason } if reason == "session closed" => None,
|
|
AcpSessionEvent::Disconnected { reason } => Some(SseEvent {
|
|
event: "run.failed".into(),
|
|
data: json!({ "error": reason }),
|
|
}),
|
|
}
|
|
}
|
|
|
|
/// Helper: get the runtime name from a profile.
|
|
/// For now, we use "hermes" or "reasonix" directly.
|
|
/// In Step 12, this will come from the profile config.
|
|
pub fn runtime_name_for_profile(profile: &str) -> &str {
|
|
match profile {
|
|
"reasonix" => "reasonix",
|
|
_ => "hermes",
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn acp_normal_session_close_does_not_emit_failure() {
|
|
let event = AcpSessionEvent::Disconnected {
|
|
reason: "session closed".into(),
|
|
};
|
|
|
|
assert!(acp_event_to_sse(event).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn acp_unexpected_disconnect_emits_failure() {
|
|
let event = AcpSessionEvent::Disconnected {
|
|
reason: "transport lost".into(),
|
|
};
|
|
|
|
let sse = acp_event_to_sse(event).expect("unexpected disconnect should be forwarded");
|
|
assert_eq!(sse.event, "run.failed");
|
|
assert_eq!(sse.data["error"], "transport lost");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_thought_delta_does_not_emit_message_delta() {
|
|
let event = AcpSessionEvent::ThoughtDelta {
|
|
text: "internal reasoning".into(),
|
|
};
|
|
|
|
let sse = acp_event_to_sse(event).expect("thought delta should be forwarded separately");
|
|
assert_eq!(sse.event, "thought.delta");
|
|
assert_eq!(sse.data["delta"], "internal reasoning");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_permission_request_emits_frontend_decision_event() {
|
|
let event = AcpSessionEvent::PermissionRequest {
|
|
permission_id: "perm_1".into(),
|
|
tool_name: "mnote.page.save".into(),
|
|
params: json!({"documentId": "doc_1"}),
|
|
decision: "denied".into(),
|
|
};
|
|
|
|
let sse = acp_event_to_sse(event).expect("permission decision should be forwarded");
|
|
assert_eq!(sse.event, "permission.denied");
|
|
assert_eq!(sse.data["permissionId"], "perm_1");
|
|
assert_eq!(sse.data["toolName"], "mnote.page.save");
|
|
assert_eq!(sse.data["params"]["documentId"], "doc_1");
|
|
assert_eq!(sse.data["decision"], "denied");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_permission_requested_emits_permission_requested_event() {
|
|
let event = AcpSessionEvent::PermissionRequest {
|
|
permission_id: "perm_2".into(),
|
|
tool_name: "mnote.page.get".into(),
|
|
params: json!({"documentId": "doc_2"}),
|
|
decision: "requested".into(),
|
|
};
|
|
|
|
let sse = acp_event_to_sse(event).expect("permission requested should be forwarded");
|
|
assert_eq!(sse.event, "permission.requested");
|
|
assert_eq!(sse.data["permissionId"], "perm_2");
|
|
assert_eq!(sse.data["toolName"], "mnote.page.get");
|
|
assert_eq!(sse.data["decision"], "requested");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_permission_allowed_emits_permission_allowed_event() {
|
|
let event = AcpSessionEvent::PermissionRequest {
|
|
permission_id: "perm_3".into(),
|
|
tool_name: "mnote.page.save".into(),
|
|
params: json!({"documentId": "doc_1"}),
|
|
decision: "allowed".into(),
|
|
};
|
|
|
|
let sse = acp_event_to_sse(event).expect("permission allowed should be forwarded");
|
|
assert_eq!(sse.event, "permission.allowed");
|
|
assert_eq!(sse.data["permissionId"], "perm_3");
|
|
assert_eq!(sse.data["decision"], "allowed");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_tool_events_keep_detail_for_collapsible_ui() {
|
|
let started = acp_event_to_sse(AcpSessionEvent::ToolCall {
|
|
tool_call_id: "tool_1".into(),
|
|
title: "mnote.page.get".into(),
|
|
kind: "read".into(),
|
|
status: crate::acp_types::ToolCallStatus::InProgress,
|
|
raw_input: Some(json!({"documentId": "doc_1", "includeBody": true})),
|
|
locations: vec!["/mnt/Data1T/mnote/src/main.rs".into()],
|
|
})
|
|
.expect("tool start");
|
|
assert_eq!(started.event, "tool.started");
|
|
assert_eq!(started.data["tool"], "mnote.page.get");
|
|
assert_eq!(started.data["status"], "in_progress");
|
|
assert_eq!(started.data["input"]["documentId"], "doc_1");
|
|
assert_eq!(
|
|
started.data["locations"][0],
|
|
"/mnt/Data1T/mnote/src/main.rs"
|
|
);
|
|
assert_eq!(started.data["locations"].as_array().unwrap().len(), 1);
|
|
|
|
let completed = acp_event_to_sse(AcpSessionEvent::ToolCallUpdate {
|
|
tool_call_id: "tool_1".into(),
|
|
status: crate::acp_types::ToolCallStatus::Completed,
|
|
content: Some(vec![crate::acp_types::ContentBlockWrapper {
|
|
wrapper_type: "content".into(),
|
|
content: crate::acp_types::TextContent {
|
|
content_type: "text".into(),
|
|
text: "读取完成".into(),
|
|
},
|
|
}]),
|
|
})
|
|
.expect("tool complete");
|
|
assert_eq!(completed.event, "tool.completed");
|
|
assert_eq!(completed.data["status"], "completed");
|
|
assert_eq!(completed.data["output"][0]["content"]["text"], "读取完成");
|
|
|
|
let running = acp_event_to_sse(AcpSessionEvent::ToolCallUpdate {
|
|
tool_call_id: "tool_1".into(),
|
|
status: crate::acp_types::ToolCallStatus::InProgress,
|
|
content: None,
|
|
})
|
|
.expect("tool running");
|
|
assert_eq!(running.event, "tool.started");
|
|
assert_eq!(running.data["status"], "in_progress");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_session_info_update_emits_session_info_updated_sse() {
|
|
let sse = acp_event_to_sse(AcpSessionEvent::SessionInfoUpdate {
|
|
title: "我的新会话标题".into(),
|
|
})
|
|
.expect("session info update should be forwarded");
|
|
assert_eq!(sse.event, "session.info.updated");
|
|
assert_eq!(sse.data["title"], "我的新会话标题");
|
|
}
|
|
|
|
#[test]
|
|
fn acp_plan_update_emits_plan_updated_sse() {
|
|
let entries = vec![
|
|
"步骤 1：读取文件".into(),
|
|
"步骤 2：修改配置".into(),
|
|
"步骤 3：验证更改".into(),
|
|
];
|
|
let sse = acp_event_to_sse(AcpSessionEvent::PlanUpdate {
|
|
entries: entries.clone(),
|
|
})
|
|
.expect("plan update should be forwarded");
|
|
assert_eq!(sse.event, "plan.updated");
|
|
let sse_entries: Vec<String> =
|
|
serde_json::from_value(sse.data["entries"].clone()).unwrap_or_default();
|
|
assert_eq!(sse_entries.len(), 3);
|
|
assert_eq!(sse_entries[0], "步骤 1：读取文件");
|
|
assert_eq!(sse_entries[1], "步骤 2：修改配置");
|
|
assert_eq!(sse_entries[2], "步骤 3：验证更改");
|
|
}
|
|
}
|