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

@@ -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",

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));
+        }
+    }
+}

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<CollabDiscoveredWire>,
+}
+
+/// 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<String>,
+    pub invite: Option<String>,
+    pub connection: Option<CollabConnectionPathWire>,
+}
+
+/// 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<CollabSessionWire>,
+    pub participants: Vec<CollabParticipantWire>,
+    pub presence: Vec<CollabPresenceWire>,
+    pub admissions: Vec<CollabAdmissionWire>,
+    pub owner_confirmation: Option<CollabOwnerConfirmationWire>,
+    pub notice: Option<CollabNoticeWire>,
+    pub pending_edit: CollabPendingEditWire,
+    pub discarded_edit: Option<CollabDiscardedEditWire>,
+}
+
+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;

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<CollabUiAction, CollabActionWireError> {
+        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, CollabActionWireError> {
+    CollabAdmissionRequestKey::new(value).ok_or(CollabActionWireError::InvalidRequestKey)
+}
+
+fn checked_address(value: String) -> Result<String, CollabActionWireError> {
+    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::<CollabActionWire>(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:?}");
+        }
+    }
+}

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<CollabConnectionPathUi> 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<CollabConnectionPathWire> 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<CollabNoticeKind> 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<CollabNotice> 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<CollabPointWire>,
+    pub selection: Vec<String>,
+    pub editing_node: Option<String>,
+    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<CollabPointWire>,
+    /// 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<String>,
+}
+
+/// 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<CollabRoleWire>,
+}
+
+/// 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<String>,
+    pub avatar_url: Option<String>,
+}
+
+/// 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<String>,
+}
+
+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,
+}

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}");
+}

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<Self, DocumentInstallError> {
+        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<DocumentInstallReport, DocumentInstallError> {
-        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 {

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")
+        }
+    }
+}

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,
 };

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"

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<Output = ()> + '_)>) {
+        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));
+}

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;

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<Vec<mpsc::Sender<u64>>>,
+    subscribers: Mutex<Vec<mpsc::Sender<SseTick>>>,
 }
 
 impl SseHub {
     /// Register a subscriber; the SSE connection thread blocks on the returned
-    /// receiver for version bumps.
-    pub(crate) fn subscribe(&self) -> Receiver<u64> {
+    /// receiver for ticks.
+    pub(crate) fn subscribe(&self) -> Receiver<SseTick> {
         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<u64>,
+    /// 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::*;

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<Mutex<WebCanvasState>>,
+    hub: &Arc<SseHub>,
+    shutdown: &Arc<AtomicBool>,
+) {
+    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<WebCanvasState>, 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<WebCanvasState>, 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<WebCanvasState> {
+        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"
+        );
+    }
+}

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;

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}");
+    }
+}

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<DocumentIdAllocator>,
+    /// 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<std::sync::Arc<super::collab_driver::CollabWaker>>,
+}
+
+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<super::collab_driver::CollabWaker>) {
+        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<DocumentIdAllocator>,
+    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<DocumentInstallReport, DocumentInstallError> {
+        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<bool, DaemonMutationRefusal> {
+        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;

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);
+}

crates/op-host-services/src/web_canvas_server/connection.rs

@@ -144,7 +144,7 @@ pub(super) fn serve_one<S: Read + Write>(
     // 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<S: Read + Write>(
             // 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<S: Read + Write>(
         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<op_editor_core::CollabGateReason> = 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<S: Read + Write>(
             &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<S: Read + Write>(
 }
 
 /// 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<S: Write>(
     stream: &mut S,
-    rx: Receiver<u64>,
-    current_version: u64,
+    rx: Receiver<SseTick>,
+    current: SseTick,
     cors_origin: Option<&str>,
 ) -> Result<()> {
     let cors_line = cors_origin
@@ -468,18 +492,18 @@ pub(super) fn serve_sse<S: Write>(
     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<S: Write>(
     }
 }
 
-/// Format + write one SSE `data:` event carrying the document version.
-pub(super) fn write_sse_event<S: Write>(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<S: Write>(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}")))?;

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::Value> = serde_json::from_str(body).ok();
     let Some(path_s) = parsed
         .as_ref()

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/"))
 }
 

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;

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]

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}"),
         }
     }
 }

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]

crates/op-host-services/src/web_chat_standard.rs

@@ -182,17 +182,29 @@ pub fn stream_standard_turn<W: Write>(
         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<WebCanvasState>,
     hub: &SseHub,
 ) -> Result<EditorState, WebChatStandardError> {
-    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<W: Write>(
     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<!-- APPLIED -->")?;
@@ -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
     }

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),

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