diff --git a/crates/op-host-services/src/web_static.rs b/crates/op-host-services/src/web_static.rs index 11b6e838c..496d0dfbc 100644 --- a/crates/op-host-services/src/web_static.rs +++ b/crates/op-host-services/src/web_static.rs @@ -436,10 +436,11 @@ mod tests { assert_eq!(reply.status, "200 OK"); assert_eq!(reply.content_type, "text/html; charset=utf-8"); let body = String::from_utf8(reply.body).expect("utf8"); - // The host page loads the glue, resets through the daemon, and mounts - // the production CanvasKit shell on the canvas. + // The host page loads the glue and mounts the production CanvasKit + // shell on the canvas. The daemon sync-reset moved into the wasm mount + // (Task 7), so the page no longer issues it directly. assert!(body.contains("/pkg/op_host_web.js"), "{body}"); - assert!(body.contains("fetch('/api/mcp/sync-reset'"), "{body}"); + assert!(!body.contains("fetch('/api/mcp/sync-reset'"), "{body}"); assert!(body.contains("await mod.mount_ck('op')"), "{body}"); assert!(body.contains("data-op-smoke"), "{body}"); assert!(!body.contains("mod.mount('op')"), "{body}"); diff --git a/crates/op-host-services/src/web_static/index.html b/crates/op-host-services/src/web_static/index.html index 96f3ad25b..216e2e106 100644 --- a/crates/op-host-services/src/web_static/index.html +++ b/crates/op-host-services/src/web_static/index.html @@ -78,21 +78,15 @@ window.__opCanvasScale = dpr; return changed; }; - const resetSyncDocument = async () => { - const response = await fetch('/api/mcp/sync-reset', { - method: 'POST', - cache: 'no-store', - }); - if (!response.ok) { - throw new Error(`sync reset failed: ${response.status}`); - } - }; // Size the canvas to the window BEFORE mount — the shell reads the // canvas width/height attributes when it builds its raster surface. + // The daemon sync-reset moved into the wasm mount (Task 7): the managed + // (VS Code) path issues it with the auth token after the bridge Init, and + // the direct-open path issues it right after mount — both gate the pull + // tick on its completion. markSmoke('booting'); sizeCanvasForDisplay(BOOT_CANVAS_PIXEL_BUDGET); try { - await resetSyncDocument(); const mod = await import('/pkg/op_host_web.js'); await mod.default(); sizeCanvasForDisplay(STEADY_CANVAS_PIXEL_BUDGET); diff --git a/crates/op-host-web/Cargo.toml b/crates/op-host-web/Cargo.toml index e77e2fdbd..1903ef147 100644 --- a/crates/op-host-web/Cargo.toml +++ b/crates/op-host-web/Cargo.toml @@ -114,6 +114,9 @@ features = [ "KeyboardEvent", # `daemon_base` derives the daemon origin from `window.location`. "Location", + # postMessage bridge (`vscode_bridge`): the window `message` listener reads + # `MessageEvent::{data,origin,source}`. + "MessageEvent", "MouseEvent", "Navigator", "Performance", diff --git a/crates/op-host-web/src/canvaskit.rs b/crates/op-host-web/src/canvaskit.rs index 81f46a4c8..bf660ec6b 100644 --- a/crates/op-host-web/src/canvaskit.rs +++ b/crates/op-host-web/src/canvaskit.rs @@ -952,10 +952,6 @@ pub async fn mount_ck(canvas_id: String) -> Result<(), JsValue> { a11y, ime, })); - crate::web_credential_sync::start(); - if let Some(json) = initial_credential_json { - crate::web_credential_sync::credential_changed(json); - } { let mut b = inner.borrow_mut(); let _ = b.resize_to_window(&window)?; @@ -988,22 +984,66 @@ pub async fn mount_ck(canvas_id: String) -> Result<(), JsValue> { // when the read lands so their text re-shapes with the imported typeface). crate::web_fonts::load_imported_fonts_at_mount(&inner); + // ---- daemon bootstrap (startup order, Task 7) ---- + // + // The `SyncController` (gate + wire client + push single-flight) is shared + // with the postMessage bridge, so both observe/mutate one instance. Build + // it FIRST so the bridge listener installs before any daemon service. + let sync_controller: crate::live_sync_glue::SharedSync = + Rc::new(RefCell::new(crate::live_sync_glue::SyncController::new())); + // 1. Install the bridge listener + observer BEFORE any daemon request, so an + // Init / OpenDocument arriving during bootstrap is never missed. + crate::vscode_bridge::install(&inner, sync_controller.clone()); + // 2. Inside a webview iframe, await the host's Init (token) with a 2s + // fallback (proceed as a direct open on timeout). A standalone browser + // tab is a direct open and continues immediately. + if crate::vscode_bridge::in_iframe(&window) { + crate::vscode_bridge::await_init(&window, 2000).await; + } + // 3. Only now start the daemon-dependent services — in managed mode the + // token is present so their requests carry the auth header. + crate::web_credential_sync::start(); + if let Some(json) = initial_credential_json { + crate::web_credential_sync::credential_changed(json); + } // Populate the chat model picker from the daemon's `/api/ai/models` // catalog (best-effort; async, repaints when the response lands). crate::web_chat::fetch_models(&inner); // Pull the brand-logo catalog (omitted from the wasm bundle) from the daemon // 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. 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); + // 4. Reset the daemon's transient sync document, THEN start the live-sync + // ticks. The 400 ms pull tick must not run before the reset completes + // (in BOTH managed + direct paths) or it pulls the pre-reset state. + // Managed mode issues the reset with the token (via the live_sync + // helper); direct open issues the legacy reset here — replacing the + // fetch removed from index.html. + { + let base = crate::daemon_base::daemon_base(); + let inner_for_sync = inner.clone(); + let sync_for_start = sync_controller.clone(); + let started = std::rc::Rc::new(std::cell::Cell::new(false)); + let start_once: std::rc::Rc = { + let started = started.clone(); + std::rc::Rc::new(move || { + if started.replace(true) { + return; + } + crate::live_sync_glue::start(&inner_for_sync, sync_for_start.clone()); + }) + }; + let on_reset: std::rc::Rc = { + let start_once = start_once.clone(); + std::rc::Rc::new(move |_body: String| start_once()) + }; + if !crate::live_sync::post_json(&format!("{base}/api/mcp/sync-reset"), "", Some(on_reset)) { + // Request could not even start — don't wedge: start the ticks anyway. + start_once(); + } + } let mut listeners: Vec = Vec::new(); let canvas_target: web_sys::EventTarget = canvas.clone().into(); diff --git a/crates/op-host-web/src/iconify_web.rs b/crates/op-host-web/src/iconify_web.rs index bd8f2422a..bb355f41a 100644 --- a/crates/op-host-web/src/iconify_web.rs +++ b/crates/op-host-web/src/iconify_web.rs @@ -316,6 +316,11 @@ fn fetch_text(url: &str, on_done: DoneFn) { resolve(&slot, Err("iconify request open failed".to_string())); return; } + // `fetch_text` serves BOTH the daemon brand-catalog fetch and the public + // Iconify CDN search/collection fetches. `attach_daemon_headers` only + // attaches the token when `url` targets the daemon origin, so the public + // Iconify requests are guaranteed never to leak the token. + crate::live_sync::attach_daemon_headers(&xhr, url); xhr.set_timeout(FETCH_TIMEOUT_MS); let xhr_cb = xhr.clone(); let slot_cb = slot.clone(); diff --git a/crates/op-host-web/src/lib.rs b/crates/op-host-web/src/lib.rs index b3d928708..0631baa1c 100644 --- a/crates/op-host-web/src/lib.rs +++ b/crates/op-host-web/src/lib.rs @@ -78,6 +78,11 @@ mod web_design_md; mod web_model_catalog; #[cfg(feature = "canvaskit")] mod web_settings; +// postMessage bridge to the VS Code extension host (token bootstrap, document +// open, snapshot, save-committed, conflict resolution). DOM wiring only — the +// wire codec lives in `op_editor_core::bridge_protocol`. +#[cfg(feature = "canvaskit")] +mod vscode_bridge; #[cfg(feature = "canvaskit")] mod web_storage; // Pure web_sys clipboard/download — Ctrl+C/X in inputs + Figma/file paste. diff --git a/crates/op-host-web/src/live_sync.rs b/crates/op-host-web/src/live_sync.rs index 856c159d2..58bc3dbd3 100644 --- a/crates/op-host-web/src/live_sync.rs +++ b/crates/op-host-web/src/live_sync.rs @@ -12,6 +12,7 @@ //! host-unit-tested. #![allow(dead_code)] +use std::cell::RefCell; use std::rc::Rc; use wasm_bindgen::prelude::*; @@ -19,6 +20,40 @@ use wasm_bindgen::JsCast; const STATUS_REQUEST_TIMEOUT_MS: u32 = 15_000; +thread_local! { + /// Managed-daemon auth token supplied by the VS Code host over the + /// postMessage bridge's `Init`. `None` for a direct-open browser tab (no + /// token → no header). Lives here (always-compiled module) rather than in + /// the `canvaskit`-gated `vscode_bridge` so [`attach_daemon_headers`] can + /// read it from every request helper regardless of build config. + static BRIDGE_TOKEN: RefCell> = const { RefCell::new(None) }; +} + +/// Store the managed-daemon token (from the bridge `Init`). Idempotent. +pub fn set_bridge_token(token: String) { + BRIDGE_TOKEN.with(|t| *t.borrow_mut() = Some(token)); +} + +/// The stored managed-daemon token, if any. `Some` only in a managed webview +/// after the host's `Init` landed. +pub fn bridge_token() -> Option { + BRIDGE_TOKEN.with(|t| t.borrow().clone()) +} + +/// Attach the managed-daemon auth header to `req` — but ONLY when `url` targets +/// the daemon (`daemon_base()` prefix). Public requests (e.g. the Iconify CDN) +/// MUST NEVER carry the token, so the prefix check is the leak guard: a URL that +/// is not the daemon origin never receives the header even when a token is set. +/// Call AFTER `open` and BEFORE `send` (request headers require an open XHR). +pub fn attach_daemon_headers(req: &web_sys::XmlHttpRequest, url: &str) { + let Some(token) = bridge_token() else { + return; + }; + if url.starts_with(&crate::daemon_base::daemon_base()) { + let _ = req.set_request_header("X-OpenPencil-Token", &token); + } +} + /// Run `tick` every `interval_ms` for the page lifetime (the interval owns /// the closure — same `forget()` idiom the previous document poll used). pub fn start_interval(interval_ms: i32, tick: Rc) -> Result<(), JsValue> { @@ -45,6 +80,7 @@ pub fn get(url: &str, on_response: Rc) -> bool { if xhr.open_with_async("GET", url, true).is_err() { return false; } + attach_daemon_headers(&xhr, url); let xhr_for_load = xhr.clone(); // `once_into_js` self-cleans after a single firing (no leaked closure per // request, unlike `forget()`). @@ -70,6 +106,7 @@ pub fn get_with_status(url: &str, on_response: Rc) -> bool if xhr.open_with_async("GET", url, true).is_err() { return false; } + attach_daemon_headers(&xhr, url); xhr.set_timeout(STATUS_REQUEST_TIMEOUT_MS); let xhr_for_load = xhr.clone(); let onloadend = Closure::::once_into_js(move || { @@ -96,6 +133,7 @@ pub fn post_json(url: &str, body: &str, on_response: Option>) if xhr.open_with_async("POST", url, true).is_err() { return false; } + attach_daemon_headers(&xhr, url); let _ = xhr.set_request_header("Content-Type", "application/json"); if let Some(on_response) = on_response { let xhr_for_load = xhr.clone(); @@ -122,6 +160,7 @@ pub fn post_json_with_status(url: &str, body: &str, on_response: Rc>>; + +thread_local! { + /// Locked host origin — recorded from the FIRST valid `init`'s + /// `event.origin`, then enforced on every later message (mismatch dropped). + static BRIDGE_ORIGIN: RefCell> = const { RefCell::new(None) }; + /// The pending `await_init` promise resolver — the `init` handler calls it + /// so `mount_ck` stops waiting the moment the token lands. + static INIT_RESOLVER: RefCell> = const { RefCell::new(None) }; +} + +// --------------------------------------------------------------------------- +// Install + startup coordination +// --------------------------------------------------------------------------- + +/// Install the window `message` listener + the outbound-event tick observer. +/// The listener `Closure` is deliberately leaked (`forget`) — the bridge lives +/// for the whole page, exactly like the other page-level listeners; returning an +/// owning handle would tear the bridge down when `mount_ck` returns. +pub(crate) fn install(inner: &Rc>, sync: SharedSync) { + let Some(window) = web_sys::window() else { + return; + }; + + // Message listener. + { + let inner = inner.clone(); + let sync = sync.clone(); + let closure = Closure::::new(move |evt: MessageEvent| { + handle_message(&inner, &sync, &evt); + }); + let _ = + window.add_event_listener_with_callback("message", closure.as_ref().unchecked_ref()); + closure.forget(); // page-lifetime listener, deliberately leaked + } + + // Outbound-event tick observer — the SOLE emitter of the three edge/state + // events. Seed the dirty triple with the current state so only real changes + // (not the starter-document baseline) produce a `dirty-changed`. + let seed = read_triple(inner); + let last_triple = Rc::new(RefCell::new(seed)); + { + let inner = inner.clone(); + let sync = sync.clone(); + let last_triple = last_triple.clone(); + let tick: Rc = Rc::new(move || observe_tick(&inner, &sync, &last_triple)); + let _ = live_sync::start_interval(BRIDGE_TICK_INTERVAL_MS, tick); + } +} + +/// True when the page runs inside a frame (`window.self != window.top`) — the +/// VS Code webview case, where `mount_ck` awaits the host's `init` before it +/// bootstraps the daemon services. +pub(crate) fn in_iframe(window: &web_sys::Window) -> bool { + let self_win = window.self_(); + // `top`/`parent` return a cross-origin-accessible WindowProxy; an identity + // compare never touches a property so it can't throw. Check both so a host + // that shadows one (some webview shells) is still detected. + let differs = |other: Result, JsValue>| { + other + .ok() + .flatten() + .map(|w| !Object::is(self_win.as_ref(), w.as_ref())) + .unwrap_or(false) + }; + differs(window.top()) || differs(window.parent()) +} + +/// Await the host's `init` (which resolves the promise from the message +/// handler) or a `timeout_ms` fallback, whichever comes first. On timeout the +/// caller proceeds as a direct open (no token). Returns after the promise +/// settles; check [`live_sync::bridge_token`] to learn which path won. +pub(crate) async fn await_init(window: &web_sys::Window, timeout_ms: i32) { + let window = window.clone(); + let promise = Promise::new(&mut |resolve, _reject| { + INIT_RESOLVER.with(|r| *r.borrow_mut() = Some(resolve.clone())); + // Timeout fallback: resolve the same promise so the await unblocks even + // if no host is listening (standalone browser tab, or a slow host). + let resolve_timeout = resolve.clone(); + let cb = Closure::once_into_js(move || { + let _ = resolve_timeout.call0(&JsValue::NULL); + }); + let _ = window + .set_timeout_with_callback_and_timeout_and_arguments_0(cb.unchecked_ref(), timeout_ms); + }); + let _ = wasm_bindgen_futures::JsFuture::from(promise).await; + // Drop the resolver so a late `init` doesn't try to settle a done promise. + INIT_RESOLVER.with(|r| *r.borrow_mut() = None); + if live_sync::bridge_token().is_none() { + web_sys::console::warn_1(&JsValue::from_str( + "[op-bridge] init not received before timeout; proceeding as direct open", + )); + } +} + +// --------------------------------------------------------------------------- +// Inbound message routing +// --------------------------------------------------------------------------- + +fn handle_message( + inner: &Rc>, + sync: &SharedSync, + evt: &MessageEvent, +) { + // Source lock: only the parent frame (the VS Code webview host) may drive + // the bridge. `event.source == window.parent`. + let parent = web_sys::window().and_then(|w| w.parent().ok().flatten()); + let source_ok = match (evt.source(), parent) { + (Some(src), Some(par)) => Object::is(src.as_ref(), par.as_ref()), + _ => false, + }; + if !source_ok { + return; + } + + // Bridge messages are JSON strings; anything else (react-devtools objects, + // etc.) is foreign traffic — silently ignored, never an error. + let Some(raw) = evt.data().as_string() else { + return; + }; + let Some(msg) = BridgeInbound::parse(&raw) else { + return; // non-bridge / malformed + }; + + // Origin lock: the first valid `init` records the origin; every later + // message (including a re-`init`) must match it. A non-init arriving before + // any lock is dropped. + let origin = evt.origin(); + let is_init = matches!(msg, BridgeInbound::Init { .. }); + let locked = BRIDGE_ORIGIN.with(|o| o.borrow().clone()); + match &locked { + Some(l) if *l != origin => return, + None if !is_init => return, + _ => {} + } + if is_init && locked.is_none() { + BRIDGE_ORIGIN.with(|o| *o.borrow_mut() = Some(origin.clone())); + } + + match msg { + BridgeInbound::Init { token } => handle_init(inner, token), + BridgeInbound::OpenDocument { json } => handle_open_document(inner, sync, json), + BridgeInbound::Snapshot { request_id, .. } => handle_snapshot(inner, sync, request_id), + BridgeInbound::SaveCommitted { + generation, + revision, + } => handle_save_committed(inner, generation, revision), + BridgeInbound::ResolveConflict { mode, request_id } => { + handle_resolve_conflict(inner, sync, mode, request_id) + } + } +} + +/// `init`: store the managed token, unblock `mount_ck`'s `await_init`, and reply +/// with the current `(generation, revision)` so the host learns the starter +/// document's identity before it sends `open-document`. +fn handle_init(inner: &Rc>, token: String) { + live_sync::set_bridge_token(token); + INIT_RESOLVER.with(|r| { + if let Some(resolve) = r.borrow_mut().take() { + let _ = resolve.call0(&JsValue::NULL); + } + }); + let (gen, rev, _) = read_triple(inner).unwrap_or((0, 0, false)); + post_to_parent(&event_ready(gen, rev)); +} + +/// `open-document`: SYNCHRONOUS PROLOGUE (fixed order, before the first await) — +/// `replace_document` mints the target generation `G`, `note_open_pending(G)` +/// scopes the open + blocks pulls, then a probe-conditional push carries the +/// opened bytes to the daemon. `opened` is NOT posted here: `note_synced` on +/// push confirmation sets the opened latch, which the observer drains. +fn handle_open_document( + inner: &Rc>, + sync: &SharedSync, + json: String, +) { + let doc: PenDocument = match serde_json::from_str(&json) { + Ok(doc) => doc, + Err(err) => { + web_sys::console::warn_1(&JsValue::from_str(&format!( + "[op-bridge] open-document: bad JSON: {err}" + ))); + return; + } + }; + + // Prologue borrow: replace + repaint + capture the opened pair/bytes. + let (pair, doc_json) = { + let Ok(mut b) = inner.try_borrow_mut() else { + return; + }; + b.host_mut().editor_state_mut().replace_document(doc); + b.host_mut().force_rotate_layer_panel_owner(); + b.host_mut().mark_editor_state_dirty(); + let (w, h) = b.viewport_size(); + b.host_mut().fit_content_to_viewport(w, h); + let _ = b.repaint(); + let s = b.host().editor_state(); + let pair = (s.document_generation(), s.document_revision()); + match serde_json::to_string(&s.doc) { + Ok(json) => (pair, json), + Err(_) => return, + } + }; + + // Scope the open to generation G and block pulls, all before any await. + if let Ok(mut s) = sync.try_borrow_mut() { + s.gate.note_open_pending(pair.0); + } else { + return; + } + + let base = crate::daemon_base::daemon_base(); + drive_open_push(sync.clone(), base, pair, doc_json); +} + +/// `snapshot`: independent of the tick. Conflict pending → reply +/// `snapshot-conflict` at once (host must resolve first). Otherwise capture the +/// pair + bytes atomically; if a push is due, send it over the uncapped +/// snapshot channel (respecting `push_busy` serialization) and reply on +/// confirmation; if nothing is due, reply immediately with the current bytes. +fn handle_snapshot( + inner: &Rc>, + sync: &SharedSync, + request_id: String, +) { + if let Some(server_v) = sync.try_borrow().ok().and_then(|s| s.gate.conflict()) { + post_to_parent(&event_snapshot_conflict(&request_id, server_v)); + return; + } + let Some((pair, doc_json)) = snapshot_state(inner) else { + return; + }; + let needs_push = sync + .try_borrow() + .map(|s| s.gate.needs_push(pair)) + .unwrap_or(false); + if !needs_push { + post_to_parent(&event_snapshot_result( + &request_id, + &doc_json, + pair.0, + pair.1, + )); + return; + } + let base = crate::daemon_base::daemon_base(); + drive_snapshot_push(sync.clone(), base, request_id, pair, doc_json); +} + +/// `save-committed`: mark the reported revision saved. A stale generation +/// (the host acked a save for a document already replaced) returns `false` and +/// is silently dropped. The resulting dirty-flag flip is reported by the +/// observer as `dirty-changed`, not here. +fn handle_save_committed( + inner: &Rc>, + generation: u64, + revision: u64, +) { + let Ok(mut b) = inner.try_borrow_mut() else { + return; + }; + let _ = b + .host_mut() + .editor_state_mut() + .mark_saved_revision_at(generation, revision); + let _ = b.repaint(); +} + +fn handle_resolve_conflict( + inner: &Rc>, + sync: &SharedSync, + mode: ConflictMode, + request_id: String, +) { + match mode { + ConflictMode::UseLocal => resolve_use_local(inner, sync, request_id), + ConflictMode::AcceptRemote => resolve_accept_remote(inner, sync, request_id), + } +} + +/// `resolve-conflict: use-local`: re-push the local document over the snapshot +/// channel using the conflict's server version as `baseVersion`. Success → +/// `mark_pushed` + `note_synced` (clears the conflict, and any pending open, +/// whose `opened` the observer then reports) + `conflict-resolved`. A second +/// conflict retries once with the fresh server version; still failing → +/// `snapshot-conflict`. +fn resolve_use_local( + inner: &Rc>, + sync: &SharedSync, + request_id: String, +) { + let Some(server_v) = sync.try_borrow().ok().and_then(|s| s.gate.conflict()) else { + // No conflict pending — nothing to re-push. Acknowledge idempotently. + post_to_parent(&event_conflict_resolved(&request_id)); + return; + }; + let Some((pair, doc_json)) = snapshot_state(inner) else { + return; + }; + let base = crate::daemon_base::daemon_base(); + drive_use_local_push(sync.clone(), base, request_id, pair, doc_json, server_v); +} + +/// `resolve-conflict: accept-remote`: first reply `snapshot-result` with the +/// LOCAL bytes (the host keeps a backup — spec "neither version is lost"), then +/// hand the gate the accept window over the current pair (pull re-opens for +/// THIS pair only; open_pending is retained), then reply `conflict-resolved`. +/// `opened` is NOT posted now — the remote is not applied yet; the resolving +/// pull's `note_synced` sets the opened latch, which the observer drains then. +fn resolve_accept_remote( + inner: &Rc>, + sync: &SharedSync, + request_id: String, +) { + let Some((pair, doc_json)) = snapshot_state(inner) else { + return; + }; + post_to_parent(&event_snapshot_result( + &request_id, + &doc_json, + pair.0, + pair.1, + )); + if let Ok(mut s) = sync.try_borrow_mut() { + s.gate.resolve_accept_remote(pair); + } else { + return; + } + post_to_parent(&event_conflict_resolved(&request_id)); +} + +// --------------------------------------------------------------------------- +// Push drivers (SharedSync-only; no `inner`, no `C`) +// --------------------------------------------------------------------------- + +/// Acquire `push_busy` (waiting via a short self-reschedule while an in-flight +/// push holds it — the open's state is already latched, and `open_pull_block` +/// keeps pulls out meanwhile), then run the probe-conditional open push. +fn drive_open_push(sync: SharedSync, base: String, pair: (u64, u64), doc_json: String) { + if !acquire_push_busy(&sync) { + schedule_once(PUSH_BUSY_RETRY_MS, move || { + drive_open_push(sync, base, pair, doc_json) + }); + return; + } + open_push_attempt(sync, base, pair, doc_json, RETRY_ONCE); +} + +/// Probe `GET /api/mcp/version` for the daemon's live version `V` (NOT +/// `last_version()` — during bootstrap that is 0 while sync-reset already +/// bumped it, forcing a spurious 409), then conditionally push with +/// `baseVersion=V`. 409 → re-probe + retry once; still 409 → `note_conflict` +/// (its latch drives the observer's `sync-conflict`), release `push_busy`, no +/// `opened`. `push_busy` is HELD across the retry (single-flight preserved). +fn open_push_attempt( + sync: SharedSync, + base: String, + pair: (u64, u64), + doc_json: String, + retries_left: u8, +) { + let version_url = format!("{base}/api/mcp/version"); + let on_version: Rc = { + let sync = sync.clone(); + Rc::new(move |body: String| { + let Some(v) = WebSyncClient::parse_version_probe(&body) else { + // Daemon down / non-JSON — abort without wedging: release the + // latch, leave open_pending (a later retry / reload recovers). + release_push_busy(&sync); + return; + }; + let push_body = WebSyncClient::wrap_push_body_with_base(&doc_json, v); + let doc_url = format!("{base}/api/mcp/document"); + let on_resp: Rc = { + let sync = sync.clone(); + let base = base.clone(); + let doc_json = doc_json.clone(); + Rc::new(move |resp: String| { + if let Some(server_v) = WebSyncClient::parse_push_conflict(&resp) { + if retries_left > 0 { + // Re-probe + retry once, still holding push_busy. + open_push_attempt( + sync.clone(), + base.clone(), + pair, + doc_json.clone(), + retries_left - 1, + ); + return; + } + if let Ok(mut s) = sync.try_borrow_mut() { + s.gate.note_conflict(server_v); + s.push_busy = false; + } + return; + } + if let Some(version) = WebSyncClient::parse_push_response(&resp) { + if let Ok(mut s) = sync.try_borrow_mut() { + s.client.mark_pushed(&doc_json, version); + s.gate.note_synced(pair.0, pair.1); + s.push_busy = false; + } + return; + } + // Unrecognized (network/parse failure): release the latch. + release_push_busy(&sync); + }) + }; + if !live_sync::post_json(&doc_url, &push_body, Some(on_resp)) { + release_push_busy(&sync); + } + }) + }; + if !live_sync::get(&version_url, on_version) { + release_push_busy(&sync); + } +} + +/// Acquire `push_busy` (waiting via self-reschedule), then push the snapshot. +fn drive_snapshot_push( + sync: SharedSync, + base: String, + request_id: String, + pair: (u64, u64), + doc_json: String, +) { + if !acquire_push_busy(&sync) { + schedule_once(PUSH_BUSY_RETRY_MS, move || { + drive_snapshot_push(sync, base, request_id, pair, doc_json) + }); + return; + } + snapshot_push_attempt(sync, base, request_id, pair, doc_json); +} + +/// Snapshot-channel push with `baseVersion = last_version()` (the established +/// baseline). Confirm → `mark_pushed` + `note_synced` + `snapshot-result`. +/// Conflict → `note_conflict` (observer will report `sync-conflict`) + +/// `snapshot-conflict` reply. +fn snapshot_push_attempt( + sync: SharedSync, + base: String, + request_id: String, + pair: (u64, u64), + doc_json: String, +) { + let base_version = sync + .try_borrow() + .map(|s| s.client.last_version()) + .unwrap_or(0); + let push_body = WebSyncClient::wrap_push_body_with_base(&doc_json, base_version); + let doc_url = format!("{base}/api/mcp/document"); + let on_resp: Rc = { + let sync = sync.clone(); + Rc::new(move |resp: String| { + if let Some(server_v) = WebSyncClient::parse_push_conflict(&resp) { + if let Ok(mut s) = sync.try_borrow_mut() { + s.gate.note_conflict(server_v); + s.push_busy = false; + } + post_to_parent(&event_snapshot_conflict(&request_id, server_v)); + return; + } + if let Some(version) = WebSyncClient::parse_push_response(&resp) { + if let Ok(mut s) = sync.try_borrow_mut() { + s.client.mark_pushed(&doc_json, version); + s.gate.note_synced(pair.0, pair.1); + s.push_busy = false; + } + post_to_parent(&event_snapshot_result( + &request_id, + &doc_json, + pair.0, + pair.1, + )); + return; + } + release_push_busy(&sync); + }) + }; + if !live_sync::post_json(&doc_url, &push_body, Some(on_resp)) { + release_push_busy(&sync); + } +} + +/// Acquire `push_busy` (waiting via self-reschedule), then run the use-local +/// re-push. +fn drive_use_local_push( + sync: SharedSync, + base: String, + request_id: String, + pair: (u64, u64), + doc_json: String, + base_version: u64, +) { + if !acquire_push_busy(&sync) { + schedule_once(PUSH_BUSY_RETRY_MS, move || { + drive_use_local_push(sync, base, request_id, pair, doc_json, base_version) + }); + return; + } + use_local_push_attempt( + sync, + base, + request_id, + pair, + doc_json, + base_version, + RETRY_ONCE, + ); +} + +/// Re-push the local document with `baseVersion = base_version` (the conflict's +/// server version). Confirm → `mark_pushed` + `note_synced` (clears conflict + +/// any pending open) + `conflict-resolved`. Second conflict → retry once with +/// the fresh server version (push_busy held); still failing → `note_conflict` +/// (keeps the gate's conflict version current for a later retry) + +/// `snapshot-conflict`. +fn use_local_push_attempt( + sync: SharedSync, + base: String, + request_id: String, + pair: (u64, u64), + doc_json: String, + base_version: u64, + retries_left: u8, +) { + let push_body = WebSyncClient::wrap_push_body_with_base(&doc_json, base_version); + let doc_url = format!("{base}/api/mcp/document"); + let on_resp: Rc = { + let sync = sync.clone(); + Rc::new(move |resp: String| { + if let Some(server_v) = WebSyncClient::parse_push_conflict(&resp) { + if retries_left > 0 { + use_local_push_attempt( + sync.clone(), + base.clone(), + request_id.clone(), + pair, + doc_json.clone(), + server_v, + retries_left - 1, + ); + return; + } + if let Ok(mut s) = sync.try_borrow_mut() { + s.gate.note_conflict(server_v); + s.push_busy = false; + } + post_to_parent(&event_snapshot_conflict(&request_id, server_v)); + return; + } + if let Some(version) = WebSyncClient::parse_push_response(&resp) { + if let Ok(mut s) = sync.try_borrow_mut() { + s.client.mark_pushed(&doc_json, version); + s.gate.note_synced(pair.0, pair.1); + s.push_busy = false; + } + post_to_parent(&event_conflict_resolved(&request_id)); + return; + } + release_push_busy(&sync); + }) + }; + if !live_sync::post_json(&doc_url, &push_body, Some(on_resp)) { + release_push_busy(&sync); + } +} + +// --------------------------------------------------------------------------- +// Outbound-event observer +// --------------------------------------------------------------------------- + +/// The SOLE emitter of `sync-conflict` / `opened` (drained from the gate's +/// consumable latches) and `dirty-changed` (on a `(generation, revision, +/// is_dirty)` triple change). If either RefCell is momentarily borrowed the +/// tick is skipped BEFORE the latches are drained, so no edge is lost. +fn observe_tick( + inner: &Rc>, + sync: &SharedSync, + last_triple: &LastTripleCell, +) { + // Read the triple first: if `inner` is busy, skip WITHOUT draining latches. + let Some(triple) = read_triple(inner) else { + return; + }; + // Drain the latches only once we can take the sync borrow; a failed borrow + // here also leaves the latches intact for the next tick. + let (opened, conflict) = { + let Ok(mut s) = sync.try_borrow_mut() else { + return; + }; + (s.gate.take_opened_edge(), s.gate.take_conflict_edge()) + }; + + if let Some(server_v) = conflict { + post_to_parent(&op_editor_core::bridge_protocol::event_sync_conflict( + triple.0, triple.1, server_v, + )); + } + if let Some(gen) = opened { + post_to_parent(&op_editor_core::bridge_protocol::event_opened(gen)); + } + let changed = *last_triple.borrow() != Some(triple); + if changed { + *last_triple.borrow_mut() = Some(triple); + post_to_parent(&op_editor_core::bridge_protocol::event_dirty_changed( + triple.0, triple.1, triple.2, + )); + } +} + +// --------------------------------------------------------------------------- +// Small helpers +// --------------------------------------------------------------------------- + +fn read_triple(inner: &Rc>) -> Option<(u64, u64, bool)> { + let b = inner.try_borrow().ok()?; + let s = b.host().editor_state(); + Some((s.document_generation(), s.document_revision(), s.is_dirty())) +} + +/// Serialize the live document + capture its `(generation, revision)` pair +/// atomically under a single borrow. +fn snapshot_state(inner: &Rc>) -> Option<((u64, u64), String)> { + let b = inner.try_borrow().ok()?; + let s = b.host().editor_state(); + let pair = (s.document_generation(), s.document_revision()); + let json = serde_json::to_string(&s.doc).ok()?; + Some((pair, json)) +} + +/// Try to claim the shared push single-flight latch. `true` when acquired. +fn acquire_push_busy(sync: &SharedSync) -> bool { + match sync.try_borrow_mut() { + Ok(mut s) if !s.push_busy => { + s.push_busy = true; + true + } + _ => false, + } +} + +fn release_push_busy(sync: &SharedSync) { + if let Ok(mut s) = sync.try_borrow_mut() { + s.push_busy = false; + } +} + +/// Post a codec string to the locked host origin (falling back to `*` only +/// before the origin is known — by which point every real reply is sent). +fn post_to_parent(json: &str) { + let Some(window) = web_sys::window() else { + return; + }; + let Some(parent) = window.parent().ok().flatten() else { + return; + }; + let target = BRIDGE_ORIGIN + .with(|o| o.borrow().clone()) + .unwrap_or_else(|| "*".to_string()); + let _ = parent.post_message(&JsValue::from_str(json), &target); +} + +/// One-shot `setTimeout`. `once_into_js` self-frees after firing. +fn schedule_once(delay_ms: i32, f: F) { + let Some(window) = web_sys::window() else { + return; + }; + let cb = Closure::once_into_js(f); + let _ = + window.set_timeout_with_callback_and_timeout_and_arguments_0(cb.unchecked_ref(), delay_ms); +} diff --git a/crates/op-host-web/src/web_ai_transport.rs b/crates/op-host-web/src/web_ai_transport.rs index c4830ceb0..730d8e0d5 100644 --- a/crates/op-host-web/src/web_ai_transport.rs +++ b/crates/op-host-web/src/web_ai_transport.rs @@ -91,7 +91,9 @@ pub fn post_ai_stream_to( on_event: Rc, ) -> Result { let xhr = web_sys::XmlHttpRequest::new()?; - xhr.open_with_async("POST", &format!("{base}{endpoint}"), true)?; + let url = format!("{base}{endpoint}"); + xhr.open_with_async("POST", &url, true)?; + crate::live_sync::attach_daemon_headers(&xhr, &url); xhr.set_request_header("Content-Type", "application/json")?; // `cursor` is the byte offset in `responseText` up to which we've already diff --git a/crates/op-host-web/src/web_model_catalog.rs b/crates/op-host-web/src/web_model_catalog.rs index a9dc32125..153c5c4d8 100644 --- a/crates/op-host-web/src/web_model_catalog.rs +++ b/crates/op-host-web/src/web_model_catalog.rs @@ -24,6 +24,7 @@ pub(crate) fn fetch_models(inner: &Rc>) if xhr.open_with_async("GET", &url, true).is_err() { return; } + crate::live_sync::attach_daemon_headers(&xhr, &url); let xhr_cb = xhr.clone(); let inner_cb = inner.clone(); let onloadend = Closure::::once_into_js(move || {