From 08a26548fedc2c567467d8a0610ee3e103af307f Mon Sep 17 00:00:00 2001 From: Kayshen-X Date: Fri, 7 Aug 2026 21:14:28 +0800 Subject: [PATCH] feat(web): daemon-side collaboration service behind /api/collab routes Give the serve-web daemon a real collaboration runtime so browser and VSCode-webview editors can drive public-relay sessions through REST, mirroring the web_auth proxy pattern: - WebCollabState + a dedicated driver thread (wake channel + 250/100ms tick) run CollabRuntime against the daemon document; documentRevision and collabSeq are separate so presence/UI changes never trigger a whole-document pull - versioned wire DTOs (CollabStateWire/CollabActionWire, wireVersion 1) map through explicit validation onto the internal UI types instead of serde on opaque internals; GET state / POST action / POST presence routes ride the existing /api/ auth + origin guards - document pushes during an Active session ingest through a split PreparedDocument::prepare (off-lock validation) + install_prepared_document (infallible, in-generation) inside a begin/finish_local_edit capture; identical pushes produce no txn - gate_daemon_mutation centralizes session-time refusals (409 collab-readonly/busy/active) across document push, sync-reset, open-recent, /mcp JSON-RPC, and AI apply paths, matching the desktop CollabGatePolicy semantics --- Cargo.lock | 1 + crates/op-editor-core/src/collab_routes.rs | 51 +++ crates/op-editor-core/src/collab_wire.rs | 257 ++++++++++++ .../op-editor-core/src/collab_wire/action.rs | 282 +++++++++++++ .../op-editor-core/src/collab_wire/parts.rs | 388 ++++++++++++++++++ .../op-editor-core/src/collab_wire/tests.rs | 220 ++++++++++ crates/op-editor-core/src/document_install.rs | 57 ++- .../src/document_install_tests.rs | 109 +++++ crates/op-editor-core/src/lib.rs | 4 +- crates/op-host-services/Cargo.toml | 4 + .../op-host-services/src/collab_blocking.rs | 31 ++ crates/op-host-services/src/lib.rs | 1 + .../op-host-services/src/web_canvas_server.rs | 131 +++++- .../src/web_canvas_server/collab_driver.rs | 250 +++++++++++ .../src/web_canvas_server/collab_routes.rs | 141 +++++++ .../web_canvas_server/collab_routes_tests.rs | 331 +++++++++++++++ .../src/web_canvas_server/collab_state.rs | 294 +++++++++++++ .../web_canvas_server/collab_state_tests.rs | 246 +++++++++++ .../src/web_canvas_server/connection.rs | 66 ++- .../src/web_canvas_server/doc_routes.rs | 17 + .../src/web_canvas_server/origin_guard.rs | 3 + .../src/web_canvas_server/run_loop.rs | 3 + .../src/web_canvas_server_conn_tests.rs | 46 ++- .../src/web_canvas_server_error.rs | 18 + .../src/web_canvas_server_export_tests.rs | 6 +- .../op-host-services/src/web_chat_standard.rs | 103 +++-- .../src/web_chat_standard_error.rs | 6 + .../src/web_chat_standard_tests.rs | 2 +- 28 files changed, 2986 insertions(+), 82 deletions(-) create mode 100644 crates/op-editor-core/src/collab_routes.rs create mode 100644 crates/op-editor-core/src/collab_wire.rs create mode 100644 crates/op-editor-core/src/collab_wire/action.rs create mode 100644 crates/op-editor-core/src/collab_wire/parts.rs create mode 100644 crates/op-editor-core/src/collab_wire/tests.rs create mode 100644 crates/op-host-services/src/collab_blocking.rs create mode 100644 crates/op-host-services/src/web_canvas_server/collab_driver.rs create mode 100644 crates/op-host-services/src/web_canvas_server/collab_routes.rs create mode 100644 crates/op-host-services/src/web_canvas_server/collab_routes_tests.rs create mode 100644 crates/op-host-services/src/web_canvas_server/collab_state.rs create mode 100644 crates/op-host-services/src/web_canvas_server/collab_state_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 859da8b8a..62ceae40c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4142,6 +4142,7 @@ dependencies = [ "op-ai", "op-ai-skills", "op-auth-bridge", + "op-collab-host", "op-config-store", "op-design-lint", "op-editor-core", diff --git a/crates/op-editor-core/src/collab_routes.rs b/crates/op-editor-core/src/collab_routes.rs new file mode 100644 index 000000000..3942e3376 --- /dev/null +++ b/crates/op-editor-core/src/collab_routes.rs @@ -0,0 +1,51 @@ +//! Collaboration HTTP route paths shared by the web shell (client) and the +//! serve-web daemon (server). +//! +//! The wasm bundle carries the collaboration *UI* but no transport: sessions +//! run inside the daemon, which is a native process and can hold the device +//! credentials a relay session needs. The browser drives it over these routes. +//! Keeping the paths in one wasm-clean crate both sides already depend on +//! means the client and server can never drift apart — the same reasoning as +//! [`auth_routes`](crate::auth_routes). + +/// Prefix shared by every JSON collaboration API route below. +/// +/// The daemon's sensitive-POST and CORS gates key off this prefix, so a route +/// added outside it would silently escape them. +pub const API_PREFIX: &str = "/api/collab/"; + +/// `GET` — the whole collaboration UI projection plus the two sequence +/// numbers a client polls on (`collabSeq`, `documentRevision`). +pub const STATE: &str = "/api/collab/state"; + +/// `POST` — enqueue one collaboration UI action. +/// +/// Accepted actions are a versioned wire enum, not the internal +/// `CollabUiAction`; see [`collab_wire`](crate::collab_wire). +pub const ACTION: &str = "/api/collab/action"; + +/// `POST` — publish the local cursor / viewport for peer presence. +pub const PRESENCE: &str = "/api/collab/presence"; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn api_routes_share_the_gating_prefix() { + for route in [STATE, ACTION, PRESENCE] { + assert!( + route.starts_with(API_PREFIX), + "{route} outside {API_PREFIX}" + ); + } + } + + #[test] + fn collaboration_routes_do_not_collide_with_the_auth_family() { + assert_ne!(API_PREFIX, crate::auth_routes::API_PREFIX); + for route in [STATE, ACTION, PRESENCE] { + assert!(!route.starts_with(crate::auth_routes::API_PREFIX)); + } + } +} diff --git a/crates/op-editor-core/src/collab_wire.rs b/crates/op-editor-core/src/collab_wire.rs new file mode 100644 index 000000000..d3e447012 --- /dev/null +++ b/crates/op-editor-core/src/collab_wire.rs @@ -0,0 +1,257 @@ +//! Versioned wire projection of the collaboration UI state. +//! +//! The daemon owns the collaboration runtime; the browser owns the panel that +//! paints it. This module is the contract between them, and it is deliberately +//! a **separate type family** from the internal `collab_*_ui` types rather than +//! a set of serde derives on them: +//! +//! * the internal types carry opaque tokens and redacting `Debug` impls whose +//! contracts a derive would quietly undo; +//! * the wire needs to stay stable across internal refactors, so it carries +//! [`COLLAB_WIRE_VERSION`] and its own enums; +//! * decoding must re-run every sanitising constructor, which only an explicit +//! mapping can guarantee. +//! +//! Two sequence numbers ride along and mean different things. `documentRevision` +//! changes only when document *content* changes, so it is what a client polls to +//! decide whether to refetch the document. `collabSeq` changes whenever this +//! projection changes — a peer joining, a cursor moving, a notice appearing — +//! and must never by itself trigger a document fetch. + +mod action; +mod parts; + +pub use action::{CollabActionWire, CollabActionWireError}; +pub use parts::{ + CollabAdmissionWire, CollabAvailabilityWire, CollabConnectErrorWire, CollabConnectionPathWire, + CollabDiscardedEditWire, CollabDiscoveredWire, CollabLocalPresenceWire, CollabNoticeKindWire, + CollabNoticeWire, CollabOwnerConfirmationWire, CollabPanelViewWire, CollabParticipantWire, + CollabPendingEditWire, CollabPhaseWire, CollabPointWire, CollabPresenceWire, + CollabRejectCodeWire, CollabRelayRegionWire, CollabRoleWire, +}; + +use serde::{Deserialize, Serialize}; + +use crate::{ + CollabAdmissionRequestKey, CollabInviteCode, CollabOwnerIdentityUi, CollabShareEndpoint, + CollabUiState, DiscoveredCollabEndpoint, +}; + +/// Bumped whenever an existing field changes meaning or disappears. +/// +/// Purely additive changes do not bump it: unknown fields are ignored on +/// decode, so an older client keeps working against a newer daemon. +pub const COLLAB_WIRE_VERSION: u32 = 1; + +/// The parts of the panel the daemon is authoritative for. +/// +/// Panel open/close, the current screen, hover, and the join-address draft are +/// deliberately absent: those are local UI, and echoing them back would make +/// every poll fight the user's own typing. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabPanelWire { + pub relay_region: CollabRelayRegionWire, + pub discovered: Vec, +} + +/// The authenticated session, once there is one. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabSessionWire { + pub session_name: String, + pub role: CollabRoleWire, + pub share_endpoint: Option, + pub invite: Option, + pub connection: Option, +} + +/// The whole collaboration projection for one poll. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabStateWire { + pub wire_version: u32, + /// Bumps on any change to this projection. + pub collab_seq: u64, + /// Bumps only on document-content change. + pub document_revision: u64, + pub availability: CollabAvailabilityWire, + pub phase: CollabPhaseWire, + pub panel: CollabPanelWire, + pub session: Option, + pub participants: Vec, + pub presence: Vec, + pub admissions: Vec, + pub owner_confirmation: Option, + pub notice: Option, + pub pending_edit: CollabPendingEditWire, + pub discarded_edit: Option, +} + +impl CollabStateWire { + /// Project the live UI state for transmission. + pub fn from_ui(ui: &CollabUiState, collab_seq: u64, document_revision: u64) -> Self { + let session = ui.authenticated_session().map(|session| { + let public = ui.public_session(); + CollabSessionWire { + session_name: session.session_name.clone(), + role: session.role.into(), + share_endpoint: session + .share_endpoint + .as_ref() + .map(|endpoint| endpoint.as_str().to_owned()), + invite: public + .and_then(|public| public.invite()) + .map(|invite| invite.as_str().to_owned()), + connection: public + .and_then(|public| public.connection()) + .map(Into::into), + } + }); + Self { + wire_version: COLLAB_WIRE_VERSION, + collab_seq, + document_revision, + availability: ui.availability.into(), + phase: ui.phase.into(), + panel: CollabPanelWire { + relay_region: ui.panel.relay_region.into(), + discovered: ui + .panel + .discovered + .iter() + .map(|found| CollabDiscoveredWire { + discovery_id: found.discovery_id.clone(), + endpoint: found.endpoint.clone(), + compatible: found.compatible, + }) + .collect(), + }, + session, + participants: ui.participants().iter().map(Into::into).collect(), + presence: ui.presence().iter().map(Into::into).collect(), + admissions: ui + .pending_admissions() + .iter() + .map(|pending| CollabAdmissionWire { + request_key: pending.request_key().as_str().to_owned(), + resume_role: pending.resume_role().map(Into::into), + }) + .collect(), + owner_confirmation: ui.pending_owner_confirmation().map(|pending| { + let identity = pending.identity(); + CollabOwnerConfirmationWire { + request_key: pending.request_key().as_str().to_owned(), + subject: identity.subject().to_owned(), + device_id: identity.device_id().to_owned(), + display_name: identity.claimed_display_name().map(str::to_owned), + avatar_url: identity.claimed_avatar_url().map(str::to_owned), + } + }), + notice: ui.notice.map(Into::into), + pending_edit: ui.pending_edit.into(), + discarded_edit: ui.discarded_edit.as_ref().map(Into::into), + } + } + + /// Install this projection into a client's UI state. + /// + /// Every write goes through a sanitising `set_*` / `publish_*` method, and + /// the call order is load-bearing because each of those fails closed on the + /// *current* phase: an authenticated session may only be installed on an + /// authenticated phase, an owner confirmation only while `Authenticating` + /// and before a session exists, and admissions only on an active owner. + /// Writing the phase first and the phase-gated payloads after is what keeps + /// a legitimate payload from being silently dropped. + pub fn apply_to(&self, ui: &mut CollabUiState, now_ms: u64) { + ui.set_availability(self.availability.into()); + + let participants: Vec<_> = self + .participants + .iter() + .map(CollabParticipantWire::to_ui) + .collect(); + match self.session.as_ref() { + Some(session) => { + let installed = ui.set_authenticated_session( + self.phase.into(), + crate::AuthenticatedCollabSession { + session_name: session.session_name.clone(), + role: session.role.into(), + share_endpoint: session + .share_endpoint + .as_deref() + .and_then(CollabShareEndpoint::new), + }, + participants, + ); + if installed { + if let Some(connection) = session.connection { + ui.set_public_session( + session.invite.as_deref().and_then(CollabInviteCode::new), + connection.into(), + ); + } + } + } + None => { + ui.clear_authenticated(); + ui.set_phase(self.phase.into()); + // The owner-confirmation prompt is the one payload that exists + // before a session does; it is gated on exactly this phase. + if let Some(confirmation) = self.owner_confirmation.as_ref() { + if let (Some(request_key), Some(identity)) = ( + CollabAdmissionRequestKey::new(confirmation.request_key.as_str()), + CollabOwnerIdentityUi::from_verified( + &confirmation.subject, + &confirmation.device_id, + confirmation.display_name.as_deref(), + confirmation.avatar_url.as_deref(), + ), + ) { + ui.publish_owner_confirmation(request_key, identity); + } + } + } + } + + ui.clear_pending_admissions(); + for admission in &self.admissions { + if let Some(request_key) = + CollabAdmissionRequestKey::new(admission.request_key.as_str()) + { + ui.publish_pending_admission(request_key, admission.resume_role.map(Into::into)); + } + } + + ui.queue_presence_snapshot( + self.presence + .iter() + .map(CollabPresenceWire::to_ui) + .collect(), + ); + ui.flush_presence(now_ms); + + ui.panel.relay_region = self.panel.relay_region.into(); + ui.panel.discovered = std::sync::Arc::new( + self.panel + .discovered + .iter() + .map(|found| DiscoveredCollabEndpoint { + discovery_id: found.discovery_id.clone(), + endpoint: found.endpoint.clone(), + compatible: found.compatible, + }) + .collect(), + ); + ui.pending_edit = self.pending_edit.into(); + ui.discarded_edit = self + .discarded_edit + .as_ref() + .map(CollabDiscardedEditWire::to_ui); + } +} + +#[cfg(test)] +#[path = "collab_wire/tests.rs"] +mod tests; diff --git a/crates/op-editor-core/src/collab_wire/action.rs b/crates/op-editor-core/src/collab_wire/action.rs new file mode 100644 index 000000000..fec6cfb44 --- /dev/null +++ b/crates/op-editor-core/src/collab_wire/action.rs @@ -0,0 +1,282 @@ +//! The versioned action enum a client may post to the daemon. +//! +//! This is a separate type from [`CollabUiAction`] on purpose. The internal +//! enum carries [`CollabAdmissionRequestKey`], whose whole contract is that it +//! is an opaque routing token built only through a validating constructor; +//! deriving `Serialize`/`Deserialize` on it would let any request body +//! materialise one. Here the key crosses the wire as a plain string and is +//! re-validated on the way in, so a malformed or oversized token is rejected +//! at the boundary instead of reaching the session actor. + +use serde::{Deserialize, Serialize}; + +use super::parts::CollabRelayRegionWire; +use crate::{CollabAdmissionRequestKey, CollabUiAction}; + +/// Longest LAN endpoint or discovery id the wire accepts. +/// +/// Both are consumed by the runtime's address parser; the cap keeps an +/// unbounded body from reaching it. +const MAX_WIRE_ADDRESS_CHARS: usize = 255; + +/// Why a posted action was refused before it reached the runtime. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CollabActionWireError { + /// The admission token is not a well-formed opaque key. + InvalidRequestKey, + /// A discovery id or endpoint was empty or over [`MAX_WIRE_ADDRESS_CHARS`]. + InvalidAddress, +} + +impl std::fmt::Display for CollabActionWireError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::InvalidRequestKey => f.write_str("malformed collaboration admission key"), + Self::InvalidAddress => f.write_str("malformed collaboration address"), + } + } +} + +impl std::error::Error for CollabActionWireError {} + +/// One collaboration action, as posted by a client. +/// +/// Externally tagged on `type` so an unknown action fails deserialization +/// loudly rather than silently decoding as a neighbouring variant. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum CollabActionWire { + OpenCreate, + Start, + StartLan, + #[serde(rename_all = "camelCase")] + SetRelayRegion { + region: CollabRelayRegionWire, + }, + OpenJoin, + BeginDiscovery, + #[serde(rename_all = "camelCase")] + JoinDiscovered { + discovery_id: String, + }, + #[serde(rename_all = "camelCase")] + JoinAddress { + endpoint: String, + }, + Cancel, + Retry, + Leave, + DiscardPending, + ReapplyDiscarded, + SaveAsFork, + #[serde(rename_all = "camelCase")] + ApproveAdmissionEditor { + request_key: String, + }, + #[serde(rename_all = "camelCase")] + ApproveAdmissionViewer { + request_key: String, + }, + #[serde(rename_all = "camelCase")] + RejectAdmission { + request_key: String, + }, + #[serde(rename_all = "camelCase")] + ConfirmOwnerIdentity { + request_key: String, + }, + #[serde(rename_all = "camelCase")] + RejectOwnerIdentity { + request_key: String, + }, +} + +impl CollabActionWire { + /// Whether this action opens a connection to something the caller named. + /// + /// A public deployment must refuse these: `JoinAddress` resolves an + /// arbitrary socket address, and the LAN paths expose the host's local + /// network. The local and managed daemons allow them — that is desktop + /// parity — so this is a classifier, not a policy. The public-mode route + /// table is what consults it. + pub const fn reaches_caller_named_network(&self) -> bool { + matches!( + self, + Self::StartLan + | Self::BeginDiscovery + | Self::JoinDiscovered { .. } + | Self::JoinAddress { .. } + ) + } + + /// Validate and convert into the internal action. + pub fn into_ui_action(self) -> Result { + Ok(match self { + Self::OpenCreate => CollabUiAction::OpenCreate, + Self::Start => CollabUiAction::Start, + Self::StartLan => CollabUiAction::StartLan, + Self::SetRelayRegion { region } => CollabUiAction::SetRelayRegion { + region: region.into(), + }, + Self::OpenJoin => CollabUiAction::OpenJoin, + Self::BeginDiscovery => CollabUiAction::BeginDiscovery, + Self::JoinDiscovered { discovery_id } => CollabUiAction::JoinDiscovered { + discovery_id: checked_address(discovery_id)?, + }, + Self::JoinAddress { endpoint } => CollabUiAction::JoinAddress { + endpoint: checked_address(endpoint)?, + }, + Self::Cancel => CollabUiAction::Cancel, + Self::Retry => CollabUiAction::Retry, + Self::Leave => CollabUiAction::Leave, + Self::DiscardPending => CollabUiAction::DiscardPending, + Self::ReapplyDiscarded => CollabUiAction::ReapplyDiscarded, + Self::SaveAsFork => CollabUiAction::SaveAsFork, + Self::ApproveAdmissionEditor { request_key } => { + CollabUiAction::ApproveAdmissionEditor { + request_key: checked_request_key(&request_key)?, + } + } + Self::ApproveAdmissionViewer { request_key } => { + CollabUiAction::ApproveAdmissionViewer { + request_key: checked_request_key(&request_key)?, + } + } + Self::RejectAdmission { request_key } => CollabUiAction::RejectAdmission { + request_key: checked_request_key(&request_key)?, + }, + Self::ConfirmOwnerIdentity { request_key } => CollabUiAction::ConfirmOwnerIdentity { + request_key: checked_request_key(&request_key)?, + }, + Self::RejectOwnerIdentity { request_key } => CollabUiAction::RejectOwnerIdentity { + request_key: checked_request_key(&request_key)?, + }, + }) + } +} + +fn checked_request_key(value: &str) -> Result { + CollabAdmissionRequestKey::new(value).ok_or(CollabActionWireError::InvalidRequestKey) +} + +fn checked_address(value: String) -> Result { + let trimmed = value.trim(); + if trimmed.is_empty() || trimmed.chars().count() > MAX_WIRE_ADDRESS_CHARS { + return Err(CollabActionWireError::InvalidAddress); + } + Ok(trimmed.to_owned()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_internal_action_has_a_wire_variant() { + // Exhaustive by construction: adding a `CollabUiAction` variant makes + // this match fail to compile, which is the reminder to extend the wire. + let sample = CollabUiAction::OpenCreate; + let covered = match sample { + CollabUiAction::OpenCreate + | CollabUiAction::Start + | CollabUiAction::StartLan + | CollabUiAction::SetRelayRegion { .. } + | CollabUiAction::OpenJoin + | CollabUiAction::BeginDiscovery + | CollabUiAction::JoinDiscovered { .. } + | CollabUiAction::JoinAddress { .. } + | CollabUiAction::Cancel + | CollabUiAction::Retry + | CollabUiAction::Leave + | CollabUiAction::DiscardPending + | CollabUiAction::ReapplyDiscarded + | CollabUiAction::SaveAsFork + | CollabUiAction::ApproveAdmissionEditor { .. } + | CollabUiAction::ApproveAdmissionViewer { .. } + | CollabUiAction::RejectAdmission { .. } + | CollabUiAction::ConfirmOwnerIdentity { .. } + | CollabUiAction::RejectOwnerIdentity { .. } => 19, + }; + assert_eq!(covered, 19); + } + + #[test] + fn parameterless_actions_round_trip_through_json() { + let json = r#"{"type":"start"}"#; + let wire: CollabActionWire = serde_json::from_str(json).expect("decodes"); + assert_eq!(wire, CollabActionWire::Start); + assert_eq!(serde_json::to_string(&wire).expect("encodes"), json); + assert_eq!(wire.into_ui_action().expect("valid"), CollabUiAction::Start); + } + + #[test] + fn admission_keys_are_revalidated_not_trusted() { + let good = CollabActionWire::RejectAdmission { + request_key: "abc-123_XYZ".into(), + }; + assert!(good.into_ui_action().is_ok()); + + for bad in ["", "has space", &"x".repeat(97), "semi;colon"] { + let wire = CollabActionWire::ApproveAdmissionEditor { + request_key: bad.to_string(), + }; + assert_eq!( + wire.into_ui_action(), + Err(CollabActionWireError::InvalidRequestKey), + "{bad:?} must not become a request key" + ); + } + } + + #[test] + fn addresses_are_trimmed_and_bounded() { + let wire = CollabActionWire::JoinAddress { + endpoint: " 192.168.1.10:4321 ".into(), + }; + assert_eq!( + wire.into_ui_action().expect("valid"), + CollabUiAction::JoinAddress { + endpoint: "192.168.1.10:4321".into() + } + ); + + for bad in ["", " ", &"a".repeat(MAX_WIRE_ADDRESS_CHARS + 1)] { + let wire = CollabActionWire::JoinAddress { + endpoint: bad.to_string(), + }; + assert_eq!( + wire.into_ui_action(), + Err(CollabActionWireError::InvalidAddress) + ); + } + } + + #[test] + fn unknown_action_types_are_rejected_outright() { + let err = serde_json::from_str::(r#"{"type":"selfDestruct"}"#); + assert!(err.is_err()); + } + + #[test] + fn caller_named_network_actions_are_classified() { + for action in [ + CollabActionWire::StartLan, + CollabActionWire::BeginDiscovery, + CollabActionWire::JoinDiscovered { + discovery_id: "d".into(), + }, + CollabActionWire::JoinAddress { + endpoint: "1.2.3.4:5".into(), + }, + ] { + assert!(action.reaches_caller_named_network(), "{action:?}"); + } + for action in [ + CollabActionWire::Start, + CollabActionWire::Leave, + CollabActionWire::SaveAsFork, + ] { + assert!(!action.reaches_caller_named_network(), "{action:?}"); + } + } +} diff --git a/crates/op-editor-core/src/collab_wire/parts.rs b/crates/op-editor-core/src/collab_wire/parts.rs new file mode 100644 index 000000000..43597e696 --- /dev/null +++ b/crates/op-editor-core/src/collab_wire/parts.rs @@ -0,0 +1,388 @@ +//! Leaf wire types for the collaboration REST projection. +//! +//! Every enum here is a *versioned copy* of an internal collaboration UI enum, +//! not a serde derive on the internal type. The duplication is the point: the +//! internal enums are free to grow variants or reorder without silently +//! changing the wire, and an unknown wire value decodes to an explicit +//! fallback instead of failing the whole payload. + +use serde::{Deserialize, Serialize}; + +use crate::{ + CollabAvailability, CollabConnectErrorUi, CollabConnectionPathUi, CollabConnectionPhase, + CollabDiscardedEditUi, CollabNotice, CollabNoticeKind, CollabPanelView, CollabParticipantUi, + CollabPendingEditUi, CollabRejectUiCode, CollabRelayRegion, CollabUiRole, RemotePresenceUi, +}; + +macro_rules! wire_enum { + ( + $(#[$meta:meta])* + $name:ident <=> $internal:path { + $($wire:ident <=> $variant:ident),+ $(,)? + } + ) => { + $(#[$meta])* + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] + #[serde(rename_all = "camelCase")] + pub enum $name { + $($wire),+ + } + + impl From<$internal> for $name { + fn from(value: $internal) -> Self { + use $internal as Internal; + match value { + $(Internal::$variant => Self::$wire),+ + } + } + } + + impl From<$name> for $internal { + fn from(value: $name) -> Self { + use $internal as Internal; + match value { + $($name::$wire => Internal::$variant),+ + } + } + } + }; +} + +wire_enum! { + /// Whether collaboration can be started at all on this deployment. + CollabAvailabilityWire <=> CollabAvailability { + Unavailable <=> Unavailable, + SignInRequired <=> SignInRequired, + Ready <=> Ready, + } +} + +wire_enum! { + /// Where the local peer sits in the session lifecycle. + CollabPhaseWire <=> CollabConnectionPhase { + Idle <=> Idle, + Starting <=> Starting, + Discovering <=> Discovering, + Joining <=> Joining, + Authenticating <=> Authenticating, + Active <=> Active, + Reconnecting <=> Reconnecting, + ReadOnly <=> ReadOnly, + Ended <=> Ended, + } +} + +wire_enum! { + /// Authority the local peer holds in the session. + CollabRoleWire <=> CollabUiRole { + Owner <=> Owner, + Editor <=> Editor, + Viewer <=> Viewer, + } +} + +wire_enum! { + /// Which collaboration panel screen the client should show. + CollabPanelViewWire <=> CollabPanelView { + Home <=> Home, + Create <=> Create, + Join <=> Join, + Session <=> Session, + } +} + +wire_enum! { + /// Public-relay service region. + CollabRelayRegionWire <=> CollabRelayRegion { + China <=> China, + Global <=> Global, + } +} + +wire_enum! { + /// State of a local edit that has not yet been accepted by the owner. + CollabPendingEditWire <=> CollabPendingEditUi { + None <=> None, + Submitting <=> Submitting, + Replaying <=> Replaying, + Conflict <=> Conflict, + } +} + +wire_enum! { + /// Why the owner refused a local edit. + CollabRejectCodeWire <=> CollabRejectUiCode { + StaleBase <=> StaleBase, + ReadOnly <=> ReadOnly, + Unsupported <=> Unsupported, + Conflict <=> Conflict, + ResourceLimit <=> ResourceLimit, + Authentication <=> Authentication, + Unknown <=> Unknown, + } +} + +wire_enum! { + /// Why a connection attempt failed. + CollabConnectErrorWire <=> CollabConnectErrorUi { + InviteUnavailable <=> InviteUnavailable, + InviteInvalid <=> InviteInvalid, + InviteExpired <=> InviteExpired, + RelayUnavailable <=> RelayUnavailable, + RelayNotConfigured <=> RelayNotConfigured, + RegionUnavailable <=> RegionUnavailable, + RateLimited <=> RateLimited, + Incompatible <=> Incompatible, + SecureKeyUnavailable <=> SecureKeyUnavailable, + OwnerNotConfirmed <=> OwnerNotConfirmed, + } +} + +/// How the session reaches its peers. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "camelCase")] +pub enum CollabConnectionPathWire { + Lan, + #[serde(rename_all = "camelCase")] + Relay { + home_region: CollabRelayRegionWire, + }, +} + +impl From for CollabConnectionPathWire { + fn from(value: CollabConnectionPathUi) -> Self { + match value { + CollabConnectionPathUi::Lan => Self::Lan, + CollabConnectionPathUi::Relay { home_region } => Self::Relay { + home_region: home_region.into(), + }, + } + } +} + +impl From for CollabConnectionPathUi { + fn from(value: CollabConnectionPathWire) -> Self { + match value { + CollabConnectionPathWire::Lan => Self::Lan, + CollabConnectionPathWire::Relay { home_region } => Self::Relay { + home_region: home_region.into(), + }, + } + } +} + +/// A transient banner the panel shows. +/// +/// `UnsupportedEdit` deliberately collapses its 17-variant feature payload to +/// the i18n key: the client renders that key and nothing else, so shipping the +/// whole enum would only create a second place to keep in sync. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "camelCase")] +pub enum CollabNoticeKindWire { + Connect { + error: CollabConnectErrorWire, + }, + DisconnectedReadOnly, + TicketExpired, + OwnerLeft, + EpochChanged, + Reject { + code: CollabRejectCodeWire, + }, + EditConflictDiscarded, + UndoConflict, + #[serde(rename_all = "camelCase")] + UnsupportedEdit { + i18n_key: String, + }, +} + +impl From for CollabNoticeKindWire { + fn from(value: CollabNoticeKind) -> Self { + match value { + CollabNoticeKind::Connect(error) => Self::Connect { + error: error.into(), + }, + CollabNoticeKind::DisconnectedReadOnly => Self::DisconnectedReadOnly, + CollabNoticeKind::TicketExpired => Self::TicketExpired, + CollabNoticeKind::OwnerLeft => Self::OwnerLeft, + CollabNoticeKind::EpochChanged => Self::EpochChanged, + CollabNoticeKind::Reject(code) => Self::Reject { code: code.into() }, + CollabNoticeKind::EditConflictDiscarded => Self::EditConflictDiscarded, + CollabNoticeKind::UndoConflict => Self::UndoConflict, + CollabNoticeKind::UnsupportedEdit(feature) => Self::UnsupportedEdit { + i18n_key: feature.i18n_key().to_owned(), + }, + } + } +} + +/// One banner plus the clock reading that dates it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabNoticeWire { + #[serde(flatten)] + pub kind: CollabNoticeKindWire, + pub created_at_ms: u64, +} + +impl From for CollabNoticeWire { + fn from(value: CollabNotice) -> Self { + Self { + kind: value.kind.into(), + created_at_ms: value.created_at_ms, + } + } +} + +/// A peer in the session roster. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabParticipantWire { + pub participant_key: String, + pub display_name: String, + pub color_rgba: u32, + pub role: CollabRoleWire, + pub is_self: bool, +} + +impl From<&CollabParticipantUi> for CollabParticipantWire { + fn from(value: &CollabParticipantUi) -> Self { + Self { + participant_key: value.participant_key.clone(), + display_name: value.display_name.clone(), + color_rgba: value.color_rgba, + role: value.role.into(), + is_self: value.is_self, + } + } +} + +impl CollabParticipantWire { + /// Rebuild the internal type. Initials are deliberately not carried on the + /// wire — the constructor rederives them, so a client cannot inject a + /// display string that disagrees with the name. + pub fn to_ui(&self) -> CollabParticipantUi { + CollabParticipantUi::new( + self.participant_key.clone(), + self.display_name.clone(), + self.color_rgba, + self.role.into(), + self.is_self, + ) + } +} + +/// A canvas-space point. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +pub struct CollabPointWire { + pub x: f64, + pub y: f64, +} + +/// One peer's cursor / selection. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabPresenceWire { + pub participant_key: String, + pub cursor: Option, + pub selection: Vec, + pub editing_node: Option, + pub updated_at_ms: u64, +} + +impl From<&RemotePresenceUi> for CollabPresenceWire { + fn from(value: &RemotePresenceUi) -> Self { + Self { + participant_key: value.participant_key.clone(), + cursor: value.cursor.map(|point| CollabPointWire { + x: point.x, + y: point.y, + }), + selection: value.selection.as_ref().clone(), + editing_node: value.editing_node.clone(), + updated_at_ms: value.updated_at_ms, + } + } +} + +impl CollabPresenceWire { + /// Rebuild the internal type through the bounding constructor, which caps + /// selection length and trims oversized ids. + pub fn to_ui(&self) -> RemotePresenceUi { + RemotePresenceUi::bounded( + self.participant_key.clone(), + self.cursor.map(|point| crate::CollabCanvasPoint { + x: point.x, + y: point.y, + }), + self.selection.iter().cloned(), + self.editing_node.clone(), + self.updated_at_ms, + ) + } +} + +/// The local peer's own cursor / selection, pushed by the browser. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabLocalPresenceWire { + pub cursor: Option, + /// Reserved for the multi-account server: identifies which browser + /// connection produced this presence. Ignored by the single-tenant daemon. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub client_id: Option, +} + +/// A guest waiting for the owner's admission decision. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabAdmissionWire { + /// Opaque routing token. It is a random per-request identifier, never a + /// peer identity, and the client may only echo it back verbatim. + pub request_key: String, + pub resume_role: Option, +} + +/// The verified owner identity a guest must confirm before anything is applied. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabOwnerConfirmationWire { + pub request_key: String, + pub subject: String, + pub device_id: String, + pub display_name: Option, + pub avatar_url: Option, +} + +/// The local edit a conflict discarded, offered for replay. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabDiscardedEditWire { + pub node_label: String, + pub fields: Vec, +} + +impl From<&CollabDiscardedEditUi> for CollabDiscardedEditWire { + fn from(value: &CollabDiscardedEditUi) -> Self { + Self { + node_label: value.node_label.clone(), + fields: value.fields.clone(), + } + } +} + +impl CollabDiscardedEditWire { + pub fn to_ui(&self) -> CollabDiscardedEditUi { + CollabDiscardedEditUi::bounded(self.node_label.clone(), self.fields.iter().cloned()) + } +} + +/// One endpoint found by LAN discovery. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CollabDiscoveredWire { + pub discovery_id: String, + pub endpoint: String, + pub compatible: bool, +} diff --git a/crates/op-editor-core/src/collab_wire/tests.rs b/crates/op-editor-core/src/collab_wire/tests.rs new file mode 100644 index 000000000..6ce962334 --- /dev/null +++ b/crates/op-editor-core/src/collab_wire/tests.rs @@ -0,0 +1,220 @@ +use super::*; +use crate::{ + CollabAvailability, CollabConnectionPathUi, CollabConnectionPhase, CollabNoticeKind, + CollabParticipantUi, CollabRejectUiCode, CollabRelayRegion, CollabUiRole, RemotePresenceUi, +}; + +fn owner_session_state() -> CollabUiState { + let mut ui = CollabUiState::default(); + ui.set_availability(CollabAvailability::Ready); + ui.set_authenticated_session( + CollabConnectionPhase::Active, + crate::AuthenticatedCollabSession { + session_name: "studio".into(), + role: CollabUiRole::Owner, + share_endpoint: CollabShareEndpoint::new("192.168.1.7:43120"), + }, + vec![CollabParticipantUi::new( + "peer-1", + "Ada Lovelace", + 0x33aaffff, + CollabUiRole::Owner, + true, + )], + ); + ui.set_public_session( + CollabInviteCode::new("ABCDEFGHJK"), + CollabConnectionPathUi::Relay { + home_region: CollabRelayRegion::Global, + }, + ); + ui +} + +#[test] +fn owner_projection_round_trips_through_json_into_an_equal_projection() { + let ui = owner_session_state(); + let wire = CollabStateWire::from_ui(&ui, 7, 42); + + let json = serde_json::to_string(&wire).expect("encodes"); + let decoded: CollabStateWire = serde_json::from_str(&json).expect("decodes"); + assert_eq!(decoded, wire); + + let mut client = CollabUiState::default(); + decoded.apply_to(&mut client, 1_000); + let reprojected = CollabStateWire::from_ui(&client, 7, 42); + assert_eq!(reprojected, wire); +} + +#[test] +fn projection_carries_the_wire_version_and_both_sequence_numbers() { + let wire = CollabStateWire::from_ui(&owner_session_state(), 9, 300); + assert_eq!(wire.wire_version, COLLAB_WIRE_VERSION); + assert_eq!(wire.collab_seq, 9); + assert_eq!(wire.document_revision, 300); +} + +#[test] +fn unknown_fields_are_ignored_so_an_older_client_survives_an_added_field() { + let wire = CollabStateWire::from_ui(&owner_session_state(), 1, 1); + let mut value = serde_json::to_value(&wire).expect("encodes"); + value + .as_object_mut() + .expect("object") + .insert("somethingAddedLater".into(), serde_json::json!({"a": 1})); + let decoded: CollabStateWire = serde_json::from_value(value).expect("tolerates new fields"); + assert_eq!(decoded, wire); +} + +#[test] +fn an_idle_projection_carries_no_session_participants_or_presence() { + let mut ui = CollabUiState::default(); + ui.set_availability(CollabAvailability::SignInRequired); + let wire = CollabStateWire::from_ui(&ui, 0, 0); + + assert_eq!(wire.phase, CollabPhaseWire::Idle); + assert_eq!(wire.availability, CollabAvailabilityWire::SignInRequired); + assert!(wire.session.is_none()); + assert!(wire.participants.is_empty()); + assert!(wire.presence.is_empty()); + assert!(wire.admissions.is_empty()); +} + +#[test] +fn applying_a_session_projection_installs_role_invite_and_share_endpoint() { + let wire = CollabStateWire::from_ui(&owner_session_state(), 3, 3); + let mut client = CollabUiState::default(); + wire.apply_to(&mut client, 500); + + let session = client + .authenticated_session() + .expect("session installed on an authenticated phase"); + assert_eq!(session.session_name, "studio"); + assert_eq!(session.role, CollabUiRole::Owner); + assert_eq!( + session + .share_endpoint + .as_ref() + .map(CollabShareEndpoint::as_str), + Some("192.168.1.7:43120") + ); + let public = client.public_session().expect("public session installed"); + assert_eq!( + public.invite().map(CollabInviteCode::as_str), + Some("ABCDEFGHJK") + ); + assert_eq!( + public.connection(), + Some(CollabConnectionPathUi::Relay { + home_region: CollabRelayRegion::Global + }) + ); +} + +#[test] +fn presence_is_installed_only_for_peers_in_the_roster() { + let mut ui = owner_session_state(); + ui.queue_presence_snapshot(vec![ + RemotePresenceUi::bounded( + "peer-1", + Some(crate::CollabCanvasPoint { x: 12.0, y: 34.0 }), + ["node-a".to_string()], + None, + 10, + ), + RemotePresenceUi::bounded("ghost", None, [], None, 11), + ]); + ui.flush_presence(100); + + let wire = CollabStateWire::from_ui(&ui, 1, 1); + let keys: Vec<_> = wire + .presence + .iter() + .map(|p| p.participant_key.as_str()) + .collect(); + assert_eq!(keys, ["peer-1"], "a non-roster peer must not be projected"); + + let mut client = CollabUiState::default(); + wire.apply_to(&mut client, 1_000); + assert_eq!(client.presence().len(), 1); + assert_eq!(client.presence()[0].participant_key, "peer-1"); + assert_eq!( + client.presence()[0].cursor, + Some(crate::CollabCanvasPoint { x: 12.0, y: 34.0 }) + ); +} + +#[test] +fn admission_keys_survive_the_round_trip_as_opaque_strings() { + let mut ui = owner_session_state(); + let key = CollabAdmissionRequestKey::new("req-abc_123").expect("valid key"); + assert!(ui.publish_pending_admission(key.clone(), Some(CollabUiRole::Editor))); + + let wire = CollabStateWire::from_ui(&ui, 1, 1); + assert_eq!(wire.admissions.len(), 1); + assert_eq!(wire.admissions[0].request_key, "req-abc_123"); + + let mut client = CollabUiState::default(); + wire.apply_to(&mut client, 1_000); + assert_eq!(client.pending_admissions().len(), 1); + assert_eq!(client.pending_admissions()[0].request_key(), &key); +} + +#[test] +fn a_malformed_admission_key_is_dropped_rather_than_installed() { + let mut wire = CollabStateWire::from_ui(&owner_session_state(), 1, 1); + wire.admissions.push(CollabAdmissionWire { + request_key: "not a key".into(), + resume_role: None, + }); + + let mut client = CollabUiState::default(); + wire.apply_to(&mut client, 1_000); + assert!(client.pending_admissions().is_empty()); +} + +#[test] +fn notices_project_their_payload_and_timestamp() { + let mut ui = owner_session_state(); + ui.set_notice( + CollabNoticeKind::Reject(CollabRejectUiCode::StaleBase), + 4_242, + ); + let wire = CollabStateWire::from_ui(&ui, 1, 1); + let notice = wire.notice.clone().expect("notice projected"); + assert_eq!(notice.created_at_ms, 4_242); + assert_eq!( + notice.kind, + CollabNoticeKindWire::Reject { + code: CollabRejectCodeWire::StaleBase + } + ); +} + +#[test] +fn the_panel_projection_omits_local_only_ui() { + let wire = CollabStateWire::from_ui(&owner_session_state(), 1, 1); + let json = serde_json::to_value(&wire.panel).expect("encodes"); + let panel = json.as_object().expect("object"); + // Panel open/view/hover and the join-address draft belong to whichever + // client is painting; echoing them would fight the user's own typing. + for local_only in ["open", "view", "hover", "joinAddress", "joinInput"] { + assert!( + !panel.contains_key(local_only), + "{local_only} must stay client-local" + ); + } + assert!(panel.contains_key("relayRegion")); + assert!(panel.contains_key("discovered")); +} + +#[test] +fn local_presence_wire_defaults_its_reserved_client_id() { + let decoded: CollabLocalPresenceWire = + serde_json::from_str(r#"{"cursor":{"x":1.5,"y":2.5}}"#).expect("decodes"); + assert_eq!(decoded.client_id, None); + assert_eq!(decoded.cursor, Some(CollabPointWire { x: 1.5, y: 2.5 })); + // Absent rather than null, so the field can gain meaning without a bump. + let json = serde_json::to_string(&decoded).expect("encodes"); + assert!(!json.contains("clientId"), "{json}"); +} diff --git a/crates/op-editor-core/src/document_install.rs b/crates/op-editor-core/src/document_install.rs index 166c3b5c3..dd4539562 100644 --- a/crates/op-editor-core/src/document_install.rs +++ b/crates/op-editor-core/src/document_install.rs @@ -68,6 +68,37 @@ struct PreparedInstall { variables: VariableUiState, } +/// A document that has cleared every structural check an install performs. +/// +/// Validation walks the whole node forest, so a server holding one shared +/// editor behind a mutex must not pay for it under the lock. Splitting the +/// fallible half out lets a caller validate off-lock and then install +/// infallibly, which is what makes an in-generation ingest atomic with +/// respect to a concurrently running collaboration session. +#[derive(Debug, Clone)] +pub struct PreparedDocument { + doc: PenDocument, +} + +impl PreparedDocument { + /// Run every structural check an install would run. + pub fn prepare(doc: PenDocument) -> Result { + validate_install_document(&doc)?; + Ok(Self { doc }) + } + + /// The validated document, for a caller that needs to compare before + /// deciding to install. + pub fn document(&self) -> &PenDocument { + &self.doc + } + + /// Give the validated document back, abandoning the install. + pub fn into_document(self) -> PenDocument { + self.doc + } +} + impl EditorState { /// Atomically install a collaboration document after neutral validation. /// @@ -81,7 +112,27 @@ impl EditorState { doc: PenDocument, origin: EditOrigin, ) -> Result { - validate_install_document(&doc)?; + Ok(self.install_prepared_document(PreparedDocument::prepare(doc)?, origin)) + } + + /// Install an already-validated document. + /// + /// Infallible by construction — [`PreparedDocument::prepare`] has already + /// done the only fallible work — so this half can run under a lock without + /// an error path that would leave a half-applied document behind. + /// + /// Every origin except `Snapshot` keeps the current + /// [`document_generation`](EditorState::document_generation). That is what + /// makes this usable as the in-generation content swap inside an open + /// [`begin_local_edit`](EditorState::begin_local_edit) capture: a bumped + /// generation would make the matching `end_local_edit` fail, and the + /// collaboration runtime reads that failure as a lost session. + pub fn install_prepared_document( + &mut self, + prepared: PreparedDocument, + origin: EditOrigin, + ) -> DocumentInstallReport { + let doc = prepared.doc; let prepared = self.prepare_document_install(&doc); let local_snapshot = (origin == EditOrigin::Local).then(|| self.snapshot_for_history()); @@ -114,11 +165,11 @@ impl EditorState { jian_ops_schema::image_thumbs::activate_for_document(&self.doc); crate::state::drop_document_after_replace(old_doc); - Ok(DocumentInstallReport { + DocumentInstallReport { origin, retained_selection: self.selection.len(), active_page_changed: prepared.active_page_changed, - }) + } } fn prepare_document_install(&self, doc: &PenDocument) -> PreparedInstall { diff --git a/crates/op-editor-core/src/document_install_tests.rs b/crates/op-editor-core/src/document_install_tests.rs index af2374f09..15a2a0795 100644 --- a/crates/op-editor-core/src/document_install_tests.rs +++ b/crates/op-editor-core/src/document_install_tests.rs @@ -236,3 +236,112 @@ fn explicit_pages_reject_root_content_and_empty_page_ids() { Err(DocumentInstallError::EmptyPageId { page_index: 0 }) ); } + +// --------------------------------------------------------------------------- +// In-generation ingest — the seam a daemon uses to push a whole document into +// an open collaboration edit capture. +// --------------------------------------------------------------------------- + +fn doc_with_rect(id: &str, width: f64) -> PenDocument { + serde_json::from_value(serde_json::json!({ + "version": "1.0", + "children": [{ + "type": "rectangle", + "id": id, + "name": "Rect", + "x": 0, + "y": 0, + "width": width, + "height": 10 + }] + })) + .expect("valid test document") +} + +#[test] +fn preparing_a_document_runs_validation_before_any_state_is_touched() { + let mut invalid = doc_with_rect("dupe", 10.0); + invalid.children.push(invalid.children[0].clone()); + assert_eq!( + crate::PreparedDocument::prepare(invalid).err(), + Some(DocumentInstallError::DuplicateId { id: "dupe".into() }) + ); + assert!(crate::PreparedDocument::prepare(doc_with_rect("ok", 10.0)).is_ok()); +} + +#[test] +fn an_in_generation_install_keeps_the_generation_so_the_capture_can_close() { + let mut state = EditorState::starter(); + let generation = state.document_generation(); + + let capture = state.begin_local_edit(); + let prepared = crate::PreparedDocument::prepare(doc_with_rect("n1", 10.0)).expect("valid"); + state.install_prepared_document(prepared, EditOrigin::Local); + + assert_eq!( + state.document_generation(), + generation, + "a bumped generation would make end_local_edit fail and the runtime \ + would read that as a lost session" + ); + let outcome = state.end_local_edit(capture).expect("capture closes"); + assert!(matches!( + outcome, + crate::edit_transaction::LocalEditOutcome::Changed(_) + )); +} + +#[test] +fn pushing_an_identical_document_produces_no_change_so_no_transaction_is_sent() { + let mut state = EditorState::starter(); + let prepared = crate::PreparedDocument::prepare(doc_with_rect("n1", 10.0)).expect("valid"); + state.install_prepared_document(prepared, EditOrigin::Local); + let revision = state.document_revision(); + + // The daemon re-pushes what the browser already has — the common case for a + // periodic whole-document sync. + let capture = state.begin_local_edit(); + let same = crate::PreparedDocument::prepare(doc_with_rect("n1", 10.0)).expect("valid"); + state.install_prepared_document(same, EditOrigin::Local); + let outcome = state.end_local_edit(capture).expect("capture closes"); + + assert!( + matches!(outcome, crate::edit_transaction::LocalEditOutcome::NoChange), + "an unchanged push must diff to nothing, or every idle sync would \ + broadcast a transaction to every peer" + ); + assert_eq!( + state.document_revision(), + revision, + "a no-change push must not move the revision either" + ); +} + +#[test] +fn a_changed_push_inside_a_capture_diffs_to_exactly_the_new_document() { + let mut state = EditorState::starter(); + state.install_prepared_document( + crate::PreparedDocument::prepare(doc_with_rect("n1", 10.0)).expect("valid"), + EditOrigin::Local, + ); + + let capture = state.begin_local_edit(); + state.install_prepared_document( + crate::PreparedDocument::prepare(doc_with_rect("n1", 25.0)).expect("valid"), + EditOrigin::Local, + ); + let outcome = state.end_local_edit(capture).expect("capture closes"); + + match outcome { + crate::edit_transaction::LocalEditOutcome::Changed(completed) => { + assert_eq!(state.doc, doc_with_rect("n1", 25.0)); + // Accepting the edit is the caller's job; rolling it back must + // restore the pre-push document exactly. + state.rollback_local_edit(completed).expect("rolls back"); + assert_eq!(state.doc, doc_with_rect("n1", 10.0)); + } + crate::edit_transaction::LocalEditOutcome::NoChange => { + panic!("a widened rect must diff as a change") + } + } +} diff --git a/crates/op-editor-core/src/lib.rs b/crates/op-editor-core/src/lib.rs index 6df175fd8..7bbb5d722 100644 --- a/crates/op-editor-core/src/lib.rs +++ b/crates/op-editor-core/src/lib.rs @@ -39,8 +39,10 @@ pub mod collab_notice_ui; pub mod collab_owner_confirm_ui; pub mod collab_panel_hover; pub mod collab_public_ui; +pub mod collab_routes; mod collab_ui_debug; pub mod collab_ui_state; +pub mod collab_wire; pub mod color_picker; pub mod color_picker_edit; mod color_picker_snapshot; @@ -316,7 +318,7 @@ pub use components::{Component, ComponentLibrary, ComponentOption}; pub use compositing::{fill_blend_mode_at, node_blend_mode, node_mask_type}; pub use design_md::{extract_design_md_from_document, generate_design_md, parse_design_md}; pub use design_md_button_state::DesignMdButton; -pub use document_install::{DocumentInstallError, DocumentInstallReport}; +pub use document_install::{DocumentInstallError, DocumentInstallReport, PreparedDocument}; pub use edit_transaction::{ CompletedLocalEdit, EditOrigin, LocalEditCapture, LocalEditError, LocalEditOutcome, }; diff --git a/crates/op-host-services/Cargo.toml b/crates/op-host-services/Cargo.toml index 9160a83f4..6170f369a 100644 --- a/crates/op-host-services/Cargo.toml +++ b/crates/op-host-services/Cargo.toml @@ -47,6 +47,10 @@ anthropic-agent-sdk = { path = "../../vendor/anthropic-agent-sdk" } # default-features=false drops winit/glutin/GL). Pulled solely for the export # endpoint's NativeBackend + NativeFrameBackend. op-host-native = { path = "../op-host-native", default-features = false } +# Collaboration runtime — the daemon runs the same owner/guest actors the +# desktop GUI runs, so a browser client gets real sessions instead of a +# permanently `Unavailable` panel. +op-collab-host = { path = "../op-collab-host" } # --- external (crates.io / forks) --- github-copilot-sdk = "0.1" diff --git a/crates/op-host-services/src/collab_blocking.rs b/crates/op-host-services/src/collab_blocking.rs new file mode 100644 index 000000000..3cc865702 --- /dev/null +++ b/crates/op-host-services/src/collab_blocking.rs @@ -0,0 +1,31 @@ +//! The blocking executor injected into `op-collab-host`. +//! +//! The collaboration runtime is synchronous but its relay bridges and JWKS +//! fetch are futures, so it asks its host for an async → sync bridge rather +//! than owning one. This crate owns +//! [`block_on_anywhere`](crate::chat_runtime::block_on_anywhere), the +//! workspace's single sanctioned bridge, so the implementation lives here and +//! every host — the desktop GUI and the headless daemon alike — installs this +//! one rather than writing its own. + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; + +use op_collab_host::BlockingExecutor; + +struct ServicesBlockingExecutor; + +impl BlockingExecutor for ServicesBlockingExecutor { + fn block_on_erased(&self, future: Pin<&mut (dyn Future + '_)>) { + crate::chat_runtime::block_on_anywhere(future); + } +} + +/// Install the process-wide executor. +/// +/// Idempotent — the first install wins — so every entry point that might be +/// the first to start a session can call it unconditionally. +pub fn install() { + let _ = op_collab_host::install_blocking_executor(Arc::new(ServicesBlockingExecutor)); +} diff --git a/crates/op-host-services/src/lib.rs b/crates/op-host-services/src/lib.rs index 1122ce0b7..96a46a082 100644 --- a/crates/op-host-services/src/lib.rs +++ b/crates/op-host-services/src/lib.rs @@ -50,6 +50,7 @@ pub mod cli_probe_error; mod cli_probe_support; pub mod cli_provider_probe; mod cli_resolver_windows; +pub mod collab_blocking; mod design_agent_diagnostics; #[cfg(test)] mod design_agent_reflow_tests; diff --git a/crates/op-host-services/src/web_canvas_server.rs b/crates/op-host-services/src/web_canvas_server.rs index 5b17d6d6f..ad5fdfa3e 100644 --- a/crates/op-host-services/src/web_canvas_server.rs +++ b/crates/op-host-services/src/web_canvas_server.rs @@ -45,19 +45,35 @@ const MAX_CONNS: usize = 64; /// detected (the heartbeat write fails once the socket is gone). const SSE_HEARTBEAT: Duration = Duration::from_secs(15); +/// One SSE payload: the two sequence numbers a client polls on. +/// +/// `version` changes only when document *content* changes, so it is what makes +/// a client refetch the document. `collab_seq` changes whenever the +/// collaboration projection changes — a peer joining, a cursor moving — and +/// must never by itself cause a document fetch. +/// +/// Carried as one struct so a bump and its broadcast stay a single atomic +/// event. The serialized form stays a superset of the original `{"version":N}` +/// payload, so a client that only reads `version` is unaffected. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct SseTick { + pub version: u64, + pub collab_seq: u64, +} + /// Broadcast hub for SSE subscribers. Each `GET /api/mcp/events` connection -/// registers a channel; a document mutation broadcasts the new version to all +/// registers a channel; a document mutation broadcasts the new tick to all /// of them, and each SSE connection thread writes it to its socket. Senders to /// disconnected clients are pruned on the next broadcast. #[derive(Default)] pub struct SseHub { - subscribers: Mutex>>, + subscribers: Mutex>>, } impl SseHub { /// Register a subscriber; the SSE connection thread blocks on the returned - /// receiver for version bumps. - pub(crate) fn subscribe(&self) -> Receiver { + /// receiver for ticks. + pub(crate) fn subscribe(&self) -> Receiver { let (tx, rx) = mpsc::channel(); self.subscribers .lock() @@ -66,13 +82,13 @@ impl SseHub { rx } - /// Broadcast a version bump to all live subscribers, pruning any whose - /// receiver was dropped (client disconnected). - pub(crate) fn broadcast(&self, version: u64) { + /// Broadcast a tick to all live subscribers, pruning any whose receiver was + /// dropped (client disconnected). + pub(crate) fn broadcast(&self, tick: SseTick) { self.subscribers .lock() .unwrap_or_else(|p| p.into_inner()) - .retain(|tx| tx.send(version).is_ok()); + .retain(|tx| tx.send(tick).is_ok()); } #[cfg(test)] @@ -129,6 +145,9 @@ pub struct WebCanvasState { /// In-flight browser device-login flow driven through the auth proxy /// endpoints (`/api/auth/login/*`); `None` when no login is running. pub(crate) auth_login_handle: Option, + /// Collaboration runtime + its projection sequence. Idle until a browser + /// posts a start/join action. + pub(crate) collab: collab_state::WebCollabState, } impl WebCanvasState { @@ -179,6 +198,7 @@ impl WebCanvasState { allow_origins: Vec::new(), reset_consumed: false, auth_login_handle: None, + collab: collab_state::WebCollabState::default(), } } @@ -217,6 +237,14 @@ impl WebCanvasState { if self.reset_consumed { return Ok(ResetOutcome { skipped: true }); } + // A reset swaps the whole document and starts a new generation, which + // the M1 protocol cannot sequence — the peers would silently diverge. + // `ReplaceDocument` is refused for every source once a session exists. + self.gate_daemon_mutation( + op_editor_core::CollabGateAction::ReplaceDocument, + op_editor_core::CollabEditSource::ExternalSync, + ) + .map_err(WebCanvasError::Collab)?; self.reset_document()?; self.reset_consumed = true; Ok(ResetOutcome { skipped: false }) @@ -267,7 +295,36 @@ impl WebCanvasState { for w in &loaded.warnings { eprintln!("openpencil-desktop --serve-web: schema warning: {w:?}"); } - let version = self.replace_document(loaded.value); + // Structural validation happens here, before any state changes, so the + // in-session install below is infallible and cannot leave a half-applied + // document inside an open edit capture. + let prepared = op_editor_core::PreparedDocument::prepare(loaded.value) + .map_err(|e| WebCanvasError::Document(e.to_string()))?; + // The gateway. Returns Ok untouched when there is no session, so a + // standalone daemon behaves exactly as it did before collaboration. + self.gate_daemon_mutation( + op_editor_core::CollabGateAction::Document( + op_editor_core::CollabDocumentMutation::NodePropertyBatch, + ), + op_editor_core::CollabEditSource::User, + ) + .map_err(WebCanvasError::Collab)?; + let version = if self.collab_session_is_active() { + // Inside the session's local-edit capture: the session core diffs + // the document and emits precise node operations, so pushing an + // unchanged document produces no transaction at all. + if !self.ingest_document_in_session(prepared) { + return Err(WebCanvasError::Collab( + collab_state::DaemonMutationRefusal::Collab( + op_editor_core::CollabGateReason::PendingEdit, + ), + )); + } + self.version = self.version.saturating_add(1); + self.version + } else { + self.replace_document(prepared.into_document()) + }; op_pen_loader::apply_editor_meta(&mut self.editor, editor_meta); Ok(PushOutcome { applied: true, @@ -275,12 +332,43 @@ impl WebCanvasState { }) } + /// The pair of sequence numbers a client polls on. + pub(crate) const fn sse_tick(&self) -> SseTick { + SseTick { + version: self.version, + collab_seq: self.collab.seq(), + } + } + #[cfg(test)] pub(crate) fn document_version_for_test(&self) -> u64 { self.version } } +/// Render an error, adding the machine-readable code when it has one. +/// +/// A collaboration refusal is not a malformed request, so a client needs the +/// code to decide between retrying, refetching, and telling the user to leave +/// the session. +fn collab_aware_error_reply(error: &WebCanvasError) -> WebReply { + match error.error_code() { + Some(code) => WebReply { + status: error.http_status(), + body: serde_json::json!({ + "ok": false, + "error": code, + "message": error.to_string(), + }) + .to_string(), + }, + None => WebReply { + status: error.http_status(), + body: crate::mcp_serve::rest_error_body(&error.to_string()), + }, + } +} + /// Outcome of [`WebCanvasState::reset_document_guarded`]. pub(crate) struct ResetOutcome { pub skipped: bool, @@ -404,14 +492,19 @@ pub fn handle_web_canvas_request( }) .to_string(), }, - Err(error) => WebReply { - status: error.http_status(), - body: crate::mcp_serve::rest_error_body(&error.to_string()), - }, + Err(error) => collab_aware_error_reply(&error), }, ("GET", "/api/mcp/version") => WebReply { + // `collabSeq` rides along so the existing 400 ms version poll also + // notices collaboration changes without a second request. Adding a + // field is backward compatible: a client that reads only `version` + // is unaffected. status: "200 OK", - body: format!(r#"{{"version":{}}}"#, state.version), + body: format!( + r#"{{"version":{},"collabSeq":{}}}"#, + state.version, + state.collab.seq() + ), }, ("GET", "/api/mcp/indicators") => WebReply { // Agent-indicator relay: design runs execute inside this @@ -431,6 +524,7 @@ pub fn handle_web_canvas_request( status: "200 OK", body: crate::mcp_serve::document_sync_ok(state.version), }, + Err(e) if e.error_code().is_some() => collab_aware_error_reply(&e), Err(e) => WebReply { status: e.http_status(), body: crate::mcp_serve::rest_error_body(&format!("sync reset failed: {e}")), @@ -522,6 +616,11 @@ pub fn handle_web_canvas_request( ("GET", op_editor_core::auth_routes::LOGIN_STATUS) => crate::web_auth::login_status(state), ("POST", op_editor_core::auth_routes::LOGIN_CANCEL) => crate::web_auth::login_cancel(state), ("POST", op_editor_core::auth_routes::LOGOUT) => crate::web_auth::logout(state), + // Collaboration: the runtime lives in this daemon, the panel in the + // browser. See `web_canvas_server/collab_routes.rs`. + ("GET", op_editor_core::collab_routes::STATE) => collab_routes::state(state), + ("POST", op_editor_core::collab_routes::ACTION) => collab_routes::action(body, state), + ("POST", op_editor_core::collab_routes::PRESENCE) => collab_routes::presence(body, state), _ => WebReply { status: "404 Not Found", body: r#"{"ok":false,"error":"Not found. Use /api/mcp/document, /api/mcp/sync-reset, /api/mcp/server, /api/file/save, /api/export/raster, /api/export/pdf, or /mcp."}"# @@ -530,6 +629,9 @@ pub fn handle_web_canvas_request( } } +mod collab_driver; +mod collab_routes; +pub(crate) mod collab_state; mod connect_routes; mod connection; mod doc_routes; @@ -538,6 +640,7 @@ mod origin_guard; mod run_loop; mod serve_options; +pub use collab_state::DaemonMutationRefusal; pub use connect_routes::*; use connection::*; use doc_routes::*; diff --git a/crates/op-host-services/src/web_canvas_server/collab_driver.rs b/crates/op-host-services/src/web_canvas_server/collab_driver.rs new file mode 100644 index 000000000..ed05a2f6f --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/collab_driver.rs @@ -0,0 +1,250 @@ +//! The daemon's collaboration pump. +//! +//! The desktop drives its runtime from the winit frame loop. A daemon has no +//! frames, so it gets a dedicated thread that ticks on a timeout and is woken +//! early by anything that gives the runtime work: a network event (through the +//! runtime's own wake notifier) or a browser POST. +//! +//! ## Locking +//! +//! The tick takes the state lock for its whole body, and the wake notifier +//! never takes it — it only pokes a bounded channel. That is what keeps a +//! network worker thread from blocking on the document mutex while the REST +//! tier is serving a large document push. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use super::{SseHub, WebCanvasState}; + +/// Tick cadence with no live session. Long enough to cost nothing, short +/// enough that a start/join action posted just after a tick still feels +/// immediate. +const IDLE_TICK: Duration = Duration::from_millis(250); + +/// Tick cadence during a session, matching the presence frame budget. +const ACTIVE_TICK: Duration = Duration::from_millis(100); + +/// Pokes the driver awake from any thread. +/// +/// A one-slot channel: the driver only needs to know that *something* +/// happened, so a wake arriving while one is already pending is dropped rather +/// than queued. `try_send` never blocks, which is what makes this safe to call +/// from a network worker holding transport locks. +pub(crate) struct CollabWaker { + tx: SyncSender<()>, +} + +impl CollabWaker { + pub(crate) fn wake(&self) { + let _ = self.tx.try_send(()); + } +} + +/// Start the collaboration driver for this daemon. +/// +/// Installs the blocking executor the runtime needs, registers the wake path, +/// and spawns the pump. The thread observes `shutdown` on every tick, so it +/// retires within one tick of the daemon being asked to stop. +pub(crate) fn spawn( + state: &Arc>, + hub: &Arc, + shutdown: &Arc, +) { + crate::collab_blocking::install(); + + let (tx, rx) = sync_channel(1); + let waker = Arc::new(CollabWaker { tx }); + { + let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); + guard.collab.set_waker(Arc::clone(&waker)); + let notifier_waker = Arc::clone(&waker); + guard + .collab + .runtime + .set_wake_notifier(Arc::new(move || notifier_waker.wake())); + } + + let state = Arc::clone(state); + let hub = Arc::clone(hub); + let shutdown = Arc::clone(shutdown); + let spawned = std::thread::Builder::new() + .name("op-serve-web-collab".into()) + .spawn(move || run(&state, &hub, &shutdown, &rx)); + if spawned.is_err() { + eprintln!( + "openpencil-desktop --serve-web: could not start the collaboration driver; \ + collaboration stays unavailable" + ); + } +} + +fn run(state: &Mutex, hub: &SseHub, shutdown: &AtomicBool, wakes: &Receiver<()>) { + while !shutdown.load(Ordering::Acquire) { + let active = tick(state, hub); + let timeout = if active { ACTIVE_TICK } else { IDLE_TICK }; + // A timeout and a wake are the same signal here; the tick is + // idempotent, so a spurious wake only costs one pass. + let _ = wakes.recv_timeout(timeout); + } +} + +/// One pass of the runtime pump. Returns whether a session is live, which +/// selects the next tick cadence. +fn tick(state: &Mutex, hub: &SseHub) -> bool { + let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); + let guard = &mut *guard; + + let revision_before = guard.editor.document_revision(); + let presence_cursor = guard.collab.presence_override(); + + let (runtime, mut host) = guard.collab_runtime_and_host(); + // Order matches the desktop frame pump: availability first so the panel + // never offers an action the runtime would refuse, then the queued action, + // then inbound network events, then outbound presence. + let mut projection_changed = runtime.refresh_availability(&mut host); + projection_changed |= runtime.drain_ui_action(&mut host); + projection_changed |= runtime.poll(&mut host); + projection_changed |= runtime.publish_local_presence(&mut host, presence_cursor); + + for status in runtime.drain_status_events() { + eprintln!("[collab] {status:?}"); + } + + // The host dirty flag is raised by every runtime write through the seam, + // including pure projection writes that change no document content. + projection_changed |= guard.collab.take_dirty(); + + let revision_after = guard.editor.document_revision(); + if revision_after != revision_before { + // Document content moved (a remote commit landed, or a local edit was + // accepted). Only this bumps the document version, so a browser + // refetches the document for content changes and nothing else. + guard.version = guard.version.saturating_add(1); + projection_changed = true; + } + if projection_changed { + guard.collab.bump_seq(); + // Broadcast inside the lock, matching the REST tier's contract: the + // bump and its broadcast must be atomic or two mutations can publish + // out of order. + hub.broadcast(guard.sse_tick()); + } + + guard.collab_session_is_active() +} + +#[cfg(test)] +mod tests { + use super::*; + use op_editor_core::EditorState; + + fn daemon() -> Mutex { + Mutex::new(WebCanvasState::new(EditorState::starter(), 0)) + } + + #[test] + fn an_idle_tick_publishes_nothing_after_the_first_settling_pass() { + let state = daemon(); + let hub = SseHub::default(); + let sub = hub.subscribe(); + + // The first pass may publish the initial availability projection. + assert!(!tick(&state, &hub)); + let settled = state.lock().unwrap_or_else(|p| p.into_inner()).collab.seq(); + while sub.try_recv().is_ok() {} + + // Every pass after that is silent, which is what makes a 250 ms idle + // cadence free: no bump, no broadcast, no browser refetch. + for _ in 0..3 { + assert!(!tick(&state, &hub)); + } + assert_eq!( + state.lock().unwrap_or_else(|p| p.into_inner()).collab.seq(), + settled + ); + assert!(sub.try_recv().is_err(), "an idle daemon must stay quiet"); + } + + #[test] + fn a_host_write_bumps_only_the_projection_sequence_and_broadcasts_it() { + let state = daemon(); + let hub = SseHub::default(); + let sub = hub.subscribe(); + tick(&state, &hub); + while sub.try_recv().is_ok() {} + + let (version_before, seq_before) = { + let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); + let before = (guard.version, guard.collab.seq()); + // Stand in for a runtime write that touched only the projection — + // a peer's cursor moving, say. + let (_runtime, mut host) = guard.collab_runtime_and_host(); + op_collab_host::CollabHost::mark_editor_state_dirty(&mut host); + before + }; + + assert!(!tick(&state, &hub)); + let guard = state.lock().unwrap_or_else(|p| p.into_inner()); + assert_eq!( + guard.version, version_before, + "presence-shaped changes must not move the document version, or \ + every cursor move would make the browser refetch the document" + ); + assert_eq!(guard.collab.seq(), seq_before + 1); + let tick_out = sub.try_recv().expect("the change was broadcast"); + assert_eq!(tick_out.version, version_before); + assert_eq!(tick_out.collab_seq, seq_before + 1); + } + + #[test] + fn a_write_that_landed_between_ticks_is_not_bumped_a_second_time() { + let state = daemon(); + let hub = SseHub::default(); + let sub = hub.subscribe(); + tick(&state, &hub); + while sub.try_recv().is_ok() {} + + // A REST push mutated the document and published its own version while + // the driver was asleep. + let version_before = { + let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); + guard.editor.mark_document_changed(); + guard.version = guard.version.saturating_add(1); + guard.version + }; + + // The driver compares the revision across its own tick, so a change + // that landed before the tick started is invisible to it. That is what + // keeps one logical mutation from publishing two versions and making + // every browser fetch the same document twice. + tick(&state, &hub); + assert_eq!( + state.lock().unwrap_or_else(|p| p.into_inner()).version, + version_before + ); + assert!(sub.try_recv().is_err(), "nothing new to announce"); + } + + #[test] + fn the_tick_drains_a_queued_action_from_the_pending_slot() { + let state = daemon(); + let hub = SseHub::default(); + { + let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); + guard.editor.editor_ui.collab.pending_action = + Some(op_editor_core::CollabUiAction::OpenCreate); + } + + tick(&state, &hub); + + let guard = state.lock().unwrap_or_else(|p| p.into_inner()); + assert!( + guard.editor.editor_ui.collab.pending_action.is_none(), + "the driver is what frees the single action slot; a stuck action \ + would 409 every later post" + ); + } +} diff --git a/crates/op-host-services/src/web_canvas_server/collab_routes.rs b/crates/op-host-services/src/web_canvas_server/collab_routes.rs new file mode 100644 index 000000000..7b2b797a1 --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/collab_routes.rs @@ -0,0 +1,141 @@ +//! `/api/collab/*` handlers. +//! +//! All three run inside the REST tier, so they hold the state lock for their +//! whole body and must never block on IO. They are deliberately thin: the +//! session work happens on the driver thread, and a posted action only lands +//! in the pending slot the driver drains. + +use op_editor_core::collab_wire::{ + CollabActionWire, CollabActionWireError, CollabLocalPresenceWire, CollabStateWire, +}; + +use super::{WebCanvasState, WebReply}; + +/// Longest body either POST route accepts. +/// +/// Both bodies are a handful of scalars; anything larger is a client bug or an +/// attempt to make the daemon allocate. +const MAX_COLLAB_BODY_BYTES: usize = 8 * 1024; + +/// `GET /api/collab/state` — the whole projection plus both sequence numbers. +pub(crate) fn state(state: &mut WebCanvasState) -> WebReply { + let wire = CollabStateWire::from_ui( + &state.editor.editor_ui.collab, + state.collab.seq(), + state.editor.document_revision(), + ); + match serde_json::to_string(&wire) { + Ok(body) => WebReply { + status: "200 OK", + body, + }, + Err(error) => WebReply { + status: "500 Internal Server Error", + body: crate::mcp_serve::rest_error_body(&error.to_string()), + }, + } +} + +/// `POST /api/collab/action` — enqueue one UI action for the driver. +/// +/// Returns 202 without waiting: the runtime call can start a network worker, +/// and the REST tier holds the state lock. The driver is woken so the action is +/// picked up on the next tick rather than at the idle timeout. +pub(crate) fn action(body: &str, state: &mut WebCanvasState) -> WebReply { + if body.len() > MAX_COLLAB_BODY_BYTES { + return error_reply( + "413 Payload Too Large", + "payload-too-large", + "body too large", + ); + } + let wire: CollabActionWire = match serde_json::from_str(body) { + Ok(wire) => wire, + Err(error) => { + return error_reply("400 Bad Request", "malformed-action", &error.to_string()) + } + }; + // Hook point for the public multi-account deployment: it must refuse + // `wire.reaches_caller_named_network()` here, because those actions resolve + // a caller-named socket address or enumerate the host's LAN. The local and + // managed daemons allow them — that is desktop parity, and the operator is + // the only client. + let action = match wire.into_ui_action() { + Ok(action) => action, + Err(error) => return action_error_reply(error), + }; + // The pending slot holds exactly one action, and the runtime consumes it + // through `take_pending_action`. Overwriting an undrained action would + // silently drop whatever the user asked for first, so this reports the + // collision instead. + if state.editor.editor_ui.collab.pending_action.is_some() { + return error_reply( + "409 Conflict", + "collab-busy", + "a collaboration action is already queued", + ); + } + state.editor.editor_ui.collab.pending_action = Some(action); + let seq = state.collab.bump_seq(); + state.collab.wake_driver(); + WebReply { + status: "202 Accepted", + body: serde_json::json!({ "ok": true, "collabSeq": seq }).to_string(), + } +} + +/// `POST /api/collab/presence` — publish the local cursor. +/// +/// Stored rather than sent: the driver replays it through +/// `publish_local_presence`, which is throttled to one frame interval. Posting +/// faster than that costs nothing beyond the request itself. +pub(crate) fn presence(body: &str, state: &mut WebCanvasState) -> WebReply { + if body.len() > MAX_COLLAB_BODY_BYTES { + return error_reply( + "413 Payload Too Large", + "payload-too-large", + "body too large", + ); + } + let wire: CollabLocalPresenceWire = match serde_json::from_str(body) { + Ok(wire) => wire, + Err(error) => { + return error_reply("400 Bad Request", "malformed-presence", &error.to_string()) + } + }; + // A presence update is not a projection change for the poller's purposes — + // it is the local peer's own cursor going out, not incoming state — so this + // deliberately does not bump `collabSeq`. + state + .collab + .set_presence_override(wire.cursor.map(|point| (point.x, point.y))); + state.collab.wake_driver(); + WebReply { + status: "202 Accepted", + body: r#"{"ok":true}"#.into(), + } +} + +fn action_error_reply(error: CollabActionWireError) -> WebReply { + let code = match error { + CollabActionWireError::InvalidRequestKey => "invalid-request-key", + CollabActionWireError::InvalidAddress => "invalid-address", + }; + error_reply("400 Bad Request", code, &error.to_string()) +} + +fn error_reply(status: &'static str, code: &str, message: &str) -> WebReply { + WebReply { + status, + body: serde_json::json!({ + "ok": false, + "error": code, + "message": message, + }) + .to_string(), + } +} + +#[cfg(test)] +#[path = "collab_routes_tests.rs"] +mod tests; diff --git a/crates/op-host-services/src/web_canvas_server/collab_routes_tests.rs b/crates/op-host-services/src/web_canvas_server/collab_routes_tests.rs new file mode 100644 index 000000000..0a3316000 --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/collab_routes_tests.rs @@ -0,0 +1,331 @@ +//! `/api/collab/*` route coverage, driven through the same +//! `handle_web_canvas_request` entry point the connection loop uses. + +use op_editor_core::collab_wire::{CollabStateWire, COLLAB_WIRE_VERSION}; +use op_editor_core::{ + AuthenticatedCollabSession, CollabConnectionPhase, CollabUiAction, CollabUiRole, EditorState, +}; + +use super::super::{handle_web_canvas_request, WebCanvasState, WebReply}; + +fn daemon() -> WebCanvasState { + WebCanvasState::new(EditorState::starter(), 0) +} + +fn call(method: &str, path: &str, body: &str, state: &mut WebCanvasState) -> WebReply { + handle_web_canvas_request(method, path, body, state) +} + +fn json(reply: &WebReply) -> serde_json::Value { + serde_json::from_str(&reply.body).expect("route bodies are JSON") +} + +#[test] +fn state_answers_a_versioned_projection_even_with_no_session() { + let mut state = daemon(); + let reply = call("GET", op_editor_core::collab_routes::STATE, "", &mut state); + + assert_eq!(reply.status, "200 OK"); + let wire: CollabStateWire = serde_json::from_str(&reply.body).expect("decodes as the DTO"); + assert_eq!(wire.wire_version, COLLAB_WIRE_VERSION); + assert_eq!(wire.collab_seq, 0); + assert!(wire.session.is_none()); + // Availability is whatever the runtime last published; the point is that + // the field is always present so a client can render a reason. + assert!(json(&reply).get("availability").is_some()); +} + +#[test] +fn state_reports_the_two_sequence_numbers_separately() { + let mut state = daemon(); + state.collab.bump_seq(); + state.collab.bump_seq(); + + let wire: CollabStateWire = serde_json::from_str( + &call("GET", op_editor_core::collab_routes::STATE, "", &mut state).body, + ) + .expect("decodes"); + assert_eq!(wire.collab_seq, 2); + assert_eq!(wire.document_revision, state.editor.document_revision()); +} + +#[test] +fn a_posted_action_lands_in_the_pending_slot_and_bumps_the_projection() { + let mut state = daemon(); + let reply = call( + "POST", + op_editor_core::collab_routes::ACTION, + r#"{"type":"start"}"#, + &mut state, + ); + + assert_eq!(reply.status, "202 Accepted"); + assert_eq!(json(&reply)["collabSeq"], 1); + assert_eq!( + state.editor.editor_ui.collab.pending_action, + Some(CollabUiAction::Start) + ); +} + +#[test] +fn a_second_action_conflicts_instead_of_overwriting_the_first() { + let mut state = daemon(); + call( + "POST", + op_editor_core::collab_routes::ACTION, + r#"{"type":"start"}"#, + &mut state, + ); + let reply = call( + "POST", + op_editor_core::collab_routes::ACTION, + r#"{"type":"leave"}"#, + &mut state, + ); + + assert_eq!(reply.status, "409 Conflict"); + assert_eq!(json(&reply)["error"], "collab-busy"); + assert_eq!( + state.editor.editor_ui.collab.pending_action, + Some(CollabUiAction::Start), + "the queued action must survive — dropping it would lose what the user asked for" + ); +} + +#[test] +fn a_malformed_action_is_refused_before_it_reaches_the_runtime() { + let mut state = daemon(); + for (body, code) in [ + (r#"{"type":"selfDestruct"}"#, "malformed-action"), + (r#"not json"#, "malformed-action"), + ( + r#"{"type":"rejectAdmission","requestKey":"has space"}"#, + "invalid-request-key", + ), + ( + r#"{"type":"joinAddress","endpoint":" "}"#, + "invalid-address", + ), + ] { + let reply = call( + "POST", + op_editor_core::collab_routes::ACTION, + body, + &mut state, + ); + assert_eq!(reply.status, "400 Bad Request", "{body}"); + assert_eq!(json(&reply)["error"], code, "{body}"); + assert!(state.editor.editor_ui.collab.pending_action.is_none()); + } +} + +#[test] +fn lan_actions_are_accepted_by_a_local_daemon() { + // Desktop parity: the operator is the only client of a local or managed + // daemon, so LAN discovery and direct joins stay available. The public + // deployment is where these get refused. + for body in [ + r#"{"type":"startLan"}"#, + r#"{"type":"beginDiscovery"}"#, + r#"{"type":"joinAddress","endpoint":"192.168.1.10:43120"}"#, + ] { + let mut state = daemon(); + let reply = call( + "POST", + op_editor_core::collab_routes::ACTION, + body, + &mut state, + ); + assert_eq!(reply.status, "202 Accepted", "{body}"); + } +} + +#[test] +fn presence_is_stored_without_bumping_the_projection_sequence() { + let mut state = daemon(); + let reply = call( + "POST", + op_editor_core::collab_routes::PRESENCE, + r#"{"cursor":{"x":12.5,"y":-3.0}}"#, + &mut state, + ); + + assert_eq!(reply.status, "202 Accepted"); + assert_eq!(state.collab.presence_override(), Some((12.5, -3.0))); + assert_eq!( + state.collab.seq(), + 0, + "the local cursor going out is not incoming state; bumping here would \ + make every mouse move wake every poller" + ); +} + +#[test] +fn presence_accepts_a_cleared_cursor() { + let mut state = daemon(); + call( + "POST", + op_editor_core::collab_routes::PRESENCE, + r#"{"cursor":{"x":1.0,"y":2.0}}"#, + &mut state, + ); + call( + "POST", + op_editor_core::collab_routes::PRESENCE, + r#"{"cursor":null}"#, + &mut state, + ); + assert_eq!(state.collab.presence_override(), None); +} + +#[test] +fn oversized_bodies_are_refused_before_parsing() { + let mut state = daemon(); + let huge = format!( + r#"{{"type":"joinAddress","endpoint":"{}"}}"#, + "a".repeat(9000) + ); + for path in [ + op_editor_core::collab_routes::ACTION, + op_editor_core::collab_routes::PRESENCE, + ] { + let reply = call("POST", path, &huge, &mut state); + assert_eq!(reply.status, "413 Payload Too Large", "{path}"); + } +} + +#[test] +fn the_version_probe_carries_the_projection_sequence_too() { + let mut state = daemon(); + state.collab.bump_seq(); + let reply = call("GET", "/api/mcp/version", "", &mut state); + + let body = json(&reply); + // Additive: the existing 400 ms poll keeps reading `version` and now also + // notices collaboration changes without a second request. + assert_eq!(body["version"], 0); + assert_eq!(body["collabSeq"], 1); +} + +#[test] +fn a_session_projects_through_the_state_route() { + let mut state = daemon(); + state.editor.editor_ui.collab.set_authenticated_session( + CollabConnectionPhase::Active, + AuthenticatedCollabSession { + session_name: "studio".into(), + role: CollabUiRole::Owner, + share_endpoint: None, + }, + Vec::new(), + ); + + let wire: CollabStateWire = serde_json::from_str( + &call("GET", op_editor_core::collab_routes::STATE, "", &mut state).body, + ) + .expect("decodes"); + let session = wire.session.expect("session projected"); + assert_eq!(session.session_name, "studio"); +} + +#[test] +fn document_push_is_refused_with_a_code_while_a_session_is_read_only_for_us() { + let mut state = daemon(); + state.editor.editor_ui.collab.set_authenticated_session( + CollabConnectionPhase::Active, + AuthenticatedCollabSession { + session_name: "studio".into(), + role: CollabUiRole::Viewer, + share_endpoint: None, + }, + Vec::new(), + ); + let before = state.editor.doc.clone(); + + let reply = call( + "POST", + "/api/mcp/document", + r#"{"document":{"version":"1.0","children":[]}}"#, + &mut state, + ); + + assert_eq!(reply.status, "409 Conflict"); + assert_eq!(json(&reply)["error"], "collab-readonly"); + assert_eq!(state.editor.doc, before, "a refused push writes nothing"); +} + +#[test] +fn sync_reset_and_open_recent_are_refused_while_a_session_is_active() { + for (path, body) in [ + ("/api/mcp/sync-reset", ""), + ("/api/file/open-recent", r#"{"path":"/tmp/x.op"}"#), + ] { + let mut state = daemon(); + state.editor.editor_ui.collab.set_authenticated_session( + CollabConnectionPhase::Active, + AuthenticatedCollabSession { + session_name: "studio".into(), + role: CollabUiRole::Owner, + share_endpoint: None, + }, + Vec::new(), + ); + + let reply = call("POST", path, body, &mut state); + assert_eq!(reply.status, "409 Conflict", "{path}"); + assert_eq!(json(&reply)["error"], "collab-active", "{path}"); + } +} + +#[test] +fn sync_reset_still_works_with_no_session() { + let mut state = daemon(); + let reply = call("POST", "/api/mcp/sync-reset", "", &mut state); + assert_eq!(reply.status, "200 OK", "{}", reply.body); + assert_eq!(json(&reply)["ok"], true); +} + +#[test] +fn the_collab_routes_are_gated_as_sensitive_browser_posts() { + for path in [ + op_editor_core::collab_routes::ACTION, + op_editor_core::collab_routes::PRESENCE, + ] { + let request = crate::mcp_serve::HttpRequest { + method: "POST".into(), + path: path.into(), + body: String::new(), + content_type: Some("application/json".into()), + origin: None, + host: None, + token: None, + }; + assert!( + super::super::is_sensitive_browser_post(&request), + "{path} must sit behind the same-origin gate" + ); + } +} + +#[test] +fn managed_mode_requires_the_token_for_every_collab_route() { + let auth = super::super::RequestAuth { + managed: true, + token: "secret".into(), + }; + for (method, path) in [ + ("GET", op_editor_core::collab_routes::STATE), + ("POST", op_editor_core::collab_routes::ACTION), + ("POST", op_editor_core::collab_routes::PRESENCE), + ] { + assert!( + !auth.allows(method, path, None), + "{path} must not be reachable without the managed token" + ); + assert!( + !auth.allows(method, path, Some("wrong")), + "{path} must reject a mismatched token" + ); + assert!(auth.allows(method, path, Some("secret")), "{path}"); + } +} diff --git a/crates/op-host-services/src/web_canvas_server/collab_state.rs b/crates/op-host-services/src/web_canvas_server/collab_state.rs new file mode 100644 index 000000000..8b6cb0414 --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/collab_state.rs @@ -0,0 +1,294 @@ +//! Collaboration runtime attached to the daemon's single editor. +//! +//! The daemon is a native process, so unlike the browser it can hold the +//! device credentials a relay session needs. It runs the same +//! `op_collab_host::CollabRuntime` the desktop GUI runs; the browser only +//! paints the projection and posts actions. +//! +//! ## The split borrow +//! +//! The runtime wants `&mut impl CollabHost`, and a host is an editor plus an +//! id-allocation slot plus a dirty flag. `HeadlessCollabHost` *owns* its +//! editor, which is right for a daemon that starts empty but wrong here: this +//! daemon's editor already lives on [`WebCanvasState`] next to the version +//! counter, the backing path, and the credential policy. So the host is a +//! borrowing view assembled per call — [`DaemonCollabHost`] — with exactly the +//! semantics of `HeadlessCollabHost`. + +use op_collab_host::{CollabHost, CollabRuntime}; +use op_editor_core::{ + next_sequential_counter, CollabEditSource, CollabGateAction, CollabGatePolicy, + CollabGateReason, DocumentIdAllocator, DocumentInstallError, DocumentInstallReport, EditOrigin, + EditorState, IdAllocError, PeerNamespace, PenDocument, PreparedDocument, +}; +use op_editor_host_core::collab::CollaborationEditorHost; + +use super::WebCanvasState; + +/// First id a standalone daemon hands out, matching the GUI host's seed. +const FIRST_STANDALONE_NODE_ID: u64 = 100; + +/// Everything the daemon keeps for collaboration, beside the editor itself. +pub(crate) struct WebCollabState { + /// The owner/guest actors and their network workers. + pub(crate) runtime: CollabRuntime, + /// Owner-assigned id namespace while a session is live. + id_allocator: Option, + /// Next id under the standalone sequential policy. + next_node_id: u64, + /// Raised by any runtime write through the host seam. + dirty: bool, + /// Bumps whenever the *projection* changes — a peer joining, a cursor + /// moving, a notice appearing. Deliberately separate from the document + /// version so presence traffic never makes a browser refetch the document. + seq: u64, + /// Last cursor the browser published, replayed to peers by the driver. + presence_override: Option<(f64, f64)>, + /// Pokes the driver thread awake. `None` until the driver starts, and in + /// the unit tests, which tick the runtime directly. + waker: Option>, +} + +impl Default for WebCollabState { + fn default() -> Self { + Self { + runtime: CollabRuntime::new(), + id_allocator: None, + next_node_id: FIRST_STANDALONE_NODE_ID, + dirty: false, + seq: 0, + presence_override: None, + waker: None, + } + } +} + +impl WebCollabState { + /// Current projection sequence number. + pub(crate) const fn seq(&self) -> u64 { + self.seq + } + + /// Advance the projection sequence number. + pub(crate) fn bump_seq(&mut self) -> u64 { + self.seq = self.seq.saturating_add(1); + self.seq + } + + /// Read and clear the host dirty flag. + pub(crate) fn take_dirty(&mut self) -> bool { + std::mem::take(&mut self.dirty) + } + + /// Record the browser's latest cursor for the next presence flush. + pub(crate) fn set_presence_override(&mut self, cursor: Option<(f64, f64)>) { + self.presence_override = cursor; + } + + pub(crate) const fn presence_override(&self) -> Option<(f64, f64)> { + self.presence_override + } + + /// Register the driver's wake path. + pub(crate) fn set_waker(&mut self, waker: std::sync::Arc) { + self.waker = Some(waker); + } + + /// Ask the driver to tick now instead of at its next timeout. + pub(crate) fn wake_driver(&self) { + if let Some(waker) = self.waker.as_ref() { + waker.wake(); + } + } +} + +/// A borrowing [`CollabHost`] over the daemon's editor. +/// +/// Assembled per call so the editor can stay where the rest of the daemon +/// already reads it. Every method matches `HeadlessCollabHost`. +pub(crate) struct DaemonCollabHost<'a> { + editor: &'a mut EditorState, + id_allocator: &'a mut Option, + next_node_id: &'a mut u64, + dirty: &'a mut bool, +} + +impl CollaborationEditorHost for DaemonCollabHost<'_> { + fn editor_state(&self) -> &EditorState { + self.editor + } + + fn editor_state_mut(&mut self) -> &mut EditorState { + self.editor + } + + fn install_collaboration_document( + &mut self, + document: PenDocument, + origin: EditOrigin, + ) -> Result { + CollaborationEditorHost::install_collaboration_document(self.editor, document, origin) + } +} + +impl CollabHost for DaemonCollabHost<'_> { + fn mark_editor_state_dirty(&mut self) { + *self.dirty = true; + } + + fn enable_collaboration_ids(&mut self, namespace: PeerNamespace) -> Result<(), IdAllocError> { + *self.id_allocator = Some(DocumentIdAllocator::namespaced_for_document( + &self.editor.doc, + namespace, + )?); + Ok(()) + } + + fn disable_collaboration_ids(&mut self) { + *self.id_allocator = None; + if let Ok(next) = next_sequential_counter(&self.editor.doc) { + *self.next_node_id = (*self.next_node_id).max(next); + } + } +} + +/// Why the daemon refused a mutation. +/// +/// The bool acks the older applier paths returned could not say *why* a write +/// was dropped, so an MCP or AI caller saw a silent no-op during an active +/// session. Every gateway rejection is now a typed reason the route can render. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DaemonMutationRefusal { + /// The collaboration policy refused this source in this phase. + Collab(CollabGateReason), +} + +impl DaemonMutationRefusal { + /// Stable machine-readable code for a REST body or JSON-RPC error. + pub const fn code(self) -> &'static str { + match self { + Self::Collab(reason) => match reason { + CollabGateReason::ReadOnly | CollabGateReason::OwnerOnlySave => "collab-readonly", + CollabGateReason::PendingEdit | CollabGateReason::SessionTransition => { + "collab-busy" + } + _ => "collab-active", + }, + } + } + + /// HTTP status the code maps to. Every refusal is a conflict with the live + /// session rather than a malformed request. + pub const fn http_status(self) -> &'static str { + "409 Conflict" + } +} + +impl std::fmt::Display for DaemonMutationRefusal { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Collab(reason) => write!( + f, + "refused by the live collaboration session ({})", + reason.i18n_key() + ), + } + } +} + +impl std::error::Error for DaemonMutationRefusal {} + +impl WebCanvasState { + /// Borrow the runtime and a host over the same editor. + /// + /// One method rather than two accessors because the borrow checker has to + /// see both borrows split out of `self` at once. + pub(crate) fn collab_runtime_and_host(&mut self) -> (&mut CollabRuntime, DaemonCollabHost<'_>) { + let collab = &mut self.collab; + ( + &mut collab.runtime, + DaemonCollabHost { + editor: &mut self.editor, + id_allocator: &mut collab.id_allocator, + next_node_id: &mut collab.next_node_id, + dirty: &mut collab.dirty, + }, + ) + } + + /// The collaboration policy derived from the live projection. + pub(crate) fn collab_policy(&self) -> CollabGatePolicy { + CollabGatePolicy::from(&self.editor.editor_ui.collab) + } + + /// The single gate every non-user daemon mutation passes through. + /// + /// MCP writes, AI writes, and bulk imports all reach the same editor from + /// different routes. Desktop refuses those while a session is active — the + /// M1 protocol has no way to sequence a write the local user did not make — + /// and this keeps the daemon's answer identical instead of letting a route + /// that predates collaboration fork the document. + /// + /// Idle and Discovering phases pass through untouched, so a daemon with no + /// session behaves exactly as it did before collaboration existed. + pub(crate) fn gate_daemon_mutation( + &self, + action: CollabGateAction, + source: CollabEditSource, + ) -> Result<(), DaemonMutationRefusal> { + self.collab_policy() + .check(action, source) + .map_err(DaemonMutationRefusal::Collab) + } + + /// Apply one editor command through the gateway. + /// + /// The bool is the editor's own "did this command do anything" ack, kept + /// so the applier contract is unchanged; a refusal is the `Err`, which is + /// what makes a rejected write distinguishable from a no-op write. + pub(crate) fn apply_gated( + &mut self, + command: op_editor_core::EditorCommand, + source: CollabEditSource, + ) -> Result { + self.collab_policy() + .check_command(&command, source) + .map_err(DaemonMutationRefusal::Collab)?; + Ok(self.editor.apply(command)) + } + + /// Whether a live session currently sequences this document. + pub(crate) fn collab_session_is_active(&self) -> bool { + self.editor.editor_ui.collab.phase == op_editor_core::CollabConnectionPhase::Active + } + + /// Install a validated document while a session is live. + /// + /// The whole point is that the swap happens *inside* the runtime's + /// local-edit capture: the session core diffs the document before and after + /// and emits a precise node transaction, so an unchanged push produces no + /// transaction at all. [`EditorState::install_prepared_document`] keeps the + /// document generation, which is what lets the matching finish succeed. + /// + /// Returns `false` when no capture could be opened (busy or read-only); the + /// caller then reports a conflict rather than writing behind the session's + /// back. + pub(crate) fn ingest_document_in_session(&mut self, prepared: PreparedDocument) -> bool { + let (runtime, mut host) = self.collab_runtime_and_host(); + if !runtime.begin_local_edit(&mut host) { + return false; + } + // From here the capture is open and MUST be closed on every path; + // `finish_local_edit` is the only close, so nothing fallible may run + // between the two calls. + host.editor_state_mut() + .install_prepared_document(prepared, EditOrigin::Local); + let (runtime, mut host) = self.collab_runtime_and_host(); + runtime.finish_local_edit(&mut host); + true + } +} + +#[cfg(test)] +#[path = "collab_state_tests.rs"] +mod tests; diff --git a/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs b/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs new file mode 100644 index 000000000..559dfd97c --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs @@ -0,0 +1,246 @@ +//! Gateway + split-borrow coverage. +//! +//! These drive the *projection* into each phase directly rather than standing +//! up a real session: the gate reads only `phase` / `role` / `pending_edit`, so +//! a projected state exercises exactly the decision under test without a +//! network peer. + +use op_editor_core::{ + AuthenticatedCollabSession, CollabConnectionPhase, CollabDocumentMutation, CollabEditSource, + CollabGateAction, CollabPendingEditUi, CollabUiRole, EditorCommand, EditorState, +}; + +use super::super::WebCanvasState; + +fn daemon() -> WebCanvasState { + WebCanvasState::new(EditorState::starter(), 0) +} + +fn in_session(state: &mut WebCanvasState, phase: CollabConnectionPhase, role: CollabUiRole) { + state.editor.editor_ui.collab.set_authenticated_session( + phase, + AuthenticatedCollabSession { + session_name: "studio".into(), + role, + share_endpoint: None, + }, + Vec::new(), + ); +} + +const DOCUMENT_EDIT: CollabGateAction = + CollabGateAction::Document(CollabDocumentMutation::NodePropertyBatch); + +#[test] +fn an_idle_daemon_passes_every_source_through_untouched() { + let state = daemon(); + for source in [ + CollabEditSource::User, + CollabEditSource::Ai, + CollabEditSource::Mcp, + CollabEditSource::Import, + CollabEditSource::ExternalSync, + ] { + assert!( + state.gate_daemon_mutation(DOCUMENT_EDIT, source).is_ok(), + "{source:?} must be unaffected when there is no session" + ); + assert!(state + .gate_daemon_mutation(CollabGateAction::ReplaceDocument, source) + .is_ok()); + } + assert!(!state.collab_session_is_active()); +} + +#[test] +fn an_active_owner_may_write_but_ai_and_mcp_may_not() { + let mut state = daemon(); + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Owner, + ); + assert!(state.collab_session_is_active()); + + assert!(state + .gate_daemon_mutation(DOCUMENT_EDIT, CollabEditSource::User) + .is_ok()); + for source in [CollabEditSource::Ai, CollabEditSource::Mcp] { + let refusal = state + .gate_daemon_mutation(DOCUMENT_EDIT, source) + .expect_err("desktop refuses these during a session, so the daemon must too"); + assert_eq!(refusal.code(), "collab-active"); + assert_eq!(refusal.http_status(), "409 Conflict"); + } +} + +#[test] +fn a_viewer_is_read_only_for_its_own_writes() { + let mut state = daemon(); + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Viewer, + ); + let refusal = state + .gate_daemon_mutation(DOCUMENT_EDIT, CollabEditSource::User) + .expect_err("a viewer cannot write"); + assert_eq!(refusal.code(), "collab-readonly"); +} + +#[test] +fn a_frozen_phase_reports_busy_rather_than_read_only() { + for phase in [ + CollabConnectionPhase::Starting, + CollabConnectionPhase::Joining, + CollabConnectionPhase::Authenticating, + ] { + let mut state = daemon(); + state.editor.editor_ui.collab.set_phase(phase); + let refusal = state + .gate_daemon_mutation(DOCUMENT_EDIT, CollabEditSource::User) + .expect_err("a transitioning session accepts no writes"); + assert_eq!(refusal.code(), "collab-busy", "{phase:?}"); + } +} + +#[test] +fn a_pending_edit_reports_busy() { + let mut state = daemon(); + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Owner, + ); + state.editor.editor_ui.collab.pending_edit = CollabPendingEditUi::Submitting; + let refusal = state + .gate_daemon_mutation(DOCUMENT_EDIT, CollabEditSource::User) + .expect_err("one edit at a time"); + assert_eq!(refusal.code(), "collab-busy"); +} + +#[test] +fn replacing_the_whole_document_is_refused_for_every_source_in_a_session() { + let mut state = daemon(); + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Owner, + ); + for source in [ + CollabEditSource::User, + CollabEditSource::Ai, + CollabEditSource::Mcp, + CollabEditSource::ExternalSync, + ] { + let refusal = state + .gate_daemon_mutation(CollabGateAction::ReplaceDocument, source) + .expect_err("a whole-document swap cannot be sequenced"); + assert_eq!(refusal.code(), "collab-active", "{source:?}"); + } +} + +#[test] +fn apply_gated_separates_a_refusal_from_a_no_op() { + let mut state = daemon(); + // No session: the editor's own ack rides through untouched, whatever it is. + // What matters is that it arrives as `Ok`, so the caller can still tell a + // refusal from whatever the editor decided. + assert!( + state + .apply_gated(EditorCommand::ClearSelection, CollabEditSource::Mcp) + .is_ok(), + "no session refuses nothing" + ); + + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Owner, + ); + let insert = EditorCommand::InsertNode { + kind: "rect".into(), + name: "From MCP".into(), + x: 0, + y: 0, + width: 10, + height: 10, + fill_hex: None, + target_parent: op_editor_core::NodeId::NONE, + page_id: None, + }; + let refusal = state + .apply_gated(insert, CollabEditSource::Mcp) + .expect_err("an MCP write during a session must be a typed refusal"); + assert_eq!(refusal.code(), "collab-active"); +} + +#[test] +fn the_split_borrow_yields_a_host_over_the_same_editor() { + use op_editor_host_core::collab::CollaborationEditorHost; + + let mut state = daemon(); + state.editor.editor_ui.preserve_authored_geometry = true; + + let (_runtime, mut host) = state.collab_runtime_and_host(); + assert!(host.editor_state().editor_ui.preserve_authored_geometry); + host.editor_state_mut().editor_ui.preserve_authored_geometry = false; + + assert!( + !state.editor.editor_ui.preserve_authored_geometry, + "the host must borrow the daemon's editor, not a copy of it" + ); +} + +#[test] +fn the_host_seam_raises_the_dirty_flag_the_driver_reads() { + use op_collab_host::CollabHost; + + let mut state = daemon(); + assert!(!state.collab.take_dirty()); + + let (_runtime, mut host) = state.collab_runtime_and_host(); + host.mark_editor_state_dirty(); + + assert!(state.collab.take_dirty()); + assert!(!state.collab.take_dirty(), "taking the flag clears it"); +} + +#[test] +fn ingest_refuses_rather_than_writing_when_no_capture_can_open() { + let mut state = daemon(); + // A projected Active phase with no real session actor: the runtime cannot + // open a capture, and the ingest must report that instead of writing + // behind the session's back. + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Owner, + ); + let before = state.editor.doc.clone(); + + let prepared = op_editor_core::PreparedDocument::prepare( + serde_json::from_value(serde_json::json!({ + "version": "1.0", + "children": [{"type": "rectangle", "id": "n1", "x": 0, "y": 0, "width": 5, "height": 5}] + })) + .expect("valid document"), + ) + .expect("validates"); + + assert!(!state.ingest_document_in_session(prepared)); + assert_eq!(state.editor.doc, before, "a refused ingest writes nothing"); +} + +#[test] +fn the_projection_sequence_is_independent_of_the_document_version() { + let mut state = daemon(); + assert_eq!(state.sse_tick().version, 0); + assert_eq!(state.sse_tick().collab_seq, 0); + + // A presence-shaped change moves only `collabSeq`, so a client polling + // `version` for document content is not made to refetch the document. + state.collab.bump_seq(); + assert_eq!(state.sse_tick().version, 0); + assert_eq!(state.sse_tick().collab_seq, 1); +} diff --git a/crates/op-host-services/src/web_canvas_server/connection.rs b/crates/op-host-services/src/web_canvas_server/connection.rs index 5f1764d78..bb8d52eb1 100644 --- a/crates/op-host-services/src/web_canvas_server/connection.rs +++ b/crates/op-host-services/src/web_canvas_server/connection.rs @@ -144,7 +144,7 @@ pub(super) fn serve_one( // monotonic). The state lock is released before the long SSE wait. if req.method == "GET" && req.path == "/api/mcp/events" { let rx = hub.subscribe(); - let current = state.lock().unwrap_or_else(|p| p.into_inner()).version; + let current = state.lock().unwrap_or_else(|p| p.into_inner()).sse_tick(); return serve_sse(stream, rx, current, cors_origin).map(|()| false); } // AI proxy stream: the browser bundle POSTs a model request and we @@ -325,7 +325,7 @@ pub(super) fn serve_one( // N-1). `broadcast` only sends to unbounded channels (non-blocking), // so the lock is held briefly. Lock order is always state→hub. if guard.version != before { - hub.broadcast(guard.version); + hub.broadcast(guard.sse_tick()); } reply }; @@ -407,6 +407,14 @@ pub(super) fn serve_one( let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); let before = guard.version; let mut applied_any = false; + // The gateway. Snapshotted before dispatch because the applier only + // borrows the editor, and `Copy` + the held state lock make the + // snapshot exact for the whole message. Desktop refuses MCP document + // mutations during a live session (`mcp_live.rs`); the daemon has to + // give the same answer or an MCP client would fork the shared document + // out from under the peers. + let policy = guard.collab_policy(); + let mut refused: Option = None; // Mechanical passthrough only — this daemon (`--serve-web`/`op // start`) is a SEPARATE request loop from `mcp_live.rs`'s // `McpLiveServer` (desktop `--live-mcp`), not the same struct; @@ -418,19 +426,35 @@ pub(super) fn serve_one( &mut guard.editor, &req.body, |_tool_name, editor, cmd| { + if let Err(reason) = + policy.check_command(cmd, op_editor_core::CollabEditSource::Mcp) + { + refused = Some(reason); + return false; + } let ok = editor.apply(cmd.clone()); applied_any |= ok; ok }, )? .unwrap_or_default(); + if let Some(reason) = refused { + // Same surfacing as the desktop live-MCP path: the tool acks + // "not applied" and the panel explains why, so a refusal is never + // a silent no-op. + guard.editor.editor_ui.collab.set_notice( + reason.notice_kind(), + crate::design_agent_tools::reveal_now_millis(), + ); + guard.collab.bump_seq(); + } if applied_any { guard.version += 1; } // Atomic bump+broadcast under the state lock (see the REST path) so SSE // version events stay monotonic across concurrent mutations. - if guard.version != before { - hub.broadcast(guard.version); + if guard.version != before || refused.is_some() { + hub.broadcast(guard.sse_tick()); } response }; @@ -444,15 +468,15 @@ pub(super) fn serve_one( } /// Stream Server-Sent Events to a subscribed client: write the SSE headers, -/// emit the current version immediately (initial sync), then forward each -/// version bump from `rx` as a `data: {"version":N}` event. A periodic +/// emit the current tick immediately (initial sync), then forward each +/// bump from `rx` as a `data: {"version":N,"collabSeq":M}` event. A periodic /// heartbeat comment keeps the connection alive AND detects a disconnected /// client (the write fails once the socket is gone). Returns when the client /// disconnects (write error) or the hub is dropped. pub(super) fn serve_sse( stream: &mut S, - rx: Receiver, - current_version: u64, + rx: Receiver, + current: SseTick, cors_origin: Option<&str>, ) -> Result<()> { let cors_line = cors_origin @@ -468,18 +492,18 @@ pub(super) fn serve_sse( stream .write_all(headers.as_bytes()) .map_err(|e| WebCanvasError::Transport(format!("sse headers: {e}")))?; - write_sse_event(stream, current_version)?; + write_sse_event(stream, current)?; loop { match rx.recv_timeout(SSE_HEARTBEAT) { - Ok(mut version) => { - // Coalesce any further queued bumps — only the latest version - // matters (the client re-fetches the whole document on it), so - // a burst of mutations collapses to a single event and the + Ok(mut tick) => { + // Coalesce any further queued bumps — only the latest tick + // matters (the client re-reads whatever the counters point at), + // so a burst of mutations collapses to a single event and the // channel can't accumulate unboundedly behind a slow client. while let Ok(next) = rx.try_recv() { - version = next; + tick = next; } - write_sse_event(stream, version)?; + write_sse_event(stream, tick)?; } Err(RecvTimeoutError::Timeout) => { // SSE comment heartbeat — no-op for the client, but a failed @@ -496,9 +520,15 @@ pub(super) fn serve_sse( } } -/// Format + write one SSE `data:` event carrying the document version. -pub(super) fn write_sse_event(stream: &mut S, version: u64) -> Result<()> { - let event = format!("data: {{\"version\":{version}}}\n\n"); +/// Format + write one SSE `data:` event carrying both sequence numbers. +/// +/// `version` stays the first field and keeps its exact spelling, so a client +/// written against the original payload keeps parsing this one. +pub(super) fn write_sse_event(stream: &mut S, tick: SseTick) -> Result<()> { + let event = format!( + "data: {{\"version\":{},\"collabSeq\":{}}}\n\n", + tick.version, tick.collab_seq + ); stream .write_all(event.as_bytes()) .map_err(|e| WebCanvasError::Transport(format!("sse write: {e}")))?; diff --git a/crates/op-host-services/src/web_canvas_server/doc_routes.rs b/crates/op-host-services/src/web_canvas_server/doc_routes.rs index db0f0a00c..a4423c52b 100644 --- a/crates/op-host-services/src/web_canvas_server/doc_routes.rs +++ b/crates/op-host-services/src/web_canvas_server/doc_routes.rs @@ -7,6 +7,23 @@ use super::*; pub(super) fn open_recent_file(body: &str, state: &mut WebCanvasState) -> WebReply { + // Swapping the open document out from under a live session would leave the + // peers editing a document this daemon no longer has. Refused for every + // source once a session exists; a no-op when there is none. + if let Err(refusal) = state.gate_daemon_mutation( + op_editor_core::CollabGateAction::ReplaceDocument, + op_editor_core::CollabEditSource::ExternalSync, + ) { + return WebReply { + status: refusal.http_status(), + body: serde_json::json!({ + "ok": false, + "error": refusal.code(), + "message": refusal.to_string(), + }) + .to_string(), + }; + } let parsed: Option = serde_json::from_str(body).ok(); let Some(path_s) = parsed .as_ref() diff --git a/crates/op-host-services/src/web_canvas_server/origin_guard.rs b/crates/op-host-services/src/web_canvas_server/origin_guard.rs index f899d9471..0a4177412 100644 --- a/crates/op-host-services/src/web_canvas_server/origin_guard.rs +++ b/crates/op-host-services/src/web_canvas_server/origin_guard.rs @@ -12,6 +12,9 @@ pub(super) fn is_sensitive_browser_post(request: &crate::mcp_serve::HttpRequest) || request .path .starts_with(op_editor_core::auth_routes::API_PREFIX) + || request + .path + .starts_with(op_editor_core::collab_routes::API_PREFIX) || request.path.starts_with("/api/figma/")) } diff --git a/crates/op-host-services/src/web_canvas_server/run_loop.rs b/crates/op-host-services/src/web_canvas_server/run_loop.rs index 9a4a139e6..3216065ff 100644 --- a/crates/op-host-services/src/web_canvas_server/run_loop.rs +++ b/crates/op-host-services/src/web_canvas_server/run_loop.rs @@ -127,6 +127,9 @@ pub fn run_web_canvas(options: ServeWebOptions) -> Result<()> { guard.managed_token = managed_token; guard.allow_origins = allow_origins; } + // The collaboration pump. It observes `shutdown` itself, so it retires + // within one tick of the daemon being asked to stop. + collab_driver::spawn(&state, &hub, &shutdown); for stream in listener.incoming() { if shutdown.load(Ordering::Acquire) { break; diff --git a/crates/op-host-services/src/web_canvas_server_conn_tests.rs b/crates/op-host-services/src/web_canvas_server_conn_tests.rs index dec14e53c..ed4f8a771 100644 --- a/crates/op-host-services/src/web_canvas_server_conn_tests.rs +++ b/crates/op-host-services/src/web_canvas_server_conn_tests.rs @@ -74,14 +74,22 @@ fn mock_stream(request: &str) -> MockStream { } } +/// One SSE payload for the hub tests. +fn tick(version: u64, collab_seq: u64) -> SseTick { + SseTick { + version, + collab_seq, + } +} + #[test] fn sse_hub_broadcasts_version_to_all_subscribers() { let hub = SseHub::default(); let a = hub.subscribe(); let b = hub.subscribe(); - hub.broadcast(5); - assert_eq!(a.recv().unwrap(), 5); - assert_eq!(b.recv().unwrap(), 5); + hub.broadcast(tick(5, 0)); + assert_eq!(a.recv().unwrap(), tick(5, 0)); + assert_eq!(b.recv().unwrap(), tick(5, 0)); } #[test] @@ -90,32 +98,46 @@ fn sse_hub_prunes_disconnected_subscribers() { let live = hub.subscribe(); drop(hub.subscribe()); // a disconnected client (receiver dropped) assert_eq!(hub.subscriber_count(), 2); - hub.broadcast(1); // prunes the dropped one + hub.broadcast(tick(1, 0)); // prunes the dropped one assert_eq!(hub.subscriber_count(), 1); - assert_eq!(live.recv().unwrap(), 1); + assert_eq!(live.recv().unwrap(), tick(1, 0)); } #[test] fn write_sse_event_emits_data_frame() { let mut stream = mock_stream(""); - write_sse_event(&mut stream, 42).expect("write"); + write_sse_event(&mut stream, tick(42, 7)).expect("write"); assert_eq!( String::from_utf8_lossy(&stream.output), - "data: {\"version\":42}\n\n" + "data: {\"version\":42,\"collabSeq\":7}\n\n" ); } +#[test] +fn sse_payload_stays_a_superset_of_the_original_version_frame() { + let mut stream = mock_stream(""); + write_sse_event(&mut stream, tick(3, 0)).expect("write"); + let out = String::from_utf8_lossy(&stream.output).into_owned(); + // A client written against the original `{"version":N}` frame parses this + // one unchanged: `version` keeps its spelling and stays the first field. + assert!(out.starts_with("data: {\"version\":3,"), "{out}"); + let payload: serde_json::Value = + serde_json::from_str(out.trim_start_matches("data: ").trim()).expect("valid JSON"); + assert_eq!(payload["version"], 3); + assert_eq!(payload["collabSeq"], 0); +} + #[test] fn serve_sse_emits_initial_then_each_version_until_hub_drops() { let (tx, rx) = mpsc::channel(); - tx.send(9).expect("send"); // one bump, then the sender drops → Disconnected + tx.send(tick(9, 0)).expect("send"); // one bump, then the sender drops → Disconnected drop(tx); let mut stream = mock_stream(""); - serve_sse(&mut stream, rx, 7, Some("*")).expect("serve_sse"); + serve_sse(&mut stream, rx, tick(7, 0), Some("*")).expect("serve_sse"); let out = String::from_utf8_lossy(&stream.output); assert!(out.contains("text/event-stream"), "{out}"); - assert!(out.contains(r#"data: {"version":7}"#), "{out}"); // initial sync - assert!(out.contains(r#"data: {"version":9}"#), "{out}"); // broadcast bump + assert!(out.contains(r#"data: {"version":7,"#), "{out}"); // initial sync + assert!(out.contains(r#"data: {"version":9,"#), "{out}"); // broadcast bump } #[test] @@ -131,7 +153,7 @@ fn serve_one_post_document_broadcasts_new_version_to_sse() { let mut stream = mock_stream(&request); serve_one(&mut stream, &state, &hub).expect("serve_one"); // The whole-doc sync bumped the version to 1 and broadcast it. - assert_eq!(sub.recv().unwrap(), 1); + assert_eq!(sub.recv().unwrap().version, 1); } #[test] diff --git a/crates/op-host-services/src/web_canvas_server_error.rs b/crates/op-host-services/src/web_canvas_server_error.rs index 50d04719c..5dde87adb 100644 --- a/crates/op-host-services/src/web_canvas_server_error.rs +++ b/crates/op-host-services/src/web_canvas_server_error.rs @@ -55,6 +55,10 @@ pub enum WebCanvasError { /// SSE stream). Never becomes an HTTP response — the socket is already /// unusable; the accept loop just logs it. Transport(String), + /// A live collaboration session refuses this write. Not a client fault and + /// not a daemon fault — the document is healthy and the request was + /// well-formed; it simply cannot be sequenced right now. + Collab(crate::web_canvas_server::DaemonMutationRefusal), } impl WebCanvasError { @@ -73,6 +77,19 @@ impl WebCanvasError { | WebCanvasError::Export(_) | WebCanvasError::Io(_) => "400 Bad Request", WebCanvasError::Config(_) | WebCanvasError::Transport(_) => "500 Internal Server Error", + WebCanvasError::Collab(refusal) => refusal.http_status(), + } + } + + /// Stable machine-readable code, when the failure has one. + /// + /// Only collaboration refusals carry a code today: a client has to tell + /// "the session is read-only for you" from "the session is busy" to decide + /// whether retrying is worth it. + pub fn error_code(&self) -> Option<&'static str> { + match self { + WebCanvasError::Collab(refusal) => Some(refusal.code()), + _ => None, } } } @@ -86,6 +103,7 @@ impl fmt::Display for WebCanvasError { | WebCanvasError::Io(m) | WebCanvasError::Config(m) | WebCanvasError::Transport(m) => f.write_str(m), + WebCanvasError::Collab(refusal) => write!(f, "{refusal}"), } } } diff --git a/crates/op-host-services/src/web_canvas_server_export_tests.rs b/crates/op-host-services/src/web_canvas_server_export_tests.rs index 78e5dd5fe..735235564 100644 --- a/crates/op-host-services/src/web_canvas_server_export_tests.rs +++ b/crates/op-host-services/src/web_canvas_server_export_tests.rs @@ -306,11 +306,13 @@ fn get_version_is_a_cheap_change_probe() { let mut s = fresh_state(); let r = handle_web_canvas_request("GET", "/api/mcp/version", "", &mut s); assert!(r.status.starts_with("200")); - assert_eq!(r.body, r#"{"version":0}"#); + // `collabSeq` rides along so one poll covers both document and + // collaboration changes; `version` keeps its exact spelling and position. + assert_eq!(r.body, r#"{"version":0,"collabSeq":0}"#); // A document mutation bumps the probed version. let _ = handle_web_canvas_request("POST", "/api/mcp/document", SYNC_BODY, &mut s); let r2 = handle_web_canvas_request("GET", "/api/mcp/version", "", &mut s); - assert_eq!(r2.body, r#"{"version":1}"#); + assert_eq!(r2.body, r#"{"version":1,"collabSeq":0}"#); } #[test] diff --git a/crates/op-host-services/src/web_chat_standard.rs b/crates/op-host-services/src/web_chat_standard.rs index 4e13654d2..e1e330c8d 100644 --- a/crates/op-host-services/src/web_chat_standard.rs +++ b/crates/op-host-services/src/web_chat_standard.rs @@ -182,17 +182,29 @@ pub fn stream_standard_turn( op_orchestrator::Intent::Design ) && clear_fresh_starter_frame_for_design(&mut snapshot) { - let version = { + let tick = { let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); - if let Some(version) = clear_live_starter_frame_for_design(&mut guard) { + // Through the gateway like every other daemon write: during a live + // session this housekeeping edit would be an unsequenced AI write. + // Skipping it only means the starter frame stays, which is strictly + // better than forking the shared document. + let gated = guard + .gate_daemon_mutation( + op_editor_core::CollabGateAction::Document( + op_editor_core::CollabDocumentMutation::NodeDelete, + ), + op_editor_core::CollabEditSource::Ai, + ) + .is_ok(); + if gated && clear_live_starter_frame_for_design(&mut guard).is_some() { snapshot = guard.editor.clone(); - Some(version) + Some(guard.sse_tick()) } else { None } }; - if let Some(version) = version { - hub.broadcast(version); + if let Some(tick) = tick { + hub.broadcast(tick); } } inject_transient_builtin(&mut snapshot, req.transient_builtin.as_ref()); @@ -251,7 +263,7 @@ fn apply_request_snapshot( state: &Mutex, hub: &SseHub, ) -> Result { - let mut broadcast_version = None; + let mut broadcast_tick = None; let mut snapshot = { let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); if let Some(agent) = req.transient_builtin.as_ref() { @@ -269,8 +281,18 @@ fn apply_request_snapshot( let loaded = op_pen_loader::load_canonical(doc_json) .map_err(|e| WebChatStandardError::Document(e.to_string()))?; if guard.editor.doc != loaded.value { - let version = guard.replace_document(loaded.value); - broadcast_version = Some(version); + // A whole-document swap during a live session is exactly what + // the collaboration protocol cannot sequence, so the gateway + // refuses it here rather than letting the AI route silently + // replace what peers are editing. + guard + .gate_daemon_mutation( + op_editor_core::CollabGateAction::ReplaceDocument, + op_editor_core::CollabEditSource::Ai, + ) + .map_err(WebChatStandardError::CollabRefused)?; + guard.replace_document(loaded.value); + broadcast_tick = Some(guard.sse_tick()); } } if let Some(meta) = req.editor_meta.clone() { @@ -300,8 +322,8 @@ fn apply_request_snapshot( } guard.editor.clone() }; - if let Some(version) = broadcast_version { - hub.broadcast(version); + if let Some(tick) = broadcast_tick { + hub.broadcast(tick); } inject_transient_builtin(&mut snapshot, req.transient_builtin.as_ref()); Ok(snapshot) @@ -422,23 +444,37 @@ fn stream_modify_route( let nodes = crate::chat_intent::parse_modify_nodes(&full_response); if !nodes.is_empty() { write_delta_event(out, &format!("\n{full_response}"))?; - let (applied, version) = { + let (applied, tick) = { let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); - let (count, mutated) = crate::chat_canvas_tools::apply_design_modification( - &mut guard.editor, - &nodes, - &target_frame_ids, - ); - let version = if mutated { - guard.version += 1; - Some(guard.version) + // `apply_design_modification` writes a batch straight into the + // editor, so the gate runs before it rather than per command. + if guard + .gate_daemon_mutation( + op_editor_core::CollabGateAction::Document( + op_editor_core::CollabDocumentMutation::NodePropertyBatch, + ), + op_editor_core::CollabEditSource::Ai, + ) + .is_err() + { + (0, None) } else { - None - }; - (count, version) + let (count, mutated) = crate::chat_canvas_tools::apply_design_modification( + &mut guard.editor, + &nodes, + &target_frame_ids, + ); + let tick = if mutated { + guard.version += 1; + Some(guard.sse_tick()) + } else { + None + }; + (count, tick) + } }; - if let Some(version) = version { - hub.broadcast(version); + if let Some(tick) = tick { + hub.broadcast(tick); } if applied > 0 { write_delta_event(out, "\n\n")?; @@ -600,25 +636,30 @@ impl DocSink for WebDesignDocSink<'_> { } fn apply(&mut self, cmd: EditorCommand) -> bool { - let (applied, version, snapshot) = { + let (applied, tick, snapshot) = { let mut guard = self.state.lock().unwrap_or_else(|p| p.into_inner()); - let applied = guard.editor.apply(cmd); - let version = if applied { + // A refusal and a no-op both ack `false` to the generator, which is + // the existing contract; the difference is visible in the session + // notice the gate raises, not in this return value. + let applied = guard + .apply_gated(cmd, op_editor_core::CollabEditSource::Ai) + .unwrap_or(false); + let tick = if applied { crate::design_session::fit_design_viewport_to_content( &mut guard.editor, 1440.0, 900.0, ); guard.version += 1; - Some(guard.version) + Some(guard.sse_tick()) } else { None }; - (applied, version, guard.editor.clone()) + (applied, tick, guard.editor.clone()) }; self.mirror = snapshot; - if let Some(version) = version { - self.hub.broadcast(version); + if let Some(tick) = tick { + self.hub.broadcast(tick); } applied } diff --git a/crates/op-host-services/src/web_chat_standard_error.rs b/crates/op-host-services/src/web_chat_standard_error.rs index bec442222..1739cf4ba 100644 --- a/crates/op-host-services/src/web_chat_standard_error.rs +++ b/crates/op-host-services/src/web_chat_standard_error.rs @@ -50,6 +50,11 @@ pub(crate) enum WebChatStandardError { NoModelConfigured, /// Provider resolution itself failed. Text is `ai_proxy`'s own message. ProviderResolve(String), + /// A live collaboration session refuses AI writes. The M1 protocol has no + /// way to sequence a write the local user did not make, so the desktop + /// refuses the same turn — this keeps the daemon's answer identical rather + /// than letting the AI route fork the shared document. + CollabRefused(crate::web_canvas_server::DaemonMutationRefusal), } impl fmt::Display for WebChatStandardError { @@ -62,6 +67,7 @@ impl fmt::Display for WebChatStandardError { "provider endpoint is not allowed: private, loopback, and reserved addresses require an OPENPENCIL_WEB_AI_ENDPOINT_ALLOWLIST entry", ), WebChatStandardError::NoModelConfigured => f.write_str("no model configured"), + WebChatStandardError::CollabRefused(refusal) => write!(f, "{refusal}"), WebChatStandardError::EndpointRejected(message) | WebChatStandardError::Document(message) | WebChatStandardError::ProviderResolve(message) => f.write_str(message), diff --git a/crates/op-host-services/src/web_chat_standard_tests.rs b/crates/op-host-services/src/web_chat_standard_tests.rs index 5b16926d1..3f5863a46 100644 --- a/crates/op-host-services/src/web_chat_standard_tests.rs +++ b/crates/op-host-services/src/web_chat_standard_tests.rs @@ -305,7 +305,7 @@ fn design_doc_sink_applies_and_bumps_version() { })); assert_eq!(state.lock().unwrap().version, 1); - assert_eq!(sub.recv().unwrap(), 1); + assert_eq!(sub.recv().unwrap().version, 1); assert_eq!(sink.state().active_children().len(), 1); }