Refactor server state

openusdconnect/adapters.py

@@ -394,6 +394,7 @@ def set_connectable_input(
         ``info_id`` is the UsdShade ``info:id`` (Sdr identifier) for
         ``UsdShade.Shader`` prims; empty string for non-shader connectables.
         ``time`` selects a USD time sample; ``None`` writes the static opinion.
+        Numeric array values use Python lists on the receive path.
         """
         raise NotImplementedError
 

openusdconnect/codec.py

@@ -1667,6 +1667,7 @@ def message_to_dict(buf: bytes | bytearray, *, numpy_arrays: bool = False) -> di
 
     With *numpy_arrays*, geometry arrays remain zero-copy views into the
     FlatBuffer. Otherwise they become lists for JSON-safe compatibility.
+    Connectable input values retain Python lists in either mode.
     """
     envelope = decode_envelope(buf)
     msg_type, obj = resolve_payload(envelope)
@@ -1850,7 +1851,7 @@ def decode_broadcast_event(
     numpy_arrays: bool = False,
 ) -> dict:
     """Convert a broadcast to the public dictionary representation."""
-    record = _decode_received_event(broadcast, numpy_arrays=numpy_arrays)
+    record = decode_received_event(broadcast, numpy_arrays=numpy_arrays)
     message = {"type": MSG_EVENT, "seq": record.seq, "event": record.event}
     for name, value in (
         ("origin", record.origin),
@@ -1875,10 +1876,10 @@ class ReceivedEvent:
     client: str | None = None
 
 
-def _decode_received_event(
+def decode_received_event(
     broadcast: BroadcastEvent,
     *,
-    numpy_arrays: bool,
+    numpy_arrays: bool = False,
 ) -> ReceivedEvent:
     """Read routing directly and convert only the event needed by USD/adapters.
 
@@ -1982,7 +1983,7 @@ def decode_messages(
                 continue
 
             if msg_type == MSG_EVENT:
-                record = _decode_received_event(payload, numpy_arrays=numpy_arrays)
+                record = decode_received_event(payload, numpy_arrays=numpy_arrays)
                 seq = record.seq
             elif msg_type == MSG_LAYER_GRAPH_STATE:
                 state = _dict_layer_graph_state(payload, msg_type)
@@ -2517,9 +2518,9 @@ def _dict_set_connectable_input(sci, kind):
         input_types[name] = _str(civ.TypeName())
         vt = civ.ValueType()
         if vt == ConnectableInputValueType.FloatArray:
-            inputs[name] = [civ.FloatArray(j) for j in range(civ.FloatArrayLength())]
+            inputs[name] = civ.FloatArrayAsNumpy().tolist() if civ.FloatArrayLength() else []
         elif vt == ConnectableInputValueType.IntArray:
-            inputs[name] = [civ.IntArray(j) for j in range(civ.IntArrayLength())]
+            inputs[name] = civ.IntArrayAsNumpy().tolist() if civ.IntArrayLength() else []
         elif vt == ConnectableInputValueType.StringArray:
             inputs[name] = [_str(civ.StringArray(j)) for j in range(civ.StringArrayLength())]
         elif vt == ConnectableInputValueType.ScalarString:

openusdconnect/server/_txn_barrier.py

@@ -1,21 +1,42 @@
-"""Read-write barrier for serializing compaction against in-flight transactions."""
+"""Read-write barrier separating history maintenance from admitted transactions."""
 
 from __future__ import annotations
 
 import threading
+from contextlib import contextmanager
 
 
 class _TxnBarrier:
-    """Read-write barrier: multiple shared (txn) holders, exclusive for compaction.
+    """Multiple shared transaction holders, exclusive history maintenance.
 
-    Lock ordering: _TxnBarrier is acquired BEFORE stage_lock / _seq_lock.
+    Acquire this barrier before EventJournal.commit_scope() and SceneState.lock.
+    Queued requests retain shared ownership until publication and checkpoint
+    capture finish; maintenance must exclude both queued and active requests.
     """
 
     def __init__(self):
         self._cond = threading.Condition(threading.Lock())
         self._readers = 0
         self._exclusive = False
 
+    @contextmanager
+    def shared(self):
+        """Scope a synchronous operation; queued work transfers ownership explicitly."""
+        self.acquire_shared()
+        try:
+            yield
+        finally:
+            self.release_shared()
+
+    @contextmanager
+    def exclusive(self):
+        """Wait for admitted transactions and exclude new ones until exit."""
+        self.acquire_exclusive()
+        try:
+            yield
+        finally:
+            self.release_exclusive()
+
     def acquire_shared(self):
         with self._cond:
             while self._exclusive:

openusdconnect/server/collaboration.py

@@ -0,0 +1,281 @@
+"""Department assignments and replay compatibility for a managed layer stack."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+
+from pxr import Sdf
+
+from .scene import SceneState
+from .types import ReplayModeConflictError
+
+LOG = logging.getLogger(__name__)
+_DEFAULT_LAYER_KEY = "default"
+_DEPARTMENT_LAYER_KEY_PREFIX = "department:"
+
+
+def _layer_key_for_department(department: str | None) -> str:
+    if not department:
+        return _DEFAULT_LAYER_KEY
+    return f"{_DEPARTMENT_LAYER_KEY_PREFIX}{department}"
+
+
+def department_for_layer_key(layer_key: str) -> str | None:
+    """Return department policy metadata for one OpenUSDConnect layer key."""
+    if layer_key == _DEFAULT_LAYER_KEY:
+        return None
+    if layer_key.startswith(_DEPARTMENT_LAYER_KEY_PREFIX):
+        department = layer_key[len(_DEPARTMENT_LAYER_KEY_PREFIX) :]
+        if department:
+            return department
+    return None
+
+
+def label_for_layer_key(layer_key: str) -> str:
+    department = department_for_layer_key(layer_key)
+    if department:
+        return department
+    if layer_key == _DEFAULT_LAYER_KEY:
+        return "Default"
+    return layer_key
+
+
+class CollaborationPolicy:
+    """Own policy state; the scene owns USD layers and their composition.
+
+    The scene lock protects assignments, replay admission, and layer lifecycle
+    together. Mutation methods release it before publishing changes.
+    """
+
+    def __init__(
+        self, scene: SceneState, *, department_priority: list[str] | None,
+        bump_snapshot_epoch: Callable[[str], None],
+        broadcast_layer_stack_state: Callable[[], None],
+    ):
+        self._scene = scene
+        self._client_layer_keys: dict[str, str] = {}
+        self.department_priority = self._validate_priority(department_priority or [])
+        self._flat_receiver_count = 0
+        self._bump_snapshot_epoch = bump_snapshot_epoch
+        self._broadcast_layer_stack_state = broadcast_layer_stack_state
+
+    @staticmethod
+    def _validate_priority(departments: list[str]) -> list[str]:
+        departments = list(departments)
+        if any(not department for department in departments):
+            raise ValueError("department names must be non-empty")
+        if len(set(departments)) != len(departments):
+            raise ValueError("department priority contains duplicates")
+        return departments
+
+    def restore_assignment(self, client_id: str, layer_key: str) -> None:
+        """Restore routing from the durable log before workers start."""
+        with self._scene.lock:
+            self._client_layer_keys[client_id] = layer_key
+
+    @property
+    def flat_receiver_count(self) -> int:
+        with self._scene.lock:
+            return self._flat_receiver_count
+
+    @property
+    def client_layers(self) -> dict[str, Sdf.Layer]:
+        """Return a snapshot of client assignments to shared collaboration layers."""
+        with self._scene.lock:
+            return {
+                client_id: self._scene.layer_stack.layer_for(layer_key)
+                for client_id, layer_key in self._client_layer_keys.items()
+            }
+
+    def get_or_create_client_layer(
+        self,
+        client_id: str,
+        department: str | None = None,
+    ) -> Sdf.Layer:
+        """Get or create a layer for this client.
+
+        With department: clients share a department layer (last-write-wins
+        within the department, department priority controls strength).
+        Without department: uses the shared edit_layer (weakest, last-write-wins).
+        """
+        layer_key = _layer_key_for_department(department)
+        added = False
+        with self._scene.lock:
+            if department:
+                if not self._scene.layer_stack.has_layer(layer_key):
+                    self._reject_layer_stack_change_for_flat_receivers()
+                layer, added = self._scene.layer_stack.ensure_layer(layer_key, label=department)
+                if added:
+                    self._apply_department_order_locked()
+            else:
+                layer = self._scene.edit_layer
+            self._client_layer_keys[client_id] = layer_key
+        if added:
+            self._bump_snapshot_epoch(f"create_department_layer:{department}")
+            self._broadcast_layer_stack_state()
+            LOG.info("Created shared layer for department %s", department)
+        return layer
+
+    def _flat_replay_rejection_reason_locked(self) -> str:
+        """Return why a new flat receiver cannot mirror this server."""
+        if self.department_priority:
+            return "department collaboration requires layered replay"
+        if len(self._scene.layer_stack.layer_keys) != 1:
+            return "multiple collaboration layers require layered replay"
+        layer = self._scene.layer_stack.ordered_layers[0]
+        if self._scene.stage.IsLayerMuted(layer.identifier):
+            return "muted collaboration layers require layered replay"
+        return ""
+
+    def reserve_receiver_replay_mode(self, layered: bool) -> tuple[bool, str]:
+        """Atomically admit a receiver under the current layer-stack contract."""
+        if layered:
+            return True, ""
+        with self._scene.lock:
+            reason = self._flat_replay_rejection_reason_locked()
+            if reason:
+                return False, reason
+            self._flat_receiver_count += 1
+        return True, ""
+
+    def release_receiver_replay_mode(self, layered: bool) -> None:
+        """Release a replay-mode reservation when a receiver disconnects."""
+        if layered:
+            return
+        with self._scene.lock:
+            if self._flat_receiver_count <= 0:
+                raise RuntimeError("flat receiver reservation underflow")
+            self._flat_receiver_count -= 1
+
+    def _reject_layer_stack_change_for_flat_receivers(self) -> None:
+        if self._flat_receiver_count:
+            raise ReplayModeConflictError(
+                "layer-stack changes require all connected receivers to use layered replay"
+            )
+
+    def apply_department_order(self) -> bool:
+        """Restore configured ordering after materializing layers during startup."""
+        with self._scene.lock:
+            return self._apply_department_order_locked()
+
+    def _apply_department_order_locked(self) -> bool:
+        priority_keys = []
+        for department in self.department_priority:
+            layer_key = _layer_key_for_department(department)
+            if self._scene.layer_stack.has_layer(layer_key):
+                priority_keys.append(layer_key)
+        priority_set = set(priority_keys)
+        unlisted_keys = [
+            layer_key
+            for layer_key in self._scene.layer_stack.layer_keys
+            if layer_key != _DEFAULT_LAYER_KEY and layer_key not in priority_set
+        ]
+        return self._scene.layer_stack.set_order(
+            [*priority_keys, *unlisted_keys, _DEFAULT_LAYER_KEY]
+        )
+
+    def resolve_layer_key(self, key: str) -> str | None:
+        """Resolve a client ID, department name, or layer key."""
+        with self._scene.lock:
+            client_key = self._client_layer_keys.get(key)
+            if client_key:
+                return client_key
+            department_key = _layer_key_for_department(key)
+            if self._scene.layer_stack.has_layer(department_key):
+                return department_key
+            if self._scene.layer_stack.has_layer(key):
+                return key
+            return None
+
+    def resolve_layer(self, key: str) -> Sdf.Layer | None:
+        """Resolve a layer by client ID, department name, or layer key."""
+        with self._scene.lock:
+            layer_key = self.resolve_layer_key(key)
+            return self._scene.layer_stack.layer_for(layer_key) if layer_key else None
+
+    def department_for_layer(self, layer: Sdf.Layer) -> str | None:
+        """Return department policy metadata for a managed layer."""
+        with self._scene.lock:
+            layer_key = self._scene.layer_stack.key_for_layer(layer)
+            return department_for_layer_key(layer_key) if layer_key is not None else None
+
+    def release_client_assignment(self, client_id: str, *, merge_into_root: bool) -> bool:
+        """Release a department assignment, optionally merging its opinions first.
+
+        Remove the layer only after its last client leaves. Default-layer
+        assignments are unaffected. Publish changes after releasing the scene lock.
+        """
+        with self._scene.lock:
+            layer_key = self._client_layer_keys.get(client_id)
+            if not layer_key or layer_key == _DEFAULT_LAYER_KEY:
+                return False
+            if merge_into_root:
+                self._scene.merge_layer_into_root(self._scene.layer_stack.layer_for(layer_key))
+            del self._client_layer_keys[client_id]
+            removed = layer_key not in self._client_layer_keys.values()
+            if removed:
+                self._scene.layer_stack.remove_layer(layer_key)
+        action = "merge_layer" if merge_into_root else "delete_layer"
+        self._bump_snapshot_epoch(f"{action}:{client_id}")
+        if removed:
+            self._broadcast_layer_stack_state()
+        LOG.info("Released department assignment for client %s (%s)", client_id, action)
+        return True
+
+    def set_department_priority(self, ordered_departments: list[str]) -> None:
+        """Set department priority ordering (strongest first)."""
+        ordered_departments = self._validate_priority(ordered_departments)
+        with self._scene.lock:
+            policy_changed = ordered_departments != self.department_priority
+            if policy_changed:
+                self._reject_layer_stack_change_for_flat_receivers()
+            self.department_priority = ordered_departments
+            order_changed = self._apply_department_order_locked()
+        if not order_changed and not policy_changed:
+            return
+        self._bump_snapshot_epoch("set_department_priority")
+        self._broadcast_layer_stack_state()
+
+    def get_layer_stack_info(self) -> list[dict]:
+        """Return ordered layer stack info for the dashboard.
+
+        Department policy metadata is projected over the generic stack.
+        Layers without clients or authored content stay out of the dashboard.
+        """
+        with self._scene.lock:
+            muted = set(self._scene.stage.GetMutedLayers())
+            clients_by_key: dict[str, list[str]] = {}
+            for client_id, layer_key in self._client_layer_keys.items():
+                clients_by_key.setdefault(layer_key, []).append(client_id)
+            rows = []
+            for layer_key in self._scene.layer_stack.layer_keys:
+                layer = self._scene.layer_stack.layer_for(layer_key)
+                clients = clients_by_key.get(layer_key, [])
+                if clients or not layer.empty:
+                    rows.append({
+                        "layer_key": layer_key,
+                        "label": self._scene.layer_stack.label_for(layer_key),
+                        "department": department_for_layer_key(layer_key),
+                        "clients": clients,
+                        "identifier": layer.identifier,
+                        "muted": layer.identifier in muted,
+                        "shared": layer_key == _DEFAULT_LAYER_KEY,
+                    })
+            return rows
+
+    def set_muted(self, key: str, muted: bool) -> bool:
+        """Change muting only when the admitted receivers can represent it."""
+        with self._scene.lock:
+            layer_key = self.resolve_layer_key(key)
+            if layer_key is None:
+                return False
+            layer = self._scene.layer_stack.layer_for(layer_key)
+            if self._scene.stage.IsLayerMuted(layer.identifier) == muted:
+                return True
+            self._reject_layer_stack_change_for_flat_receivers()
+            self._scene.layer_stack.set_muted(layer_key, muted)
+        action = "mute_layer" if muted else "unmute_layer"
+        self._bump_snapshot_epoch(f"{action}:{key}")
+        self._broadcast_layer_stack_state()
+        return True