diff --git a/crates/op-host-web/src/canvaskit.rs b/crates/op-host-web/src/canvaskit.rs index e22d32711..81f46a4c8 100644 --- a/crates/op-host-web/src/canvaskit.rs +++ b/crates/op-host-web/src/canvaskit.rs @@ -995,8 +995,12 @@ pub async fn mount_ck(canvas_id: String) -> Result<(), JsValue> { // in the background so the icon picker / figma can resolve simple-icons. crate::iconify_web::fetch_brand_catalog(&inner); // Bidirectional live-canvas sync with the daemon (pull on version bump, - // push local edits + selection) — same loops the skia mount wires. - crate::live_sync_glue::start(&inner); + // push local edits + selection) — same loops the skia mount wires. The + // `SyncController` (gate + wire client + push single-flight) is shared + // with the Task 7 postMessage bridge, so both observe/mutate one instance. + let sync_controller: crate::live_sync_glue::SharedSync = + Rc::new(RefCell::new(crate::live_sync_glue::SyncController::new())); + crate::live_sync_glue::start(&inner, sync_controller.clone()); // Mirror the daemon's agent-indicator registry so design runs paint // their agent borders / badges / reveal animations on web too. crate::agent_indicator_sync::start(&inner); diff --git a/crates/op-host-web/src/live_sync_glue.rs b/crates/op-host-web/src/live_sync_glue.rs index 06c1dc9fb..6ceec43c7 100644 --- a/crates/op-host-web/src/live_sync_glue.rs +++ b/crates/op-host-web/src/live_sync_glue.rs @@ -7,16 +7,22 @@ //! only when the daemon's monotonic version advanced. TS receives //! `document:update` pushes over SSE instead — a documented transport //! divergence (the daemon's SSE stream carries version bumps only, and the -//! verified XHR machinery is reused); worst-case latency is one tick. +//! verified XHR machinery is reused); worst-case latency is one tick. Every +//! pull decision — the tick gate AND the apply-time re-check — is keyed on +//! the CURRENT `(document_generation, document_revision)` pair via +//! [`op_editor_core::sync_gate::SyncGate`], not on any cheap dirty hint. //! * **Push** (browser edits → daemon, so external MCP/CLI clients see them): //! a 2000 ms tick — the TS `PUSH_DEBOUNCE_MS` cadence — serializes the live -//! document when the host flagged a possible change, skips when the content -//! hash matches the last applied/pushed baseline (the race-free analog of -//! TS's `skipPushUntilRef` + `sourceClientId` echo suppression), and POSTs -//! `{document}` to `/api/mcp/document`. Pushes over 2 MiB are skipped with -//! a one-shot console warning (TS `SYNC_MAX_BODY_BYTES` parity). Failures -//! are dropped best-effort (TS catch{} parity) — the next local edit -//! retriggers. +//! document when `SyncGate::needs_push` says the current pair moved past +//! the last-synced baseline (the sole authoritative gate; a stored conflict +//! holds it closed until explicit resolution), skips the actual POST when +//! the content hash matches the daemon baseline (re-baselining directly so +//! the gate can't deadlock), and POSTs `{document, baseVersion}` to +//! `/api/mcp/document`. Pushes over 2 MiB are skipped with a one-shot +//! console warning (TS `SYNC_MAX_BODY_BYTES` parity), baseline untouched. A +//! `version-conflict` response suspends the gate until the host resolves it +//! (no silent retry); other failures are dropped best-effort (TS catch{} +//! parity) — the next local edit retriggers. //! * **Selection push**: the 400 ms tick also samples the selection key and //! POSTs `{selectedIds, activePageId}` to `/api/mcp/selection` when it //! changed (TS debounces 300 ms; a 400 ms trailing sample is the same @@ -25,17 +31,19 @@ //! Architectural divergence (documented): TS's BROWSER is the document //! authority (it pushes its document on `client:id` and the Nitro server only //! caches), while the Rust daemon is the authority after mount — so this glue -//! never pushes before the first daemon document has been applied. The static -//! host page calls `/api/mcp/sync-reset` before mounting so a browser refresh -//! starts from the starter document instead of replaying the previous transient -//! web `.op` state; bootstrap pushes remain disabled so stale page state cannot -//! overwrite a deliberately opened daemon document. Live screenshot requests are -//! served by the `--serve-web` daemon's MCP route from this same synced document -//! authority, rather than by browser-side capture. +//! never pushes before the first daemon document has been applied (subsumed +//! by `SyncGate::needs_push`'s None-baseline rule). The static host page calls +//! `/api/mcp/sync-reset` before mounting so a browser refresh starts from the +//! starter document instead of replaying the previous transient web `.op` +//! state; bootstrap pushes remain disabled so stale page state cannot +//! overwrite a deliberately opened daemon document. Live screenshot requests +//! are served by the `--serve-web` daemon's MCP route from this same synced +//! document authority, rather than by browser-side capture. use std::cell::{Cell, RefCell}; use std::rc::Rc; +use op_editor_core::sync_gate::SyncGate; use op_editor_core::web_sync::{self, WebSyncClient}; use crate::live_sync; @@ -49,15 +57,51 @@ const PUSH_INTERVAL_MS: i32 = 2000; /// are not pushed (warned once), mirroring the renderer's oversize guard. const SYNC_MAX_BODY_BYTES: usize = 2 * 1024 * 1024; +/// Shared sync state: the gate deciding pull/push eligibility, the wire-level +/// client (version/hash bookkeeping), and the push single-flight latch. Built +/// once in `mount_ck` and handed to both this module's ticks and the Task 7 +/// postMessage bridge, so both sides observe and mutate the exact same +/// `SyncGate` instance (a v1 defect — the bridge couldn't reach a local +/// `WebSyncClient` — is fixed by sharing this one struct). +pub(crate) struct SyncController { + pub gate: SyncGate, + pub client: WebSyncClient, + pub push_busy: bool, +} + +impl SyncController { + pub(crate) fn new() -> Self { + Self { + gate: SyncGate::default(), + client: WebSyncClient::new(), + push_busy: false, + } + } +} + +pub(crate) type SharedSync = Rc>; + +/// The document-identity pair every gating decision is keyed on. Read fresh +/// from the live editor state at each decision point — never cached — so an +/// edit that lands between a tick firing and its async response landing is +/// always observed. +fn current_pair(b: &C) -> (u64, u64) { + let s = b.host().editor_state(); + (s.document_generation(), s.document_revision()) +} + /// Wire the bidirectional sync loops onto the mounted shell. Called once from -/// `mount()`; both intervals run for the page lifetime. -pub(crate) fn start(inner: &Rc>) { +/// `mount_ck`; both intervals run for the page lifetime. `sync` is shared with +/// the Task 7 bridge — this module only ever borrows it for the duration of a +/// single decision, never across an await/callback boundary. +pub(crate) fn start(inner: &Rc>, sync: SharedSync) { let base = crate::daemon_base::daemon_base(); - let sync = Rc::new(RefCell::new(WebSyncClient::new())); // One document fetch / one push at a time; ticks observing an in-flight // request skip (the TS hook queues at most one — same effective shape). + // `fetch_busy` stays a plain Cell (pull-side, local to this module); + // `push_busy` moved onto the shared controller so the bridge can also + // observe it. let fetch_busy = Rc::new(Cell::new(false)); - let push_busy = Rc::new(Cell::new(false)); let oversize_warned = Rc::new(Cell::new(false)); // `None` forces a selection re-push on the next tick (used after a doc // apply, which resets daemon-side selection). Seeded with the current key @@ -84,24 +128,42 @@ pub(crate) fn start(inner: &Rc>) { // ---- document push tick ---- { let inner = inner.clone(); + let sync = sync.clone(); let tick: Rc = Rc::new(move || { - push_document_if_changed(&inner, &base, &sync, &push_busy, &oversize_warned); + push_document_if_changed(&inner, &base, &sync, &oversize_warned); }); let _ = live_sync::start_interval(PUSH_INTERVAL_MS, tick); } } /// Probe the daemon version; on a newer version fetch + apply the document. +/// Gated on the sync-gate FIRST (before any network round-trip): an accept +/// window broken by an intervening local edit re-enters the conflict flow, +/// and otherwise `pull_allowed` must hold for the current pair. fn poll_version( inner: &Rc>, base: &str, - sync: &Rc>, + sync: &SharedSync, fetch_busy: &Rc>, last_selection_key: &Rc>>, ) { if fetch_busy.get() { return; } + let pair = current_pair(&*inner.borrow()); + // Borrow discipline: Rust 2021 extends an `if let` scrutinee's temporary + // borrow through the whole arm body, so a `borrow_mut()` inside the body + // below would panic at runtime. Binding the `Option` first lets the + // `borrow()` temporary drop at the end of THIS statement instead. + let broken = sync.borrow().gate.accept_window_broken(pair); // borrow ends HERE + if let Some(v) = broken { + sync.borrow_mut().gate.note_conflict(v); // observer latch reports; user decides again + return; + } + if !sync.borrow().gate.pull_allowed(pair) { + return; + } + let inner = inner.clone(); let sync = sync.clone(); let base_owned = base.to_string(); @@ -113,7 +175,7 @@ fn poll_version( }; let wants_version = sync .try_borrow() - .map(|sync| sync.wants_version(version)) + .map(|s| s.client.wants_version(version)) .unwrap_or(false); if !wants_version { return; @@ -141,37 +203,51 @@ fn poll_version( /// commits that exact version only when the closure returns `true` (swap + /// repaint both succeeded) — so the committed version is never stale and a /// failed repaint is retried on the next poll. On success the local -/// serialization becomes the push baseline (echo suppression) and the -/// selection key is invalidated (the doc swap reset daemon + local selection, -/// so the daemon must be told the browser's current one again). +/// serialization becomes the push baseline (echo suppression), the selection +/// key is invalidated (the doc swap reset daemon + local selection, so the +/// daemon must be told the browser's current one again), and the sync-gate +/// baseline is committed to the PAIR AFTER the apply (a `replace_document` +/// bumps generation, so the pre-apply pair would be stale). fn apply_document_response( inner: &Rc>, body: &str, - sync: &Rc>, + sync: &SharedSync, last_selection_key: &Rc>>, ) { let Ok(mut inner_mut) = inner.try_borrow_mut() else { return; }; let inner_ref = &mut *inner_mut; + // Apply-time re-check: a local edit may have landed while the fetch was + // in flight; re-read the CURRENT pair and re-run `pull_allowed` before + // taking any mutable borrow of the controller. Borrow ends at the end of + // this statement (plain `if`, not `if let` — the condition's temporary + // does not extend into the body). + if !sync.borrow().gate.pull_allowed(current_pair(inner_ref)) { + return; // a local edit landed while the fetch was in flight — abort apply + } // Captured BEFORE sync() flips it on the first apply: the // mount-time pull (starter doc → daemon doc) must NOT become an // undo step; every later external apply (AI turn, MCP client) must. let undoable = sync .try_borrow() - .map(|sync| sync.initialized()) + .map(|s| s.client.initialized()) .unwrap_or(false); + // The existing `WebSyncClient::sync` closure runs while `sync` is held + // mutably borrowed (via `s` below) — it must never itself touch `sync`; + // it only touches `inner_ref`, which is a separate RefCell. let applied = sync .try_borrow_mut() .ok() - .and_then(|mut sync| { - sync.sync(body, |doc, _version| { - inner_ref - .host_mut() - .replace_document_from_sync(doc, undoable); - inner_ref.repaint().is_ok() - }) - .ok() + .and_then(|mut s| { + s.client + .sync(body, |doc, _version| { + inner_ref + .host_mut() + .replace_document_from_sync(doc, undoable); + inner_ref.repaint().is_ok() + }) + .ok() }) .unwrap_or(false); if applied { @@ -179,58 +255,73 @@ fn apply_document_response( // push tick compares apples to apples (serde normalization differs // from the daemon's wire bytes). if let Ok(json) = serde_json::to_string(&inner_ref.host().editor_state().doc) { - if let Ok(mut sync) = sync.try_borrow_mut() { - sync.note_applied_snapshot(&json); + if let Ok(mut s) = sync.try_borrow_mut() { + s.client.note_applied_snapshot(&json); } } if let Ok(mut last_selection_key) = last_selection_key.try_borrow_mut() { *last_selection_key = None; } + // Commit the sync-gate baseline AFTER the apply closure has returned + // (its mutable borrow above is released) and using the POST-apply + // pair (`replace_document` bumped generation). + let post_apply = current_pair(inner_ref); + sync.borrow_mut() + .gate + .note_synced(post_apply.0, post_apply.1); } } -/// Serialize + push the local document when it changed since the last -/// applied/pushed baseline. Never pushes before the first daemon apply -/// (daemon authority — see the module docs). +/// Serialize + conditionally push the local document. `SyncGate::needs_push` +/// is the SOLE authoritative gate (false while a conflict is pending, or +/// while the current pair matches the last-synced baseline — including +/// before the first daemon apply, daemon-authority per the module docs); +/// `take_doc_sync_dirty` is consumed purely as a cheap "maybe changed" hint +/// and never substitutes for the gate. fn push_document_if_changed( inner: &Rc>, base: &str, - sync: &Rc>, - push_busy: &Rc>, + sync: &SharedSync, oversize_warned: &Rc>, ) { - if push_busy.get() { + let busy = sync.borrow().push_busy; + if busy { return; } - let initialized = sync - .try_borrow() - .map(|sync| sync.initialized()) - .unwrap_or(false); - let doc_json = { + let (pair, doc_json) = { let Ok(mut b) = inner.try_borrow_mut() else { return; }; - // Cheap gate: only serialize when the host flagged a possible change - // since the last tick (a conservative superset of document edits; the - // hash check below absorbs the false positives). - if !b.host_mut().take_doc_sync_dirty() { + // Consume the hint — NEVER a gate. A false positive here just costs + // one wasted `needs_push` check below; a false negative would be + // caught by the gate anyway since it reads the live pair directly. + let _ = b.host_mut().take_doc_sync_dirty(); + let pair = current_pair(&*b); + let due = sync.borrow().gate.needs_push(pair); + if !due { return; } let Ok(json) = serde_json::to_string(&b.host().editor_state().doc) else { return; }; - json + (pair, json) }; - let should_push = sync - .try_borrow() - .map(|sync| initialized && sync.should_push(&doc_json)) - .unwrap_or(false); + + let should_push = sync.borrow().client.should_push(&doc_json); if !should_push { + // Bytes already match the daemon baseline (e.g. a generation-only + // replace — same content, new generation): no push to send, but the + // baseline MUST advance directly or `needs_push` stays true forever, + // the hash check keeps skipping the push, and the pull gate never + // reopens (deadlock). + sync.borrow_mut().gate.note_synced(pair.0, pair.1); return; } - if doc_json.len() > SYNC_MAX_BODY_BYTES { - // TS warns once per oversize streak and skips the push. - if !oversize_warned.get() { + if !SyncGate::periodic_push_allowed(doc_json.len()) { + // TS warns once per oversize streak and skips the push; baseline + // stays put (the uncapped snapshot channel is the fallback path). + let warned = oversize_warned.get(); + if !warned { oversize_warned.set(true); web_sys::console::warn_1( &format!( @@ -244,33 +335,48 @@ fn push_document_if_changed( return; } oversize_warned.set(false); - let body = WebSyncClient::wrap_push_body(&doc_json); - push_busy.set(true); - let sync = sync.clone(); - let push_busy_done = push_busy.clone(); + let base_version = sync.borrow().client.last_version(); + let body = WebSyncClient::wrap_push_body_with_base(&doc_json, base_version); + sync.borrow_mut().push_busy = true; + let sync_done = sync.clone(); + // The pair captured AT SERIALIZATION TIME (above), not whatever the + // current pair is when the response lands — an edit landing while this + // push is in flight must keep reporting `needs_push`. + let pushed_pair = pair; let on_response: Rc = Rc::new(move |resp: String| { - push_busy_done.set(false); - if let Some(version) = WebSyncClient::parse_push_response(&resp) { - // Commit baseline + version so our own push is never echoed back. - if let Ok(mut sync) = sync.try_borrow_mut() { - sync.mark_pushed(&doc_json, version); - } + sync_done.borrow_mut().push_busy = false; + // Conflict check first: a rejected push must never fall through to + // the success branch's `note_synced` (which would wrongly reopen + // the pull gate on rejected bytes). + let conflict_version = WebSyncClient::parse_push_conflict(&resp); + if let Some(server_v) = conflict_version { + sync_done.borrow_mut().gate.note_conflict(server_v); + return; } - // A rejected/failed push is dropped best-effort (TS parity) — the - // next local edit re-flags the host and retries. + let accepted_version = WebSyncClient::parse_push_response(&resp); + if let Some(version) = accepted_version { + let mut s = sync_done.borrow_mut(); + s.client.mark_pushed(&doc_json, version); + s.gate.note_synced(pushed_pair.0, pushed_pair.1); + } + // Neither shape recognized (network/parse failure): dropped + // best-effort (TS catch{} parity) — the next local edit re-flags + // needs_push and retries. }); if !live_sync::post_json( &format!("{base}/api/mcp/document"), &body, Some(on_response), ) { - push_busy.set(false); + sync.borrow_mut().push_busy = false; } } /// POST the selection to the daemon when it changed since the last sample /// (TS pushes `{selectedIds, activePageId}` debounced 300 ms; this samples on /// the 400 ms tick). Fire-and-forget like the TS fetch().catch(() => {}). +/// Unaffected by the sync gate (one-way browser → daemon presentation state, +/// not document content). fn push_selection_if_changed( inner: &Rc>, base: &str,