chore: 保存当前架构收口与 bug 修复快照

归档本轮 P0/P1 bug 修复、设计审查迁移、AI selection scope 收口与 stream contract 调整，并保留当前 05 主线迁移起点。

rust/crates/mnote-web/src/routes/sse.rs

@@ -2,9 +2,9 @@ use crate::app::AppState;
 use crate::context::RequestContext;
 use crate::error::WebError;
 use crate::routes::stream_support::{
-    build_stream_delta_payload, load_stream_overview, load_stream_snapshot,
-    read_stream_cursor_from_payload, resolve_stream_change, with_stream_kind, StreamChangeKind,
-    StreamSnapshotQuery,
+    build_stream_delta_payload, build_stream_push_delta_hint, load_stream_overview,
+    load_stream_snapshot, read_stream_cursor_from_payload, resolve_stream_change, with_stream_kind,
+    StreamChangeKind, StreamSnapshotQuery,
 };
 use axum::extract::{Extension, Query, State};
 use axum::http::{HeaderMap, HeaderName, HeaderValue};
@@ -94,15 +94,11 @@ async fn events_with_stream_delta(
             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"),
-                        });
+                        let hint = build_stream_push_delta_hint(
+                            &payload,
+                            &state.context.trace.request_id,
+                            &state.context.trace.trace_id,
+                        );
                         return Some((Ok(stream_event("delta", &hint)), Some(state)));
                     }
                     Err(tokio::sync::broadcast::error::TryRecvError::Empty) => {}
@@ -113,9 +109,68 @@ async fn events_with_stream_delta(
                 }
             }
 
-            // If push-driven and we already checked both broadcasts, brief sleep then re-check
+            // If push-driven and broadcasts are quiet, polling remains the safety net.
             if state.stream_delta_rx.is_some() {
-                sleep(Duration::from_millis(250)).await;
+                if let Some(max_polls) = max_polls {
+                    if state.polls >= max_polls {
+                        return None;
+                    }
+                }
+                state.polls += 1;
+                sleep(Duration::from_millis(poll_ms)).await;
+
+                let poll_query = live_poll_query(&state.query);
+                let Ok((workspace_id, overview)) =
+                    load_stream_overview(state.app_state.config(), &state.context, &poll_query)
+                        .await
+                else {
+                    return None;
+                };
+                if let Some(change) =
+                    resolve_stream_change(&overview, state.current_cursor.as_deref())
+                {
+                    state.current_cursor = change.cursor.clone();
+                    match change.kind {
+                        StreamChangeKind::Delta => {
+                            let Ok(payload) = build_stream_delta_payload(
+                                state.app_state.config(),
+                                &state.context,
+                                &poll_query,
+                                &workspace_id,
+                                &overview,
+                                change.cursor,
+                                change
+                                    .delta
+                                    .unwrap_or_else(|| serde_json::json!({ "op": "noop" })),
+                            )
+                            .await
+                            else {
+                                return None;
+                            };
+                            return Some((Ok(stream_event("delta", &payload)), Some(state)));
+                        }
+                        StreamChangeKind::Resync => {
+                            let Ok(snapshot_payload) = load_stream_snapshot(
+                                state.app_state.config(),
+                                &state.context,
+                                &poll_query,
+                            )
+                            .await
+                            else {
+                                return None;
+                            };
+                            state.current_cursor =
+                                read_stream_cursor_from_payload(&snapshot_payload);
+                            return Some((
+                                Ok(stream_event(
+                                    "resync",
+                                    &with_stream_kind(&snapshot_payload, "resync"),
+                                )),
+                                Some(state),
+                            ));
+                        }
+                    }
+                }
                 return Some((Ok(stream_event("heartbeat", &json!({}))), Some(state)));
             }
 
@@ -289,6 +344,8 @@ mod tests {
     use crate::routes::stream_support::StreamSnapshotQuery;
     use axum::body::{to_bytes, Body};
     use axum::http::{Request, StatusCode};
+    use std::time::Duration;
+    use tokio::time::timeout;
     use tower::util::ServiceExt;
 
     fn app() -> axum::Router {
@@ -395,6 +452,31 @@ mod tests {
         assert!(text.contains("\"revision\""));
     }
 
+    #[tokio::test]
+    async fn tree_events_push_mode_honors_polling_safety_net_max_polls() {
+        let response = app()
+            .oneshot(
+                Request::builder()
+                    .uri("/api/tree/events?workspaceId=ws_demo&maxPolls=1&pollMs=1")
+                    .body(Body::empty())
+                    .expect("request"),
+            )
+            .await
+            .expect("response");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = timeout(
+            Duration::from_secs(2),
+            to_bytes(response.into_body(), usize::MAX),
+        )
+        .await
+        .expect("push SSE stream should stop after maxPolls")
+        .expect("body");
+        let text = String::from_utf8(body.to_vec()).expect("utf8");
+        assert!(text.contains("event: snapshot") || text.contains("event:snapshot"));
+        assert!(text.contains("event: heartbeat") || text.contains("event:heartbeat"));
+    }
+
     #[test]
     fn live_poll_query_drops_bridge_pagination_cursor() {
         let query = StreamSnapshotQuery {