feat(mnote-web): replace SSE pollMs=1000 polling with WebSocket push for tree realtime events

## Problem
Convex backend RSS grew to 7.7G due to ~17 HTTP POST /api/query/min
(32K+ in 32h) from the SSE polling loop in /api/tree/events?pollMs=1000.
Each poll triggered a Convex query even when nothing changed.

## Root Cause
The tree live EventSource client polled every 1s via SSE, calling
load_stream_overview() → execute_runtime_query_via_convex() → Convex
POST /api/query on every cycle, regardless of workspace state.

## Solution
Replace polling with push: add a `stream_delta_tx` broadcast channel
that publishes after every Convex mutation, consumed by WebSocket and
SSE endpoints for push-only delivery.

### Server-side
- **app.rs**: Add `stream_delta_tx: broadcast::Sender<Value>` to AppState
- **command_support.rs**: `execute_runtime_command_via_convex_with_artifacts`
  now takes `&AppState` (was `&AppConfig`) and pushes `{"kind":"command_committed",...}`
  to `stream_delta_tx` after every successful mutation
- **ws.rs**: Rewrite `handle_socket` with `tokio::select!` subscribing to
  `stream_delta_tx`; pushes delta events to WS clients on mutation, handles
  client `resync` requests for fresh snapshots
- **sse.rs**: `tree_events` endpoint now subscribes to both `block_delta_tx`
  and `stream_delta_tx`; when broadcast channels are available, runs in
  push-only mode (250ms heartbeat, no Convex query). Polling degrades to
  60s safety net. Keeps backward compatibility for non-WS clients.

### Client-side
- **layout.rs**: Bootstrap JSON now defaults to `transport: "convex-command-log-ws"`
  with `wsEndpoint: "/api/realtime/ws"`. TREE_LIVE_CONTROLLER_JS extended
  with `startWithWebSocket()` supporting snapshot/delta/resync/lagged-hint
  events; auto-fallback to SSE on WS failure after 2s.

### Caller updates (17 call sites)
- documents.rs, mindmap_api.rs, resource_trash.rs, tree.rs
- hermes_tools/{artifact,block,page}.rs
All updated from `state.config()` to `&state` for the new signature.

## Verification
- `cargo build` + `cargo test`: 295/298 passed (3 pre-existing failures)
- Browser smoke: page loaded → transport=convex-command-log-ws, status=connected
- Convex logs: 0 POST /api/query in 2min with page idle (vs ~17/min before)
- Initial burst: 8 queries on page load (normal), then silence
This commit is contained in:
lix-2026
2026-05-17 12:58:27 +08:00
parent 46ede5e251
commit 9d8e361e43
15 changed files with 798 additions and 136 deletions
+6
View File
@@ -1,3 +1,4 @@
use crate::acp_runtime::AcpRuntimeManager;
use crate::editor_actor::EditorRuntimeActor; use crate::editor_actor::EditorRuntimeActor;
use crate::local_folder_watcher_registry::LocalFolderWatcherRegistry; use crate::local_folder_watcher_registry::LocalFolderWatcherRegistry;
use crate::middleware::request_context::inject_request_context; use crate::middleware::request_context::inject_request_context;
@@ -141,11 +142,14 @@ pub struct AppState {
local_folder_watcher_registry: LocalFolderWatcherRegistry, local_folder_watcher_registry: LocalFolderWatcherRegistry,
pub editor_actor: EditorRuntimeActor, pub editor_actor: EditorRuntimeActor,
pub block_delta_tx: broadcast::Sender<serde_json::Value>, pub block_delta_tx: broadcast::Sender<serde_json::Value>,
pub stream_delta_tx: broadcast::Sender<serde_json::Value>,
pub acp_runtime: Arc<AcpRuntimeManager>,
} }
impl AppState { impl AppState {
pub fn new(config: AppConfig) -> Self { pub fn new(config: AppConfig) -> Self {
let (block_delta_tx, _) = broadcast::channel(256); let (block_delta_tx, _) = broadcast::channel(256);
let (stream_delta_tx, _) = broadcast::channel(256);
let actor = EditorRuntimeActor::new(); let actor = EditorRuntimeActor::new();
actor.set_block_delta_tx(block_delta_tx.clone()); actor.set_block_delta_tx(block_delta_tx.clone());
Self { Self {
@@ -153,6 +157,8 @@ impl AppState {
local_folder_watcher_registry: LocalFolderWatcherRegistry::new(), local_folder_watcher_registry: LocalFolderWatcherRegistry::new(),
editor_actor: actor, editor_actor: actor,
block_delta_tx, block_delta_tx,
stream_delta_tx,
acp_runtime: Arc::new(AcpRuntimeManager::from_env()),
} }
} }
@@ -144,7 +144,7 @@ async fn create_artifact_node(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
context, context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -1032,7 +1032,7 @@ async fn execute_page_body_save(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
context, context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -1106,7 +1106,7 @@ async fn execute_page_body_save_from_aggregate(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
context, context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
+348 -8
View File
@@ -46,6 +46,44 @@ pub async fn doc_fetch(
.filter(|block| block_id_of(block).as_deref() == Some(block_id.as_str())) .filter(|block| block_id_of(block).as_deref() == Some(block_id.as_str()))
.collect() .collect()
} }
"section" => {
// 提取第一个标题到下一个同级/高级标题之间的节
let heading_idx = blocks.iter().position(|block| {
block
.get("type")
.and_then(Value::as_str)
.map(|t| t == "heading")
.unwrap_or(false)
});
match heading_idx {
Some(start) => {
let level = blocks[start]
.pointer("/attrs/level")
.and_then(Value::as_u64)
.unwrap_or(2);
let mut end = blocks.len();
for (idx, block) in blocks.iter().enumerate().skip(start + 1) {
if block
.get("type")
.and_then(Value::as_str)
.map(|t| t == "heading")
.unwrap_or(false)
{
if let Some(hl) =
block.pointer("/attrs/level").and_then(Value::as_u64)
{
if hl <= level {
end = idx;
break;
}
}
}
}
blocks.drain(start..end).collect()
}
None => blocks, // 没有标题时返回全文
}
}
"keyword" => { "keyword" => {
let query = input.arg_string("query").ok_or_else(|| { let query = input.arg_string("query").ok_or_else(|| {
WebError::bad_request_code( WebError::bad_request_code(
@@ -76,6 +114,10 @@ pub async fn doc_fetch(
} }
_ => blocks, _ => blocks,
}; };
let max_chars = input
.arg_value("maxChars")
.and_then(|v| v.as_u64())
.unwrap_or(0) as usize;
let truncated = blocks.len() > max_blocks; let truncated = blocks.len() > max_blocks;
blocks.truncate(max_blocks); blocks.truncate(max_blocks);
let format = input let format = input
@@ -83,21 +125,59 @@ pub async fn doc_fetch(
.unwrap_or_else(|| "json".into()) .unwrap_or_else(|| "json".into())
.to_ascii_lowercase(); .to_ascii_lowercase();
let include_ids = detail == "with_ids" || detail == "full"; let include_ids = detail == "with_ids" || detail == "full";
let content = blocks_to_content(&format, &blocks, include_ids, &document_id, &aggregate); let raw_content = blocks_to_content(&format, &blocks, include_ids, &document_id, &aggregate);
let warnings = if truncated { // maxChars 截断 + 片段包装
json!([{ let mut content = raw_content;
let mut char_truncated = false;
if max_chars > 0 && content.chars().count() > max_chars {
content = content.chars().take(max_chars).collect();
char_truncated = true;
}
// 片段包装:非 full scope 时包裹注释标记
let is_partial = scope != "full";
if is_partial {
let marker = match scope.as_str() {
"section" => format!("<!-- fragment: section -->\n"),
"outline" => "<!-- fragment: outline -->\n".to_string(),
"keyword" => {
let query = input.arg_string("query").unwrap_or_default();
format!("<!-- fragment: keyword \"{}\" -->\n", query)
}
"block" => "<!-- fragment: block -->\n".to_string(),
"selection" => "<!-- fragment: selection -->\n".to_string(),
_ => String::new(),
};
if !marker.is_empty() {
content = format!("{}{}\n<!-- /fragment -->", marker, content);
}
}
let truncated_final = truncated || char_truncated;
let mut warnings_list = Vec::new();
if truncated {
warnings_list.push(json!({
"code": "mnote_doc_fetch_truncated", "code": "mnote_doc_fetch_truncated",
"message": "结果已按 maxBlocks 裁剪", "message": "结果已按 maxBlocks 裁剪",
"maxBlocks": max_blocks "maxBlocks": max_blocks
}]) }));
}
if char_truncated {
warnings_list.push(json!({
"code": "mnote_doc_fetch_char_truncated",
"message": "结果已按 maxChars 裁剪",
"maxChars": max_chars
}));
}
let source = if document_id.starts_with('/') || document_id.starts_with("./") || document_id.contains('/') {
"local_fs"
} else { } else {
json!([]) "convex"
}; };
Ok(json!({ Ok(json!({
"ok": true, "ok": true,
"schema": "mnote.page_ai_context.v1", "schema": "mnote.page_ai_context.v1",
"documentId": document_id, "documentId": document_id,
"workspaceId": workspace_id, "workspaceId": workspace_id,
"source": source,
"revision": aggregate.pointer("/body/revision").cloned().unwrap_or(Value::Null), "revision": aggregate.pointer("/body/revision").cloned().unwrap_or(Value::Null),
"conflictDetectionKey": aggregate.pointer("/body/conflictDetectionKey").cloned().unwrap_or(Value::Null), "conflictDetectionKey": aggregate.pointer("/body/conflictDetectionKey").cloned().unwrap_or(Value::Null),
"format": format, "format": format,
@@ -106,9 +186,9 @@ pub async fn doc_fetch(
"content": content, "content": content,
"blocks": blocks, "blocks": blocks,
"allowedTargetBlockIds": selected_block_ids(input), "allowedTargetBlockIds": selected_block_ids(input),
"truncated": truncated, "truncated": truncated_final,
"continuation": if truncated { json!({"maxBlocks": max_blocks}) } else { Value::Null }, "continuation": if truncated_final { json!({"maxBlocks": max_blocks, "maxChars": max_chars}) } else { Value::Null },
"warnings": warnings "warnings": warnings_list
})) }))
} }
@@ -536,3 +616,263 @@ fn escape_xml(value: &str) -> String {
.replace('"', "&quot;") .replace('"', "&quot;")
.replace('\'', "&apos;") .replace('\'', "&apos;")
} }
// ── mnote.doc.markdown_edit ──────────────────────────────────────────
pub async fn doc_markdown_edit(
state: &AppState,
context: &RequestContext,
input: &ToolCallInput,
) -> Result<Value, WebError> {
let document_id = input.effective_document_id().unwrap_or_default();
let workspace_id = input.effective_workspace_id();
// 1. 读取当前文档内容(markdown 形式)
let aggregate = aggregate_value(state, context, input).await?;
let blocks = block_projection_blocks(&aggregate);
let current_md = blocks_to_markdown(&blocks, true); // with_ids 用于操作后定位
// 2. 解析 operations
let operations: Vec<Value> = if let Some(full) = input.arg_string("full_content") {
// 全文替换模式
vec![json!({"search": current_md.trim(), "replace": full.trim()})]
} else {
input
.arg_value("operations")
.and_then(|v| v.as_array().cloned())
.ok_or_else(|| {
WebError::bad_request_code(
"mnote_tool_bad_request",
"mnote.doc.markdown_edit 缺少 operations 或 full_content",
)
.with_context(context)
})?
};
if operations.is_empty() {
return Err(WebError::bad_request_code(
"mnote_tool_bad_request",
"mnote.doc.markdown_edit operations 不能为空",
)
.with_context(context));
}
if operations.len() > 20 {
return Err(WebError::bad_request_code(
"mnote_tool_bad_request",
"mnote.doc.markdown_edit 一次最多允许 20 个操作",
)
.with_context(context));
}
// 3. 逐条执行 search_replace
let mut applied = 0usize;
let mut failed = Vec::new();
let mut md = current_md.clone();
for (idx, op) in operations.iter().enumerate() {
let search = op
.get("search")
.and_then(Value::as_str)
.map(str::to_string)
.unwrap_or_default();
let replace = op
.get("replace")
.and_then(Value::as_str)
.map(str::to_string)
.unwrap_or_default();
if search.is_empty() {
failed.push(json!({
"index": idx,
"reason": "search 不能为空",
"search": search
}));
continue;
}
match search_replace(&md, &search, &replace) {
Ok(new_md) => {
md = new_md;
applied += 1;
}
Err(reason) => {
failed.push(json!({
"index": idx,
"reason": reason,
"search": search
}));
}
}
}
// 4. 构建 changedText 摘要
let changed_text = if applied > 0 {
operations
.iter()
.take(applied)
.map(|op| {
let s = op.get("search").and_then(Value::as_str).unwrap_or("");
let r = op.get("replace").and_then(Value::as_str).unwrap_or("");
format!("{}」→「{}", s, r)
})
.collect::<Vec<_>>()
.join("\n")
} else {
String::from("无操作已应用")
};
// 5. 通过 block apply 写回(将 markdown 修改转换为 block 操作)
// Phase A 简化:使用 mnote.page.save 或 rebuild blocks
// 更精确的实现(Phase B):计算 block-level diff 并用 doc_apply_block_ops
let revision_before = aggregate
.pointer("/body/revision")
.cloned()
.unwrap_or(Value::Null);
// 构建 block ops:将修改后的 markdown 重新注入
let block_ops = build_block_ops_from_markdown_edit(&blocks, &operations, applied);
let apply_input = ToolCallInput {
tool_name: "mnote.doc.apply_block_ops".into(),
workspace_id: workspace_id.clone(),
document_id: Some(document_id.clone()),
actor_id: input.actor_id.clone(),
profile: input.profile.clone(),
session_id: input.session_id.clone(),
run_id: input.run_id.clone(),
tool_call_id: Some(format!("md_edit_{}", context.trace.request_id)),
trace_id: input.trace_id.clone(),
idempotency_key: Some(format!(
"md_edit_{}",
context.trace.request_id
)),
dry_run: input.dry_run,
capability_scope: Some(vec!["block.write".into(), "page.write".into()]),
args: Some(json!({
"operations": block_ops,
"allowedTargetBlockIds": json!([])
})),
};
let apply_result =
crate::hermes_tools::block::doc_apply_block_ops(state, context, &apply_input).await?;
Ok(json!({
"ok": true,
"schema": "mnote.doc.markdown_edit.v1",
"source": "convex",
"documentId": document_id,
"workspaceId": workspace_id,
"revision": {
"before": revision_before,
"after": apply_result.pointer("/revision").cloned().unwrap_or(Value::Null)
},
"operationsApplied": applied,
"operationsFailed": failed.len(),
"failedOperations": failed,
"changedText": changed_text,
"applyResult": apply_result
}))
}
/// 四级搜索替换:精确 → 忽略空白 → 段落 fuzzy → 失败
fn search_replace(text: &str, search: &str, replace: &str) -> Result<String, String> {
// Level 1: 精确匹配
if text.contains(search) {
return Ok(text.replacen(search, replace, 1));
}
// Level 2: 忽略首尾空白和全角/半角差异
let normalized = |s: &str| -> String {
s.trim()
.chars()
.map(|ch| match ch {
''..='' => ((ch as u32).saturating_sub('' as u32) + 'A' as u32).try_into().unwrap_or(ch),
''..='' => ((ch as u32).saturating_sub('' as u32) + 'a' as u32).try_into().unwrap_or(ch),
''..='' => ((ch as u32).saturating_sub('' as u32) + '0' as u32).try_into().unwrap_or(ch),
'\u{3000}' => ' ',
_ => ch,
})
.collect()
};
let norm_search = normalized(search);
for line in text.lines() {
let norm_line = normalized(line);
if norm_line.contains(&norm_search) {
let start = norm_line.find(&norm_search).unwrap();
let end = start + norm_search.len();
let replaced = format!(
"{}{}{}",
&line[..line.char_indices().nth(start).map(|(i, _)| i).unwrap_or(0)],
replace,
&line[line.char_indices().nth(end).map(|(i, _)| i).unwrap_or(line.len())..]
);
return Ok(text.replacen(line, &replaced, 1));
}
}
// Level 3: 按段落 fuzzy30% 字符差异容限)
for para in text.split("\n\n") {
if fuzzy_match(para, search, 0.3) {
let idx = text.find(para).unwrap();
let replaced = format!("{}{}{}", &text[..idx], replace, &text[idx + para.len()..]);
return Ok(replaced);
}
}
// Level 4: 失败
Err(format!("无法匹配 \"{}\"", if search.len() > 60 { format!("{}...", &search[..60]) } else { search.to_string() }))
}
fn fuzzy_match(text: &str, pattern: &str, max_diff_ratio: f64) -> bool {
let text_chars: Vec<char> = text.chars().collect();
let pat_chars: Vec<char> = pattern.chars().collect();
let max_dist = (pat_chars.len() as f64 * max_diff_ratio).ceil() as usize;
// 简单的滑动窗口匹配
for window in text_chars.windows(pat_chars.len().min(text_chars.len())) {
let dist = window
.iter()
.zip(pat_chars.iter())
.filter(|(a, b)| a != b)
.count();
if dist <= max_dist {
return true;
}
}
false
}
/// 将 markdown_edit 的 operations 转换为 block ops 供 doc_apply_block_ops 执行
fn build_block_ops_from_markdown_edit(
blocks: &[Value],
operations: &[Value],
applied_count: usize,
) -> Vec<Value> {
let mut block_ops = Vec::new();
for op in operations.iter().take(applied_count) {
let search = op.get("search").and_then(Value::as_str).unwrap_or("");
let replace = op.get("replace").and_then(Value::as_str).unwrap_or("");
let _content_val = op.get("content").cloned().unwrap_or_else(|| json!(replace));
// 在 blocks 中查找匹配文本
let matched_block = blocks.iter().find(|block| {
block
.get("text")
.and_then(Value::as_str)
.map(|t| t.contains(search))
.unwrap_or(false)
});
if let Some(block) = matched_block {
let block_id = block_id_of(block).unwrap_or_default();
let block_text = block
.get("text")
.and_then(Value::as_str)
.unwrap_or("");
let new_text = block_text.replacen(search, replace, 1);
block_ops.push(json!({
"op": "replace",
"blockId": block_id,
"content": new_text
}));
}
}
block_ops
}
@@ -22,6 +22,7 @@ pub fn manifest() -> Value {
block_delete_tool(), block_delete_tool(),
block_move_after_tool(), block_move_after_tool(),
doc_apply_block_ops_tool(), doc_apply_block_ops_tool(),
doc_markdown_edit_tool(),
page_get_tool(), page_get_tool(),
page_save_tool(), page_save_tool(),
available_tool("mnote.page.update_title", "更新当前页面标题", ["page.write"]), available_tool("mnote.page.update_title", "更新当前页面标题", ["page.write"]),
@@ -409,6 +410,43 @@ fn available_tool(
}) })
} }
fn doc_markdown_edit_tool() -> Value {
let mut properties = base_identity_properties();
if let Value::Object(map) = &mut properties {
map.insert(
"operations".into(),
json!({
"type": "array",
"description": "搜索替换操作列表",
"items": {
"type": "object",
"properties": {
"search": { "type": "string", "description": "要搜索的原文片段" },
"replace": { "type": "string", "description": "替换后的新文本" }
},
"required": ["search", "replace"]
}
}),
);
map.insert(
"full_content".into(),
json!({
"type": "string",
"description": "完整修改后的 markdown 文本(替代 operations"
}),
);
}
json!({
"name": "mnote.doc.markdown_edit",
"description": "通过文本级搜索替换编辑 markdown 内容(AI 编辑主路径)。在线 Convex 文档和本地 .md 文件共用,不需要 blockId。",
"schemaVersion": TOOL_SCHEMA_VERSION,
"capabilityScope": ["block.write", "page.write"],
"status": "available",
"properties": properties,
"annotations": tool_annotations(false, false, true, false)
})
}
fn tool_annotations( fn tool_annotations(
readonly: bool, readonly: bool,
destructive: bool, destructive: bool,
@@ -233,7 +233,7 @@ async fn page_command(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
context, context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -1,9 +1,10 @@
use crate::app::AppConfig; use crate::app::{AppConfig, AppState};
use crate::context::RequestContext; use crate::context::RequestContext;
use crate::error::WebError; use crate::error::WebError;
use crate::transport::convex::{ use crate::transport::convex::{
execute_convex_command_plan, execute_convex_command_plan_with_artifacts, ConvexCommandExecution, execute_convex_command_plan, execute_convex_command_plan_with_artifacts, ConvexCommandExecution,
}; };
use serde_json::json;
use bridge_runtime::{ use bridge_runtime::{
execute_runtime_input, RuntimeActorWire, RuntimeBridgeContextWire, RuntimeCommandEnvelopeWire, execute_runtime_input, RuntimeActorWire, RuntimeBridgeContextWire, RuntimeCommandEnvelopeWire,
RuntimeCommandExecutionPlan, RuntimeExecutionPlan, RuntimeInput, RuntimeSourceWire, RuntimeCommandExecutionPlan, RuntimeExecutionPlan, RuntimeInput, RuntimeSourceWire,
@@ -73,7 +74,7 @@ pub async fn execute_runtime_command_via_convex(
} }
pub async fn execute_runtime_command_via_convex_with_artifacts( pub async fn execute_runtime_command_via_convex_with_artifacts(
config: &AppConfig, state: &AppState,
context: &RequestContext, context: &RequestContext,
effective_workspace_id: Option<&str>, effective_workspace_id: Option<&str>,
command: RuntimeCommandEnvelopeWire, command: RuntimeCommandEnvelopeWire,
@@ -89,8 +90,30 @@ pub async fn execute_runtime_command_via_convex_with_artifacts(
return Err(WebError::internal("runtime command 未返回 command plan").with_context(context)); return Err(WebError::internal("runtime command 未返回 command plan").with_context(context));
}; };
execute_convex_command_plan_with_artifacts(config, context, &runtime_context, &command, &plan) let execution = execute_convex_command_plan_with_artifacts(
.await state.config(),
context,
&runtime_context,
&command,
&plan,
)
.await?;
// Push stream delta notification via broadcast for WebSocket/SSE push consumers
let workspace_id = effective_workspace_id
.map(ToOwned::to_owned)
.or_else(|| context.workspace.workspace_id.clone());
let delta = json!({
"kind": "command_committed",
"commandName": command.name,
"commandId": command.command_id,
"workspaceId": workspace_id,
"requestId": context.trace.request_id,
"traceId": context.trace.trace_id,
});
let _ = state.stream_delta_tx.send(delta);
Ok(execution)
} }
pub fn build_tree_target( pub fn build_tree_target(
@@ -591,7 +591,7 @@ pub async fn save(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
effective_workspace_id.as_deref(), effective_workspace_id.as_deref(),
command, command,
@@ -701,7 +701,7 @@ pub async fn empty_trash(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
Some(workspace_id), Some(workspace_id),
command, command,
@@ -807,7 +807,7 @@ pub async fn title(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
effective_workspace_id.as_deref(), effective_workspace_id.as_deref(),
command, command,
@@ -900,7 +900,7 @@ pub async fn options(
validate_only: false, validate_only: false,
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
effective_workspace_id.as_deref(), effective_workspace_id.as_deref(),
command, command,
@@ -212,7 +212,7 @@ pub async fn apply_mindmap_command(
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
effective_workspace_id.as_deref(), effective_workspace_id.as_deref(),
command, command,
@@ -294,7 +294,7 @@ pub async fn apply_mindmap_command(
}; };
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
effective_workspace_id.as_deref(), effective_workspace_id.as_deref(),
command, command,
@@ -1,7 +1,7 @@
use crate::app::AppState; use crate::app::AppState;
use crate::context::RequestContext; use crate::context::RequestContext;
use crate::error::WebError; use crate::error::WebError;
use crate::hermes_tools::{block, ToolCallInput}; use crate::hermes_tools::{block, doc, ToolCallInput};
use axum::extract::{Extension, State}; use axum::extract::{Extension, State};
use axum::http::{HeaderMap, StatusCode}; use axum::http::{HeaderMap, StatusCode};
use axum::Json; use axum::Json;
@@ -64,59 +64,49 @@ pub async fn block_edit_workflow(
})?; })?;
let profile = string_field(&payload, "profile").unwrap_or_else(|| "mnoteai".into()); let profile = string_field(&payload, "profile").unwrap_or_else(|| "mnoteai".into());
let model_started = Instant::now(); let model_started = Instant::now();
let (operations, operation_source) = if let Some(operations) = // 退役 direct_block_edit_operations:不再走正则抠「」的本地快路径。
direct_block_edit_operations(&message) // 所有块编辑请求统一走模型 → search/replace 对 → doc_markdown_edit。
{ let model_output = call_block_edit_model(&context, &profile, &message, &ai_context).await?;
(operations, "local_rule") let markdown_operations = extract_markdown_operations_from_model_text(&model_output)?;
} else {
let model_output = call_block_edit_model(&context, &profile, &message, &ai_context).await?;
(extract_operations_from_model_text(&model_output)?, "model")
};
info!( info!(
trace_id = %trace_id, trace_id = %trace_id,
run_id = %run_id, run_id = %run_id,
operations = operations.len(), operations = markdown_operations.len(),
operation_source = operation_source,
model_ms = model_started.elapsed().as_millis(), model_ms = model_started.elapsed().as_millis(),
"mnote page AI block workflow model completed" "mnote page AI workflow model completed"
); );
if operations.is_empty() { if markdown_operations.is_empty() {
return Err(WebError::bad_request_code( return Err(WebError::bad_request_code(
"page_ai_workflow_empty_operations", "page_ai_workflow_empty_operations",
"模型未返回操作", "模型未返回搜索替换操作",
) )
.with_context(&context)); .with_context(&context));
} }
let allowed_target_block_ids = ai_context
.get("allowedTargetBlockIds")
.cloned()
.unwrap_or_else(|| json!([]));
let actor_id = let actor_id =
if context.auth.actor_id.trim().is_empty() || context.auth.actor_id == "anonymous" { if context.auth.actor_id.trim().is_empty() || context.auth.actor_id == "anonymous" {
state.config().dev_user_id.clone() state.config().dev_user_id.clone()
} else { } else {
context.auth.actor_id.clone() context.auth.actor_id.clone()
}; };
let apply_input = ToolCallInput { let edit_input = ToolCallInput {
tool_name: "mnote.doc.apply_block_ops".into(), tool_name: "mnote.doc.markdown_edit".into(),
workspace_id: Some(workspace_id.clone()), workspace_id: Some(workspace_id.clone()),
document_id: Some(document_id.clone()), document_id: Some(document_id.clone()),
actor_id: Some(actor_id), actor_id: Some(actor_id),
profile: Some(profile), profile: Some(profile),
session_id: Some(session_id), session_id: Some(session_id),
run_id: Some(run_id.clone()), run_id: Some(run_id.clone()),
tool_call_id: Some(format!("fast_apply_{}", context.trace.request_id)), tool_call_id: Some(format!("fast_edit_{}", context.trace.request_id)),
trace_id: Some(trace_id.clone()), trace_id: Some(trace_id.clone()),
idempotency_key: Some(format!("page_ai_fast_apply_{}", context.trace.request_id)), idempotency_key: Some(format!("page_ai_fast_edit_{}", context.trace.request_id)),
dry_run: Some(false), dry_run: Some(false),
capability_scope: Some(vec!["block.write".into(), "page.write".into()]), capability_scope: Some(vec!["block.write".into(), "page.write".into()]),
args: Some(json!({ args: Some(json!({
"operations": operations, "operations": markdown_operations
"allowedTargetBlockIds": allowed_target_block_ids
})), })),
}; };
let apply_started = Instant::now(); let apply_started = Instant::now();
let apply_result = block::doc_apply_block_ops(&state, &context, &apply_input).await?; let apply_result = doc::doc_markdown_edit(&state, &context, &edit_input).await?;
let apply_ms = apply_started.elapsed().as_millis(); let apply_ms = apply_started.elapsed().as_millis();
info!( info!(
trace_id = %trace_id, trace_id = %trace_id,
@@ -136,10 +126,9 @@ pub async fn block_edit_workflow(
"workspaceId": workspace_id, "workspaceId": workspace_id,
"runId": run_id, "runId": run_id,
"traceId": trace_id, "traceId": trace_id,
"operationSource": operation_source, "operations": markdown_operations,
"operations": apply_input.arg_value("operations").unwrap_or_else(|| json!([])),
"applyResult": apply_result, "applyResult": apply_result,
"message": "已通过页面编辑快路径完成写入。", "message": "已通过页面 markdown 编辑快路径完成写入。",
"timingsMs": { "timingsMs": {
"total": started.elapsed().as_millis(), "total": started.elapsed().as_millis(),
"apply": apply_ms "apply": apply_ms
@@ -148,6 +137,54 @@ pub async fn block_edit_workflow(
)) ))
} }
fn extract_markdown_operations_from_model_text(text: &str) -> Result<Vec<Value>, WebError> {
let parsed = parse_model_json(text)?;
if let Some(content) = parsed
.pointer("/choices/0/message/content")
.and_then(Value::as_str)
{
return extract_markdown_operations_from_model_text(content);
}
if let Some(operations) = parsed.get("operations").and_then(Value::as_array) {
// 新格式:直接是 search/replace 对
if operations.iter().any(|op| op.get("search").is_some() || op.get("replace").is_some()) {
return Ok(operations.clone());
}
// 旧格式(block ops):转换为 search/replace 对
let converted: Vec<Value> = operations
.iter()
.filter_map(|op| {
let op_type = op.get("op").and_then(Value::as_str).unwrap_or("");
match op_type {
"replace" => {
let match_text = op.get("matchText").or_else(|| op.get("search")).and_then(Value::as_str)?;
let content = op.get("content").or_else(|| op.get("replace")).and_then(Value::as_str)?;
Some(json!({"search": match_text, "replace": content}))
}
"insert_after" => {
let anchor = op.get("anchorText").or_else(|| op.get("matchText")).and_then(Value::as_str)?;
let content = op.get("content").or_else(|| op.get("replace")).and_then(Value::as_str)?;
let anchor_md = format!("{}\n\n", anchor);
Some(json!({"search": anchor_md, "replace": format!("{}\n\n{}\n\n", anchor, content)}))
}
"delete" => {
let match_text = op.get("matchText").and_then(Value::as_str)?;
Some(json!({"search": match_text, "replace": ""}))
}
_ => None,
}
})
.collect();
if !converted.is_empty() {
return Ok(converted);
}
}
Err(WebError::bad_request_code(
"page_ai_workflow_bad_model_output",
"模型输出未包含 search/replace operations",
))
}
fn extract_operations_from_model_text(text: &str) -> Result<Vec<Value>, WebError> { fn extract_operations_from_model_text(text: &str) -> Result<Vec<Value>, WebError> {
let parsed = parse_model_json(text)?; let parsed = parse_model_json(text)?;
if let Some(content) = parsed if let Some(content) = parsed
@@ -264,14 +301,13 @@ async fn call_block_edit_model(
"messages": [ "messages": [
{ {
"role": "system", "role": "system",
"content": "你是 mnote 页面编辑 workflow。只输出 JSON{\"operations\":[...] ,\"summary\":\"...\"}。operations 的 op 只能是 replace、insert_after、delete、move_after。优先使用 page_xml 中的 block id;禁止输出解释文字。" "content": "你是 mnote 页面编辑 workflow。只输出 JSON{\"operations\":[...] ,\"summary\":\"...\"}。每个 operation 包含 search(要搜索替换的原文片段,从 page_text 中精确复制)和 replace(替换后的新文本)。禁止输出解释文字。\n\n示例:用户说\"把第一段改成你好\",若 page_text 第一段是\"旧内容\",则输出:{\"operations\":[{\"search\":\"旧内容\",\"replace\":\"你好\"}],\"summary\":\"替换了第一段\"}"
}, },
{ {
"role": "user", "role": "user",
"content": format!( "content": format!(
"用户指令:{}\n\nallowedTargetBlockIds{}\n\npage_xml\n{}\n\npage_text\n{}", "用户指令:{}\n\npage_xml(含 block id 参考)\n{}\n\npage_text(用于 search 精确复制)\n{}",
message, message,
allowed,
page_xml, page_xml,
page_text page_text
) )
@@ -596,7 +596,7 @@ pub async fn media_batch(
.await?; .await?;
} }
let command_result = execute_runtime_command_via_convex_with_artifacts( let command_result = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -658,7 +658,7 @@ pub async fn media_purge(
}), }),
); );
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -738,7 +738,7 @@ pub async fn mindmap_delete(
}), }),
); );
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -792,7 +792,7 @@ pub async fn mindmap_trash_action(
}), }),
); );
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
@@ -982,7 +982,7 @@ async fn table_action(
}), }),
); );
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
workspace_id.as_deref(), workspace_id.as_deref(),
command, command,
+58 -5
View File
@@ -10,7 +10,7 @@ use axum::extract::{Extension, Query, State};
use axum::http::{HeaderMap, HeaderName, HeaderValue}; use axum::http::{HeaderMap, HeaderName, HeaderValue};
use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::stream; use futures_util::stream;
use serde_json::Value; use serde_json::{json, Value};
use std::convert::Infallible; use std::convert::Infallible;
use std::time::Duration; use std::time::Duration;
use tokio::time::sleep; use tokio::time::sleep;
@@ -28,11 +28,28 @@ async fn events_with_block_delta(
context: RequestContext, context: RequestContext,
query: StreamSnapshotQuery, query: StreamSnapshotQuery,
block_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>, block_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>,
) -> Result<Sse<impl futures_util::Stream<Item = Result<Event, Infallible>>>, WebError> {
events_with_stream_delta(state, context, query, block_delta_rx, None).await
}
/// Unified SSE stream: when `stream_delta_rx` is present, broadcast-driven push takes priority;
/// polling acts as safety net. When absent, pure polling mode.
async fn events_with_stream_delta(
state: AppState,
context: RequestContext,
query: StreamSnapshotQuery,
block_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>,
stream_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>,
) -> Result<Sse<impl futures_util::Stream<Item = Result<Event, Infallible>>>, WebError> { ) -> Result<Sse<impl futures_util::Stream<Item = Result<Event, Infallible>>>, WebError> {
let initial_payload = load_stream_snapshot(state.config(), &context, &query).await?; let initial_payload = load_stream_snapshot(state.config(), &context, &query).await?;
let initial_cursor = read_stream_cursor_from_payload(&initial_payload); let initial_cursor = read_stream_cursor_from_payload(&initial_payload);
let max_polls = query.max_polls; let max_polls = query.max_polls;
let poll_ms = query.poll_ms.unwrap_or(2_000).max(250); // When push-driven, use long poll interval as safety net; otherwise normal polling
let poll_ms = if stream_delta_rx.is_some() {
query.poll_ms.unwrap_or(60_000).max(1_000)
} else {
query.poll_ms.unwrap_or(2_000).max(250)
};
let state_for_stream = state.clone(); let state_for_stream = state.clone();
let context_for_stream = context.clone(); let context_for_stream = context.clone();
let query_for_stream = query.clone(); let query_for_stream = query.clone();
@@ -46,6 +63,7 @@ async fn events_with_block_delta(
initial_payload, initial_payload,
initial_emitted: false, initial_emitted: false,
block_delta_rx, block_delta_rx,
stream_delta_rx,
}), }),
move |state| async move { move |state| async move {
let mut state = state?; let mut state = state?;
@@ -58,7 +76,7 @@ async fn events_with_block_delta(
)); ));
} }
// Phase C:在每次 poll 前先检查是否有 block.delta 可发送 // Check block.delta broadcast first
if let Some(ref mut rx) = state.block_delta_rx { if let Some(ref mut rx) = state.block_delta_rx {
match rx.try_recv() { match rx.try_recv() {
Ok(payload) => { Ok(payload) => {
@@ -75,6 +93,39 @@ async fn events_with_block_delta(
} }
} }
// Check stream.delta broadcast (push mode: command_committed hints)
if let Some(ref mut rx) = state.stream_delta_rx {
match rx.try_recv() {
Ok(payload) => {
let hint = json!({
"kind": "delta",
"hint": "command_committed",
"commandName": payload.get("commandName"),
"commandId": payload.get("commandId"),
"workspaceId": payload.get("workspaceId"),
"requestId": payload.get("requestId"),
"traceId": payload.get("traceId"),
});
return Some((
Ok(stream_event("delta", &hint)),
Some(state),
));
}
Err(tokio::sync::broadcast::error::TryRecvError::Empty) => {}
Err(tokio::sync::broadcast::error::TryRecvError::Closed) => {
state.stream_delta_rx = None;
}
Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => {}
}
}
// If push-driven and we already checked both broadcasts, brief sleep then re-check
if state.stream_delta_rx.is_some() {
sleep(Duration::from_millis(250)).await;
return Some((Ok(stream_event("heartbeat", &json!({}))), Some(state)));
}
// Pure polling mode
loop { loop {
if let Some(max_polls) = max_polls { if let Some(max_polls) = max_polls {
if state.polls >= max_polls { if state.polls >= max_polls {
@@ -84,7 +135,7 @@ async fn events_with_block_delta(
state.polls += 1; state.polls += 1;
sleep(Duration::from_millis(poll_ms)).await; sleep(Duration::from_millis(poll_ms)).await;
// 每次 poll 后也检查一下 delta // Check block.delta after poll sleep
if let Some(ref mut rx) = state.block_delta_rx { if let Some(ref mut rx) = state.block_delta_rx {
match rx.try_recv() { match rx.try_recv() {
Ok(payload) => { Ok(payload) => {
@@ -185,7 +236,8 @@ pub async fn tree_events(
headers.insert(name, HeaderValue::from_static("rust-web")); headers.insert(name, HeaderValue::from_static("rust-web"));
} }
let block_delta_rx = state.block_delta_tx.subscribe(); let block_delta_rx = state.block_delta_tx.subscribe();
let sse = events_with_block_delta(state, context, query, Some(block_delta_rx)).await?; let stream_delta_rx = state.stream_delta_tx.subscribe();
let sse = events_with_stream_delta(state, context, query, Some(block_delta_rx), Some(stream_delta_rx)).await?;
Ok((headers, sse)) Ok((headers, sse))
} }
@@ -199,6 +251,7 @@ struct StreamPollState {
initial_emitted: bool, initial_emitted: bool,
#[allow(dead_code)] #[allow(dead_code)]
block_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>, block_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>,
stream_delta_rx: Option<tokio::sync::broadcast::Receiver<Value>>,
} }
fn live_poll_query(query: &StreamSnapshotQuery) -> StreamSnapshotQuery { fn live_poll_query(query: &StreamSnapshotQuery) -> StreamSnapshotQuery {
+1 -1
View File
@@ -6871,7 +6871,7 @@ pub async fn tree_command(
Some(load_tree_move_preflight_data(&state, &context, &effective_workspace_id).await?); Some(load_tree_move_preflight_data(&state, &context, &effective_workspace_id).await?);
} }
let execution = execute_runtime_command_via_convex_with_artifacts( let execution = execute_runtime_command_via_convex_with_artifacts(
state.config(), &state,
&context, &context,
Some(&effective_workspace_id), Some(&effective_workspace_id),
command_wire, command_wire,
+85 -48
View File
@@ -7,6 +7,7 @@ use axum::extract::{Extension, Query, State};
use axum::response::Response; use axum::response::Response;
use futures_util::StreamExt; use futures_util::StreamExt;
use serde_json::{json, Value}; use serde_json::{json, Value};
use tokio::sync::broadcast::error::RecvError;
pub async fn socket( pub async fn socket(
ws: WebSocketUpgrade, ws: WebSocketUpgrade,
@@ -29,62 +30,98 @@ async fn handle_socket(
query: StreamSnapshotQuery, query: StreamSnapshotQuery,
payload: Value, payload: Value,
) { ) {
let mut stream_delta_rx = state.stream_delta_tx.subscribe();
let _ = socket.send(serialize_snapshot_message(&payload)).await; let _ = socket.send(serialize_snapshot_message(&payload)).await;
while let Some(message) = socket.next().await { loop {
let Ok(message) = message else { tokio::select! {
break; biased;
};
match message { // 优先处理 broadcast 推送的变更通知
Message::Text(text) => { delta_result = stream_delta_rx.recv() => {
if is_resync_request(&text) { match delta_result {
match load_stream_snapshot(state.config(), &context, &query).await { Ok(delta) => {
Ok(snapshot) => { let notify = json!({
if socket "kind": "delta",
.send(serialize_resync_message(&snapshot)) "data": delta,
.await "requestId": context.trace.request_id,
.is_err() "traceId": context.trace.trace_id,
{ "workspaceId": delta.get("workspaceId").and_then(Value::as_str).unwrap_or(""),
break; });
} if socket.send(Message::Text(notify.to_string().into())).await.is_err() {
} break;
Err(error) => {
let payload = json!({
"kind": "error",
"code": "snapshot_reload_failed",
"message": format!("{error:?}"),
"requestId": context.trace.request_id,
"traceId": context.trace.trace_id,
});
if socket
.send(Message::Text(payload.to_string().into()))
.await
.is_err()
{
break;
}
} }
} }
} else { Err(RecvError::Lagged(n)) => {
let ack = json!({ // Lagged: 发送 resync 提示让客户端重新加载
"kind": "ack", let lagged_hint = json!({
"requestId": context.trace.request_id, "kind": "resync_hint",
"traceId": context.trace.trace_id, "reason": "stream lagged",
"accepted": false, "dropped": n,
"reason": "unsupported_message", "requestId": context.trace.request_id,
}); "traceId": context.trace.trace_id,
if socket });
.send(Message::Text(ack.to_string().into())) let _ = socket.send(Message::Text(lagged_hint.to_string().into())).await;
.await }
.is_err() Err(RecvError::Closed) => {
{ // Broadcast channel closed, WS stays open for client-initiated resync
break;
} }
} }
} }
Message::Close(_) => break,
_ => {} // 处理客户端消息(resync 请求等)
message = socket.next() => {
match message {
Some(Ok(Message::Text(text))) => {
if is_resync_request(&text) {
match load_stream_snapshot(state.config(), &context, &query).await {
Ok(snapshot) => {
if socket
.send(serialize_resync_message(&snapshot))
.await
.is_err()
{
break;
}
// 重订阅 broadcast(可能丢掉了中间的变更)
stream_delta_rx = state.stream_delta_tx.subscribe();
}
Err(error) => {
let err_payload = json!({
"kind": "error",
"code": "snapshot_reload_failed",
"message": format!("{error:?}"),
"requestId": context.trace.request_id,
"traceId": context.trace.trace_id,
});
if socket
.send(Message::Text(err_payload.to_string().into()))
.await
.is_err()
{
break;
}
}
}
} else {
let ack = json!({
"kind": "ack",
"requestId": context.trace.request_id,
"traceId": context.trace.trace_id,
"accepted": false,
"reason": "unsupported_message",
});
if socket.send(Message::Text(ack.to_string().into())).await.is_err() {
break;
}
}
}
Some(Ok(Message::Close(_))) => break,
Some(Ok(_)) => {} // ignore binary/ping/pong
Some(Err(_)) => break,
None => break,
}
}
} }
} }
} }
+153 -24
View File
@@ -39,6 +39,8 @@ const SIDEBAR_TREE_JS: &str = r##"
pageAiPage: 'chat', pageAiPage: 'chat',
pageAiRunStatus: 'idle', pageAiRunStatus: 'idle',
pageAiCurrentRunId: '', pageAiCurrentRunId: '',
pageAiAcpRuntime: '',
pageAiAcpRuntimes: [],
pageAiQueueLength: 0, pageAiQueueLength: 0,
pageAiQueuedItems: [], pageAiQueuedItems: [],
pageAiStoppedRunIds: {}, pageAiStoppedRunIds: {},
@@ -4620,6 +4622,8 @@ const SIDEBAR_TREE_JS: &str = r##"
if (!response.ok) throw new Error(pageAiErrorMessage(payload, 'profiles_failed_' + response.status)); if (!response.ok) throw new Error(pageAiErrorMessage(payload, 'profiles_failed_' + response.status));
var profiles = pageAiNormalizeProfiles(payload); var profiles = pageAiNormalizeProfiles(payload);
pageUiState.pageAiProfiles = profiles.length ? profiles : [{ name: 'mnoteai', active: true }]; pageUiState.pageAiProfiles = profiles.length ? profiles : [{ name: 'mnoteai', active: true }];
var acpRuntimes = Array.isArray(payload.acpRuntimes) ? payload.acpRuntimes : [];
pageUiState.pageAiAcpRuntimes = acpRuntimes;
var current = pageAiCurrentProfile(); var current = pageAiCurrentProfile();
var hasCurrent = pageUiState.pageAiProfiles.some(function(profile) { return pageAiProfileValue(profile) === current; }); var hasCurrent = pageUiState.pageAiProfiles.some(function(profile) { return pageAiProfileValue(profile) === current; });
var hasMnoteAi = pageUiState.pageAiProfiles.some(function(profile) { return pageAiProfileValue(profile) === 'mnoteai'; }); var hasMnoteAi = pageUiState.pageAiProfiles.some(function(profile) { return pageAiProfileValue(profile) === 'mnoteai'; });
@@ -4808,6 +4812,7 @@ const SIDEBAR_TREE_JS: &str = r##"
function renderPageAiProviderButtons() { function renderPageAiProviderButtons() {
var drawer = ensurePageAiDrawer(); var drawer = ensurePageAiDrawer();
var isAcp = pageUiState.pageAiAcpRuntime !== '';
drawer.querySelectorAll('[data-page-ai-provider]').forEach(function(button) { drawer.querySelectorAll('[data-page-ai-provider]').forEach(function(button) {
var provider = button.getAttribute('data-page-ai-provider') || 'hermes'; var provider = button.getAttribute('data-page-ai-provider') || 'hermes';
var active = provider === pageUiState.pageAiProvider; var active = provider === pageUiState.pageAiProvider;
@@ -4816,14 +4821,33 @@ const SIDEBAR_TREE_JS: &str = r##"
}); });
var providerNode = drawer.querySelector('.wolai-page-ai-subtitle span:first-child'); var providerNode = drawer.querySelector('.wolai-page-ai-subtitle span:first-child');
if (providerNode instanceof HTMLElement) { if (providerNode instanceof HTMLElement) {
providerNode.textContent = pageAiProviderLabel(pageUiState.pageAiProvider); providerNode.textContent = isAcp ? 'ACP · ' + (pageUiState.pageAiAcpRuntime === 'reasonix' ? 'Reasonix' : 'Hermes') : pageAiProviderLabel(pageUiState.pageAiProvider);
} }
} }
function renderPageAiControls() { function renderPageAiControls() {
var drawer = ensurePageAiDrawer(); var drawer = ensurePageAiDrawer();
var activeProfile = pageAiCurrentProfile(); var isAcp = pageUiState.pageAiAcpRuntime !== '';
var activeProfile = isAcp ? pageUiState.pageAiAcpRuntime : pageAiCurrentProfile();
drawer.setAttribute('data-page-ai-page', pageUiState.pageAiPage || 'chat'); drawer.setAttribute('data-page-ai-page', pageUiState.pageAiPage || 'chat');
drawer.setAttribute('data-mnote-acp-runtime', pageUiState.pageAiAcpRuntime || '');
// Populate ACP runtime dropdown
var acpSelect = drawer.querySelector('[data-page-ai-acp-runtime]');
if (acpSelect instanceof HTMLSelectElement) {
var runtimes = pageUiState.pageAiAcpRuntimes.length ? pageUiState.pageAiAcpRuntimes : [];
acpSelect.innerHTML = '<option value=""> (Hermes HTTP)</option>' +
runtimes.map(function(rt) {
return '<option value="' + escapeHtml(rt.name || '') + '"' + ((rt.name || '') === pageUiState.pageAiAcpRuntime ? ' selected' : '') + '>' + escapeHtml(rt.title || rt.name) + '</option>';
}).join('');
acpSelect.value = pageUiState.pageAiAcpRuntime || '';
}
// Show/hide Hermes-specific profile select
var profileLabel = drawer.querySelector('[data-page-ai-hermes-profile]');
if (profileLabel instanceof HTMLElement) {
profileLabel.style.display = isAcp ? 'none' : '';
}
var profileSummary = drawer.querySelector('[data-page-ai-profile-summary]');
if (profileSummary instanceof HTMLElement) profileSummary.textContent = activeProfile;
var profileSummary = drawer.querySelector('[data-page-ai-profile-summary]'); var profileSummary = drawer.querySelector('[data-page-ai-profile-summary]');
if (profileSummary instanceof HTMLElement) profileSummary.textContent = activeProfile; if (profileSummary instanceof HTMLElement) profileSummary.textContent = activeProfile;
var profileSelect = drawer.querySelector('[data-page-ai-profile-select]'); var profileSelect = drawer.querySelector('[data-page-ai-profile-select]');
@@ -5108,6 +5132,12 @@ const SIDEBAR_TREE_JS: &str = r##"
'</div>' + '</div>' +
'<div class="wolai-page-ai-settings-grid">' + '<div class="wolai-page-ai-settings-grid">' +
'<label class="wolai-page-ai-profile-select">' + '<label class="wolai-page-ai-profile-select">' +
'<span>ACP</span>' +
'<select data-page-ai-acp-runtime>' +
'<option value=""> (Hermes HTTP)</option>' +
'</select>' +
'</label>' +
'<label class="wolai-page-ai-profile-select" data-page-ai-hermes-profile>' +
'<span>agent / profile</span>' + '<span>agent / profile</span>' +
'<select data-page-ai-profile-select></select>' + '<select data-page-ai-profile-select></select>' +
'</label>' + '</label>' +
@@ -5460,7 +5490,7 @@ const SIDEBAR_TREE_JS: &str = r##"
workspaceId: resolveWorkspaceId(document.body), workspaceId: resolveWorkspaceId(document.body),
documentId: currentDocumentId(), documentId: currentDocumentId(),
sessionId: pageUiState.pageAiActiveSessionId, sessionId: pageUiState.pageAiActiveSessionId,
profile: pageAiCurrentProfile(), profile: pageUiState.pageAiAcpRuntime || pageAiCurrentProfile(),
contextScope: pageUiState.pageAiContextScope, contextScope: pageUiState.pageAiContextScope,
message: prompt, message: prompt,
model: pageAiMnoteToolModel(), model: pageAiMnoteToolModel(),
@@ -6498,6 +6528,15 @@ const SIDEBAR_TREE_JS: &str = r##"
}); });
document.addEventListener('change', function(event) { document.addEventListener('change', function(event) {
var pageAiAcpRuntimeSelect = closestAction(event.target, '[data-page-ai-acp-runtime]');
if (pageAiAcpRuntimeSelect instanceof HTMLSelectElement) {
var next = String(pageAiAcpRuntimeSelect.value || '').trim();
pageUiState.pageAiAcpRuntime = next;
void pageAiLoadProfiles();
renderPageAiControls();
renderPageAiProviderButtons();
return;
}
var pageAiProfileSelect = closestAction(event.target, '[data-page-ai-profile-select]'); var pageAiProfileSelect = closestAction(event.target, '[data-page-ai-profile-select]');
if (pageAiProfileSelect instanceof HTMLSelectElement) { if (pageAiProfileSelect instanceof HTMLSelectElement) {
void pageAiSwitchProfile(pageAiProfileSelect.value); void pageAiSwitchProfile(pageAiProfileSelect.value);
@@ -6864,26 +6903,7 @@ const TREE_LIVE_CONTROLLER_JS: &str = r##"
} }
} }
function start() { function startWithSse(bootstrap, workspaceId, url) {
if (!('EventSource' in window)) {
applyStatus('unsupported');
return;
}
var bootstrap = readBootstrap();
var params = new URLSearchParams(window.location.search);
var sourceKind = (params.get('sourceKind') || '').trim();
if (sourceKind === 'local_folder') {
applyTransport('local-folder-static');
applyStatus('static');
return;
}
applyTransport(bootstrap.transport || 'convex-command-log-sse');
var workspaceId = bootstrap.workspaceId || resolveWorkspaceId();
var url = new URL(bootstrap.endpoint || '/api/tree/events', window.location.origin);
url.searchParams.set('workspaceId', workspaceId);
if (Array.isArray(bootstrap.rootIds) && bootstrap.rootIds.length === 1) {
url.searchParams.set('rootNodeId', bootstrap.rootIds[0]);
}
var failures = 0; var failures = 0;
var source = new EventSource(url.toString()); var source = new EventSource(url.toString());
window.__mnoteTreeLiveEventSource = source; window.__mnoteTreeLiveEventSource = source;
@@ -6915,12 +6935,120 @@ const TREE_LIVE_CONTROLLER_JS: &str = r##"
dispatchTreeEvent('tree:resync', { payload: payload, revision: revision, bootstrap: bootstrap }); dispatchTreeEvent('tree:resync', { payload: payload, revision: revision, bootstrap: bootstrap });
}); });
source.addEventListener('block.delta', function(event){
var payload = JSON.parse(event.data || '{}');
dispatchTreeEvent('tree:block-delta', { payload: payload, bootstrap: bootstrap });
});
source.onerror = function(){ source.onerror = function(){
failures += 1; failures += 1;
applyStatus(failures >= 3 ? 'resync-pending' : 'reconnecting'); applyStatus(failures >= 3 ? 'resync-pending' : 'reconnecting');
}; };
} }
function startWithWebSocket(bootstrap, workspaceId, wsUrl) {
var proto = (window.location.protocol === 'https:') ? 'wss:' : 'ws:';
var url = new URL(wsUrl || bootstrap.wsEndpoint || '/api/realtime/ws', window.location.origin);
url.protocol = proto;
url.searchParams.set('workspaceId', workspaceId);
if (Array.isArray(bootstrap.rootIds) && bootstrap.rootIds.length === 1) {
url.searchParams.set('rootNodeId', bootstrap.rootIds[0]);
}
var ws = new WebSocket(url.toString());
window.__mnoteTreeLiveEventSource = ws;
applyStatus('connecting');
ws.onopen = function() {
applyStatus('connected');
};
ws.onmessage = function(event) {
var payload;
try { payload = JSON.parse(event.data); } catch (_) { return; }
var kind = payload.kind || '';
var revision = payload.revision || payload.cursor || '';
if (kind === 'snapshot') {
document.documentElement.setAttribute('data-mnote-tree-live-revision', String(revision));
dispatchTreeEvent('tree:snapshot', { payload: payload, revision: revision, bootstrap: bootstrap });
} else if (kind === 'delta') {
document.documentElement.setAttribute('data-mnote-tree-live-revision', String(revision));
dispatchTreeEvent('tree:delta', { payload: payload, revision: revision, bootstrap: bootstrap });
// Delta indicates something changed; request fresh resync from server
if (ws.readyState === WebSocket.OPEN) {
ws.send('resync');
}
} else if (kind === 'resync') {
document.documentElement.setAttribute('data-mnote-tree-live-revision', String(revision));
dispatchTreeEvent('tree:resync', { payload: payload, revision: revision, bootstrap: bootstrap });
} else if (kind === 'resync_hint') {
// Server suggests re-sync after lagged events
if (ws.readyState === WebSocket.OPEN) {
ws.send('resync');
}
}
};
ws.onclose = function() {
applyStatus('closed');
// Fall back to SSE after a short delay
setTimeout(function() {
if (window.__mnoteTreeLiveEventSource === ws) {
startWithSseFallback(bootstrap, workspaceId);
}
}, 2000);
};
ws.onerror = function() {
applyStatus('error');
};
}
function startWithSseFallback(bootstrap, workspaceId) {
applyTransport('convex-command-log-sse');
var url = new URL(bootstrap.endpoint || '/api/tree/events', window.location.origin);
url.searchParams.set('workspaceId', workspaceId);
if (Array.isArray(bootstrap.rootIds) && bootstrap.rootIds.length === 1) {
url.searchParams.set('rootNodeId', bootstrap.rootIds[0]);
}
startWithSse(bootstrap, workspaceId, url);
}
function start() {
var bootstrap = readBootstrap();
var params = new URLSearchParams(window.location.search);
var sourceKind = (params.get('sourceKind') || '').trim();
if (sourceKind === 'local_folder') {
applyTransport('local-folder-static');
applyStatus('static');
return;
}
var workspaceId = bootstrap.workspaceId || resolveWorkspaceId();
// Prefer WebSocket transport when available
var preferWs = bootstrap.transport === 'convex-command-log-ws' && 'WebSocket' in window;
if (preferWs) {
applyTransport('convex-command-log-ws');
startWithWebSocket(bootstrap, workspaceId, bootstrap.wsEndpoint || '/api/realtime/ws');
return;
}
// Fallback: SSE / EventSource
if (!('EventSource' in window)) {
applyStatus('unsupported');
return;
}
applyTransport(bootstrap.transport || 'convex-command-log-sse');
var url = new URL(bootstrap.endpoint || '/api/tree/events', window.location.origin);
url.searchParams.set('workspaceId', workspaceId);
if (Array.isArray(bootstrap.rootIds) && bootstrap.rootIds.length === 1) {
url.searchParams.set('rootNodeId', bootstrap.rootIds[0]);
}
startWithSse(bootstrap, workspaceId, url);
}
window.__mnoteTreeLiveClose = closeActiveSource; window.__mnoteTreeLiveClose = closeActiveSource;
window.addEventListener('pagehide', closeActiveSource, { once: true }); window.addEventListener('pagehide', closeActiveSource, { once: true });
@@ -6992,11 +7120,12 @@ pub fn PageLayout(
}); });
let tree_live_bootstrap = serde_json::json!({ let tree_live_bootstrap = serde_json::json!({
"schema": "mnote.tree_live_bootstrap.v1", "schema": "mnote.tree_live_bootstrap.v1",
"transport": "convex-command-log-sse", "transport": "convex-command-log-ws",
"workspaceId": null, "workspaceId": null,
"rootIds": [], "rootIds": [],
"initialRevision": null, "initialRevision": null,
"endpoint": "/api/tree/events", "endpoint": "/api/tree/events",
"wsEndpoint": "/api/realtime/ws",
"views": ["page-tree", "file-tree"] "views": ["page-tree", "file-tree"]
}) })
.to_string(); .to_string();