fix: fallback tree live events to polling

2026-05-25 04:38:23 +08:00
parent aa293fec3f
commit 3d338fdde3
2 changed files with 90 additions and 7 deletions
@@ -164,4 +164,9 @@ UI / 浏览器可见项必须有真实浏览器截图或结构化 smoke 证据
  - Worker 验证：`node --check scripts/task487-local-folder-tree-live-consumer-smoke.js`、限定 `git diff --check` 通过；因当时 3000 未启动，未运行真实浏览器 smoke。
  - Codex 主控验证：`node --check scripts/task487-local-folder-tree-live-consumer-smoke.js`、`MNOTE_WEB_SMOKE_BASE_URL=http://127.0.0.1:3001 NODE_PATH=/mnt/Data1T/mnote/node_modules node scripts/task487-local-folder-tree-live-consumer-smoke.js`。结果仍为 RED，`transport=local-folder-events`、PageTree/FileTree 外部变更阶段 `navigationEvents=0`，但 `treeLiveApplied` 仍为空，说明 live consumer 证据仍未收敛到 snapshot/resync。
 - 2026-05-25：Batch P Reasonix 派发失败并拒收：三个输出目录 `reasonix-2026-05-24T20-21-00-338Z-53f263c7` / `reasonix-2026-05-24T20-21-00-457Z-9384a1e5` / `reasonix-2026-05-24T20-21-00-569Z-8a3585e8` 均只有 `prompt.md` / `memory-recall.json` / `reasonix-transcript.jsonl`，没有 `result.json`、`process-handoff.md/json` 或 `final.md`；记录中的三个 worktree 目录当前不存在，Codex 未采纳任何 diff。后续 B/C/D 等实现 worker 必须在独立 worktree 产出候选 patch，再由 Codex 读取 Hindsight recall、handoff、diff 和验证证据后选择性合入。
- 2026-05-25：Batch Q 未经主控批准再次自动启动 3 个 Reasonix runner（`0525-q-worker-a/b/c-*`），超过当前 goal 的 2 worker 上限；Codex 已终止进程，删除本轮自动生成的 3 份 batch-q 草稿并移除空 worktree，不采纳任何结果。后续在修复 runner 自动派发前，不再让 Reasonix 直接负责多 worker 规划。
+- 2026-05-25：Batch Q 未经主控批准再次自动启动 3 个 Reasonix runner（`0525-q-worker-a/b/c-*`），超过当前 goal 的 2 worker 上限；Codex 已终止进程，删除本轮自动生成的 3 份 batch-q 草稿并移除空 worktree，不采纳任何 Reasonix 结果。
+- 2026-05-25：Codex 修复本机 Reasonix runner / skill 的协同约束：`reasonix-coding-runner.mjs` 生成 prompt 时新增“Reasonix 是叶子 worker，禁止创建 `.codex/reasonix-tasks`、启动 runner/ACP/Claude/Codex/Hermes、操作其他 worktree”的硬约束；同步更新 `reasonix-coding-worker` 与 `reasonix-parallel-coding-flow` skill，并新增 runner `--dry-run` 用于不启动 Reasonix 的 prompt 合同 smoke。
+  - 已验证：`node --check reasonix-coding-runner.mjs`、`node --check reasonix-process-extractor.mjs`、`node --test reasonix-process-extractor.test.mjs`、runner `--dry-run --no-memory-recall` 输出的 `prompt.md` 含叶子 worker 禁止项，且 `ps` 未出现 `reasonix` / `reasonix-coding-runner` 进程。
+- 2026-05-25：Codex 独立复核并采纳 Batch Q Worker A 遗留候选 patch：`local_folder_events.rs` 在 `treeLive=true` 且 watcher subscribe 失败时不再返回 500，而是返回同一 SSE 响应并降级为 revision polling；初始仍发送 `snapshot`，revision 变化后通过既有 `rebuild_tree_resync_payload` 发送 `resync`。该 patch 来自无 handoff 的失控 worker，未按 Reasonix 结果直接采纳，而是由 Codex 本地阅读 diff、跑验证后作为主控修复纳入。
+  - 修改：`rust/crates/mnote-web/src/routes/local_folder_events.rs`。
+  - 已验证：`cargo fmt --manifest-path rust/Cargo.toml --all --check`、`cargo check --manifest-path rust/Cargo.toml -p mnote-web`、`cargo test --manifest-path rust/Cargo.toml -p mnote-web local_folder_events -- --test-threads=1`、临时 `MNOTE_WEB_BIND=127.0.0.1:3002 MNOTE_WEB_PUBLIC_BIND=127.0.0.1:3002 cargo run --manifest-path rust/Cargo.toml -p mnote-web --bin mnote-web` 后运行 `MNOTE_WEB_SMOKE_BASE_URL=http://127.0.0.1:3002 NODE_PATH=/mnt/Data1T/mnote/node_modules node scripts/task487-local-folder-tree-live-consumer-smoke.js`，结果 `ok:true`、`treeLiveApplied=\"resync\"`、`transport=\"local-folder-events\"`。
@@ -17,8 +17,9 @@ use serde_json::{json, Value};
 use std::convert::Infallible;
 use std::path::PathBuf;
 use std::pin::Pin;
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::broadcast::error::RecvError;
+use tokio::time::{interval, MissedTickBehavior};

 type BoxedEventStream =
    Pin<Box<dyn futures_util::Stream<Item = Result<SseEvent, Infallible>> + Send>>;
@@ -123,11 +124,6 @@ async fn build_tree_live_stream(
    canonical_root: PathBuf,
    root_uri: String,
 ) -> Result<(HeaderMap, BoxedEventStream), WebError> {
-    let subscription = state
-        .local_folder_watcher_registry()
-        .subscribe(&canonical_root)
-        .map_err(|error| WebError::internal(error).with_context(&context))?;
-
    let workspace_id = local_workspace_id_from_root_uri(&root_uri)
        .map_err(|error| error.with_context(&context))?;

@@ -148,6 +144,27 @@ async fn build_tree_live_stream(
        &file_tree_snapshot,
    );

+    let subscription = match state
+        .local_folder_watcher_registry()
+        .subscribe(&canonical_root)
+    {
+        Ok(subscription) => subscription,
+        Err(error) => {
+            tracing::warn!(
+                error = %error,
+                root_uri = %root_uri,
+                "local_folder tree live watcher unavailable; falling back to revision polling"
+            );
+            let stream = build_tree_live_polling_stream(
+                root_uri,
+                workspace_id,
+                revision.revision,
+                initial_payload,
+            );
+            return Ok((HeaderMap::new(), stream));
+        }
+    };
+
    let stream = stream::unfold(
        (Some(initial_payload), subscription, root_uri, workspace_id),
        |(payload, mut subscription, root_uri, workspace_id)| async move {
@@ -184,6 +201,67 @@ async fn build_tree_live_stream(
    Ok((HeaderMap::new(), stream))
 }

+fn build_tree_live_polling_stream(
+    root_uri: String,
+    workspace_id: String,
+    initial_revision: String,
+    initial_payload: Value,
+) -> BoxedEventStream {
+    let mut poll_interval = interval(Duration::from_millis(1_200));
+    poll_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+    stream::unfold(
+        Some(TreeLivePollingState {
+            root_uri,
+            workspace_id,
+            current_revision: initial_revision,
+            initial_payload: Some(initial_payload),
+            poll_interval,
+        }),
+        |state| async move {
+            let mut state = state?;
+            if let Some(payload) = state.initial_payload.take() {
+                return Some((Ok(stream_event("snapshot", &payload)), Some(state)));
+            }
+
+            loop {
+                state.poll_interval.tick().await;
+                let Some((next_revision, resync_payload)) = tree_live_polling_resync_payload(
+                    &state.root_uri,
+                    &state.workspace_id,
+                    &state.current_revision,
+                ) else {
+                    continue;
+                };
+                state.current_revision = next_revision;
+                return Some((Ok(stream_event("resync", &resync_payload)), Some(state)));
+            }
+        },
+    )
+    .boxed()
+}
+
+struct TreeLivePollingState {
+    root_uri: String,
+    workspace_id: String,
+    current_revision: String,
+    initial_payload: Option<Value>,
+    poll_interval: tokio::time::Interval,
+}
+
+fn tree_live_polling_resync_payload(
+    root_uri: &str,
+    workspace_id: &str,
+    current_revision: &str,
+) -> Option<(String, Value)> {
+    let next_revision = local_folder_watch_revision(root_uri).ok()?;
+    if next_revision.revision == current_revision {
+        return None;
+    }
+    let resync_payload = rebuild_tree_resync_payload(root_uri, workspace_id)?;
+    Some((next_revision.revision, resync_payload))
+}
+
 fn rebuild_tree_resync_payload(root_uri: &str, workspace_id: &str) -> Option<Value> {
    let revision = local_folder_watch_revision(root_uri).ok()?;
    let sidebar_snapshot = load_local_folder_page_tree_snapshot(root_uri).ok()?;