SemianiakaVY/openpencil
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

feat(web): route live sync through the shared gate with conditional pushes

Browse source
This commit is contained in:
Kayshen-X 2026-07-17 00:25:20 +08:00
parent f0111bfc4f
commit 8dfd5bfbe7
2 changed files with 185 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
crates/op-host-web/src/canvaskit.rs
View file

@ -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);

252
crates/op-host-web/src/live_sync_glue.rs
View file

@ -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<RefCell<SyncController>>;
/// 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<C: RepaintContext>(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<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) {
/// `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<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>, 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<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) {
// ---- document push tick ----
{
let inner = inner.clone();
let sync = sync.clone();
let tick: Rc<dyn Fn()> = 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<C: RepaintContext + 'static>(
inner: &Rc<RefCell<C>>,
base: &str,
sync: &Rc<RefCell<WebSyncClient>>,
sync: &SharedSync,
fetch_busy: &Rc<Cell<bool>>,
last_selection_key: &Rc<RefCell<Option<String>>>,
) {
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<C: RepaintContext + 'static>(
};
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<C: RepaintContext + 'static>(
/// 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<C: RepaintContext + 'static>(
inner: &Rc<RefCell<C>>,
body: &str,
sync: &Rc<RefCell<WebSyncClient>>,
sync: &SharedSync,
last_selection_key: &Rc<RefCell<Option<String>>>,
) {
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<C: RepaintContext + 'static>(
// 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<C: RepaintContext + 'static>(
inner: &Rc<RefCell<C>>,
base: &str,
sync: &Rc<RefCell<WebSyncClient>>,
push_busy: &Rc<Cell<bool>>,
sync: &SharedSync,
oversize_warned: &Rc<Cell<bool>>,
) {
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<C: RepaintContext + 'static>(
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<dyn Fn(String)> = 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<C: RepaintContext + 'static>(
inner: &Rc<RefCell<C>>,
base: &str,