fix(scopes): validate peer names on create, close namespace squatting and upsert race

Addresses three review findings against 48047a6a.

1. `PeerSpec` let API callers create invalid and reserved-prefix peers.
   Widening `get_or_create_peers` to accept a pattern-free schema fixed the
   lookup 500s but also removed validation from the *insert* path, and
   request-controlled names reach it via message authors, session peer maps, and
   the chat observer path — none of which carry a charset pattern of their own.
   Confirmed: `POST /sessions/{id}/messages` with `peer_id: "scope.x"` returned
   201 and minted an unflagged squatter, after which `POST /scopes {id: x}` was
   permanently 409-blocked — namespace denial of service by any caller able to
   post a message. `peer_id: "not a valid name!@#"` was likewise created.

   Fixed by validating only names about to be INSERTed
   (`_validate_new_peer_names`), so already-existing names — legacy dotted
   names, scope peers — still resolve without a spurious 422. That keeps the
   Dreamer fix intact, since it reads through `get_peer`.

2. Existing reserved-prefix squatters could not be updated. The name-based guard
   on `PUT /peers/{peer_id}` refused every `scope.` name, contradicting the
   invariant that an unflagged squatter stays a normal peer. Now flag-based, so
   behavior is three-way: a real scope is refused, an existing unflagged peer
   updates, and a missing reserved-prefix name is refused by (1) rather than
   minted.

3. Scope checks raced with get-or-create and the membership upsert. The
   route-level guards run before peers are resolved, so a scope created
   concurrently in that window would be attached by the generic path with a
   default `SessionPeerConfig()`, clobbering its observer membership config.
   Adds `_reject_resolved_scope_peers`, which runs on the resolved rows in the
   same transaction as the upsert — no window, no extra query. The early checks
   stay for better error messages.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/crud/peer.py

@@ -1,5 +1,6 @@
 """CRUD helpers for peer records and peer-scoped session queries."""
 
+import re
 from collections.abc import Iterable
 from logging import getLogger
 from typing import Any, Literal
@@ -20,6 +21,7 @@ from src.exceptions import (
     ValidationException,
 )
 from src.models import Peer
+from src.schemas.api import RESOURCE_NAME_PATTERN
 from src.utils import scopes as scopes_util
 from src.utils.filter import apply_filter
 from src.utils.types import GetOrCreateResult
@@ -42,6 +44,27 @@ def peer_cache_key(workspace_name: str, peer_name: str) -> str:
     )
 
 
+def _validate_new_peer_names(names: list[str]) -> None:
+    """Validate peer names that are about to be created.
+
+    Mirrors ``PeerCreate``'s contract for peers arriving through crud rather than
+    the peers route. The reserved prefix is reported separately because it is
+    also outside ``RESOURCE_NAME_PATTERN``, so the charset check would otherwise
+    mask the real problem.
+
+    Raises:
+        ValidationException: On a reserved-prefix or non-conforming name.
+    """
+    scopes_util.validate_no_scope_peer_names(
+        names, action="Use the scopes routes to create scopes."
+    )
+    offenders = sorted({n for n in names if not re.fullmatch(RESOURCE_NAME_PATTERN, n)})
+    if offenders:
+        raise ValidationException(
+            f"Peer name(s) {offenders} must match pattern {RESOURCE_NAME_PATTERN}"
+        )
+
+
 def scope_peer_clause() -> ColumnElement[bool]:
     """SQL form of ``is_scope_peer()``: reserved name prefix AND the internal kind flag.
 
@@ -166,6 +189,17 @@ async def get_or_create_peers(
     existing_names = {p.name for p in existing_peers}
     peers_to_create = [p for p in peers if p.name not in existing_names]
 
+    # Names are validated on the *create* path only. `PeerSpec` deliberately
+    # carries no charset pattern so already-existing names (legacy dotted names,
+    # scope peers) can be looked up without a spurious 422 — but a name we are
+    # about to INSERT is a new peer, and new peers must obey the public contract.
+    # Without this, request-controlled names reach here unvalidated via message
+    # authors, session peer maps, and the chat observer path, letting a caller
+    # mint `scope.x` squatters (permanently 409-blocking that scope) or peers
+    # that violate RESOURCE_NAME_PATTERN outright.
+    if peers_to_create:
+        _validate_new_peer_names([p.name for p in peers_to_create])
+
     # Create new peers
     new_peers = [
         models.Peer(

src/crud/session.py

@@ -27,9 +27,10 @@ from src.exceptions import (
     ConflictException,
     ObserverException,
     ResourceNotFoundException,
+    ValidationException,
 )
 from src.utils.filter import apply_filter
-from src.utils.scopes import scope_peer_name
+from src.utils.scopes import is_scope_peer, scope_peer_name
 from src.utils.types import GetOrCreateResult
 from src.vector_store import get_external_vector_store
 
@@ -101,6 +102,30 @@ async def _fetch_session(
     }
 
 
+def _reject_resolved_scope_peers(peers: list[models.Peer]) -> None:
+    """Reject scope peers among rows already resolved for a membership upsert.
+
+    The route-level guards check names *before* peers are resolved, which leaves a
+    check-then-upsert window: if a scope is created concurrently between that
+    check and the upsert below, the generic path would attach the now-flagged
+    scope peer with a default ``SessionPeerConfig()``, clobbering its
+    ``observe_others=True/observe_me=False`` membership config. This runs on the
+    resolved rows inside the same transaction as the upsert, so there is no
+    window and no extra query.
+
+    Raises:
+        ValidationException: If any resolved peer is a scope.
+    """
+    offenders = sorted(
+        p.name for p in peers if is_scope_peer(p.name, p.internal_metadata)
+    )
+    if offenders:
+        raise ValidationException(
+            f"Peer name(s) {offenders} are scopes."
+            + " Scope membership is managed via the scopes routes."
+        )
+
+
 def count_observers_in_config(
     peer_configs: dict[str, schemas.SessionPeerConfig],
 ) -> int:
@@ -267,6 +292,7 @@ async def get_or_create_session(
                 schemas.PeerSpec(name=peer_name) for peer_name in session.peer_names
             ],
         )
+        _reject_resolved_scope_peers(peers_result.resource)
         await _get_or_add_peers_to_session(
             db,
             workspace_name=workspace_name,
@@ -992,6 +1018,7 @@ async def set_peers_for_session(
         workspace_name=workspace_name,
         peers=[schemas.PeerSpec(name=peer_name) for peer_name in peer_names],
     )
+    _reject_resolved_scope_peers(peers_result.resource)
 
     # Add new peers to session
     peers = await _get_or_add_peers_to_session(

src/routers/peers.py

@@ -138,17 +138,15 @@ async def update_peer(
 ):
     """Update a Peer's metadata and/or configuration.
 
-    Reserved-namespace names are refused outright. Name-based rather than
-    flag-based on purpose: every scope peer is named by ``scope_peer_name``, so
-    the prefix covers real scopes *and* stops ``crud.update_peer``'s get-or-create
-    from minting a new unflagged peer inside the reserved namespace — which a
-    flag-based check would wave through. It also needs no DB round-trip. The
-    trade-off is that a legacy peer occupying the namespace can't be updated via
-    this route; it couldn't be before either, and `configuration` is replaced
-    wholesale here, so the generic route must not touch the facade's namespace.
+    Three-way on the reserved namespace: a real scope is refused here (this route
+    replaces `configuration` wholesale, so it must never touch a facade-managed
+    peer); an existing *unflagged* peer that merely occupies the namespace is a
+    normal peer and updates fine; and a reserved-prefix name that does not exist
+    is refused by ``get_or_create_peers``' create-path validation rather than
+    being minted.
     """
-    validate_no_scope_peer_names(
-        [peer_id], action="Use the scopes routes to manage scopes."
+    await crud.reject_scope_peers(
+        db, workspace_id, [peer_id], action="Use the scopes routes to manage scopes."
     )
     updated_peer = await crud.update_peer(
         db, workspace_name=workspace_id, peer_name=peer_id, peer=peer

tests/routes/test_scopes.py

@@ -15,9 +15,10 @@ from nanoid import generate as generate_nanoid
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src import crud, models
+from src import crud, models, schemas
 from src.config import settings
 from src.deriver.enqueue import enqueue
+from src.exceptions import ValidationException
 from src.models import Peer, QueueItem, Workspace
 from src.schemas.api import RESOURCE_NAME_PATTERN
 from src.security import JWTParams, create_jwt
@@ -814,3 +815,153 @@ async def test_crud_get_peer_resolves_scope_and_dotted_names(
     assert resolved.name == dotted
     # Has neither half of the invariant
     assert not is_scope_peer(resolved.name, resolved.internal_metadata)
+
+
+# ---------------------------------------------------------------------------
+# Name validation on the create path. PeerSpec carries no charset pattern so
+# existing names can be *looked up*, but anything crud is about to INSERT is a
+# new peer and must obey the public contract — otherwise request-controlled
+# names (message authors, session peer maps) mint squatters.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "bad_name",
+    ["not a valid name!@#", "has spaces", "emoji-\U0001f600"],
+)
+def test_message_author_cannot_create_invalid_peer_name(
+    client: TestClient, sample_data: tuple[Workspace, Peer], bad_name: str
+):
+    """A message author must not be able to create a non-conforming peer."""
+    test_workspace, _ = sample_data
+    session_name = _create_session(client, test_workspace.name)
+
+    response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/sessions/{session_name}/messages",
+        json={"messages": [{"peer_id": bad_name, "content": "hello"}]},
+    )
+    assert response.status_code == 422, response.text
+
+    response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/peers/list", json={"kind": "all"}
+    )
+    assert bad_name not in [p["id"] for p in response.json()["items"]]
+
+
+def test_message_author_cannot_squat_scope_namespace(
+    client: TestClient, sample_data: tuple[Workspace, Peer]
+):
+    """The reserved namespace must not be reachable via the message author path.
+
+    Without create-path validation this minted an unflagged `scope.<name>` peer,
+    which then permanently 409-blocked creating that scope — a denial of service
+    on the namespace by any caller who can post a message.
+    """
+    test_workspace, _ = sample_data
+    scope_name = str(generate_nanoid())
+    session_name = _create_session(client, test_workspace.name)
+
+    response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/sessions/{session_name}/messages",
+        json={
+            "messages": [{"peer_id": scope_peer_name(scope_name), "content": "hello"}]
+        },
+    )
+    assert response.status_code == 422, response.text
+    assert SCOPE_PEER_PREFIX in response.json()["detail"]
+
+    # The scope name is still available
+    assert _create_scope(client, test_workspace.name, scope_name).status_code == 201
+
+
+def test_session_peer_map_cannot_squat_scope_namespace(
+    client: TestClient, sample_data: tuple[Workspace, Peer]
+):
+    """Same hole, via the session peer mapping rather than a message author."""
+    test_workspace, _ = sample_data
+    scope_name = str(generate_nanoid())
+
+    response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/sessions",
+        json={
+            "id": str(generate_nanoid()),
+            "peers": {scope_peer_name(scope_name): {}},
+        },
+    )
+    assert response.status_code == 422, response.text
+    assert _create_scope(client, test_workspace.name, scope_name).status_code == 201
+
+
+async def test_unflagged_squatter_can_still_be_updated(
+    client: TestClient,
+    db_session: AsyncSession,
+    sample_data: tuple[Workspace, Peer],
+):
+    """An existing unflagged peer in the namespace is a normal peer.
+
+    Three-way behavior on PUT /peers/{id}: a real scope is refused, an existing
+    unflagged squatter updates fine, and a missing reserved-prefix name is
+    refused by create-path validation rather than being minted.
+    """
+    test_workspace, _ = sample_data
+    squatter = scope_peer_name(str(generate_nanoid()))
+    db_session.add(models.Peer(workspace_name=test_workspace.name, name=squatter))
+    await db_session.commit()
+
+    # existing unflagged peer -> allowed
+    response = client.put(
+        f"/v3/workspaces/{test_workspace.name}/peers/{squatter}",
+        json={"metadata": {"k": "v"}},
+    )
+    assert response.status_code == 200, response.text
+    assert response.json()["metadata"] == {"k": "v"}
+
+    # real scope -> refused
+    scope_name = str(generate_nanoid())
+    assert _create_scope(client, test_workspace.name, scope_name).status_code == 201
+    response = client.put(
+        f"/v3/workspaces/{test_workspace.name}/peers/{scope_peer_name(scope_name)}",
+        json={"metadata": {"k": "v"}},
+    )
+    assert response.status_code == 422, response.text
+
+    # missing reserved-prefix name -> refused, not created
+    missing = scope_peer_name(str(generate_nanoid()))
+    response = client.put(
+        f"/v3/workspaces/{test_workspace.name}/peers/{missing}",
+        json={"metadata": {"k": "v"}},
+    )
+    assert response.status_code == 422, response.text
+    response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/peers/list", json={"kind": "all"}
+    )
+    assert missing not in [p["id"] for p in response.json()["items"]]
+
+
+async def test_resolved_scope_peer_rejected_at_membership_upsert(
+    client: TestClient,
+    db_session: AsyncSession,
+    sample_data: tuple[Workspace, Peer],
+):
+    """The last-line guard runs on resolved rows, closing the check-then-upsert race.
+
+    Simulates the race by flagging the peer *after* the route-level name check
+    would have passed: the peer exists and is unflagged when named, and is a real
+    scope by the time membership is upserted.
+    """
+    test_workspace, _ = sample_data
+    scope_name = str(generate_nanoid())
+    assert _create_scope(client, test_workspace.name, scope_name).status_code == 201
+    backing = scope_peer_name(scope_name)
+    session_name = _create_session(client, test_workspace.name)
+
+    # crud-level: the resolved row is a scope, so membership must be refused even
+    # though the peer already exists (no create-path validation fires).
+    with pytest.raises(ValidationException):
+        await crud.get_or_create_session(
+            db_session,
+            session=schemas.SessionCreate(
+                name=session_name, peers={backing: schemas.SessionPeerConfig()}
+            ),
+            workspace_name=test_workspace.name,
+        )