refactor(public-safety): import the remaining identifier shapes from their owner

loopx/capabilities/benchmark_toolkit/integrity.py

@@ -14,6 +14,7 @@
 from urllib.parse import urlsplit
 
 from ...control_plane.content_digest import BARE_SHA256_PATTERN as _SHA256_PATTERN
+from ...public_safe_text import COMPACT_TOKEN_PATTERN as _PUBLIC_EVIDENCE_ID_PATTERN
 
 from .environment_access import (
     _heredoc_delimiters,
@@ -156,7 +157,6 @@
 # metadata, so their narrative arguments cannot themselves access benchmark data.
 # Unknown tools remain fail-closed and continue through the access-request scan.
 _NON_ACCESS_CONTROL_TOOLS = frozenset({"update_plan"})
-_PUBLIC_EVIDENCE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
 _TASK_ENVIRONMENT_NAMESPACE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{1,30}_$")
 _TASK_ENVIRONMENT_VARIABLE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{1,79}$")
 _REQUIRED_ADJUDICATION_REVIEW_SURFACES = frozenset(

loopx/control_plane/goals/deletion_service.py

@@ -7,7 +7,6 @@
 import hashlib
 import os
 from pathlib import Path
-import re
 import shutil
 from typing import Any
 import uuid
@@ -25,6 +24,7 @@
     lock_incident_path,
 )
 from ...history import load_registry
+from ...public_safe_text import OPAQUE_ID_PATTERN as _OPAQUE_ID
 from ...registry import atomic_write_json, read_json
 from ...registry_writability import probe_registry_write_path
 from ..coordination.shadow_management import shadow_maintenance_lock_target
@@ -48,7 +48,6 @@
     "loopx_goal_deletion_state_fingerprint_v1"
 )
 GOAL_DELETION_RECOVERY_SCHEMA_VERSION = "loopx_goal_deletion_recovery_v1"
-_OPAQUE_ID = re.compile(r"^[A-Za-z0-9._:-]{1,200}$")
 _MAX_RECOVERY_RECEIPT_BYTES = 64 * 1024
 
 

loopx/extensions/presentation.py

@@ -17,7 +17,10 @@
 
 from ..file_lock import exclusive_file_lock
 # Public extension presentation utility; keep path classification single-owned.
-from ..public_safe_text import find_public_safe_local_path as find_public_safe_local_path
+from ..public_safe_text import (
+    COMPACT_TOKEN_PATTERN as _ID_RE,
+    find_public_safe_local_path as find_public_safe_local_path,
+)
 from .runtime import (
     MAX_EXTENSION_RESPONSE_BYTES,
     _resolved_active_extension,
@@ -45,7 +48,6 @@
     "extension_projection_publish_receipt_v0"
 )
 
-_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
 _ANCHOR_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)*$")
 _MARKUP_RE = re.compile(r"<[^>]*>|javascript:", re.IGNORECASE)
 _LOCAL_PATH_RE = re.compile(

tests/control_plane/test_compact_identifier_shape_owner.py

@@ -7,6 +7,7 @@
 import pytest
 
 from loopx import chat_action_normalization, chat_actions, chat_action_store
+from loopx.capabilities.benchmark_toolkit import integrity
 from loopx.capabilities.decision_context import packets, profile, sources
 from loopx.capabilities.material_lifecycle import _validation
 from loopx.capabilities.reward_memory import (
@@ -19,7 +20,12 @@
     registry,
 )
 from loopx.capabilities.semantic_preference import contract
-from loopx.control_plane.goals import botmux_runtime, goal_ref_validation
+from loopx.control_plane.goals import (
+    botmux_runtime,
+    deletion_service,
+    goal_ref_validation,
+)
+from loopx.extensions import presentation
 from loopx.public_safe_text import (
     COMPACT_TOKEN_PATTERN,
     OPAQUE_ID_PATTERN as PUBLIC_OPAQUE_ID,
@@ -34,11 +40,11 @@
 SURFACE_SHAPE = "^[a-z][a-z0-9_-]*(?:\\.[a-z][a-z0-9_-]*)+$"
 OPAQUE_SHAPE = "^[A-Za-z0-9._:-]{1,200}$"
 
-# A file that still compiles this body itself, with the number of constructions
-# allowed there. The goal-deletion service is being restructured in #5854, so this
-# slice declares that site instead of racing it; converting it deletes the entry,
-# and leaving an entry whose site is already converted fails too.
-OPEN_OPAQUE_SITES = {"loopx/control_plane/goals/deletion_service.py": 1}
+# Sites still allowed to state each body locally, per file and count. Both are
+# empty: every converted module imports the owner instead. A new restatement, or
+# an entry whose site has already been converted, fails the census.
+OPEN_OPAQUE_SITES: dict[str, int] = {}
+OPEN_COMPACT_SITES: dict[str, int] = {}
 
 
 SHAPE_CONSUMERS = {
@@ -53,10 +59,12 @@
         "loopx/capabilities/semantic_preference/contract.py",
     ),
     COMPACT_SHAPE: (
+        "loopx/capabilities/benchmark_toolkit/integrity.py",
         "loopx/capabilities/decision_context/packets.py",
         "loopx/capabilities/decision_context/profile.py",
         "loopx/capabilities/decision_context/sources.py",
         "loopx/capabilities/material_lifecycle/_validation.py",
+        "loopx/extensions/presentation.py",
     ),
     SURFACE_SHAPE: (
         "loopx/capabilities/reward_memory/application.py",
@@ -107,11 +115,24 @@ def test_the_opaque_id_body_is_compiled_only_by_the_owner_and_declared_sites() -
     assert found == {OWNER_MODULE: 1, **OPEN_OPAQUE_SITES}
 
 
+def test_the_compact_token_body_is_compiled_only_by_the_owner_and_declared_sites() -> None:
+    found: dict[str, int] = {}
+    for path in sorted(REPOSITORY_ROOT.glob("loopx/**/*.py")):
+        relative = path.relative_to(REPOSITORY_ROOT).as_posix()
+        count = path.read_text(encoding="utf-8").count(COMPACT_SHAPE)
+        if count:
+            found[relative] = count
+    assert found == {OWNER_MODULE: 1, **OPEN_COMPACT_SITES}
+
+
 def test_each_contract_holds_the_owners_object_and_reports_its_own_field() -> None:
     assert chat_actions._OPAQUE_ID is PUBLIC_OPAQUE_ID
     assert chat_action_store._OPAQUE_ID is PUBLIC_OPAQUE_ID
+    assert deletion_service._OPAQUE_ID is PUBLIC_OPAQUE_ID
     assert goal_ref_validation.GOAL_ID is PUBLIC_OPAQUE_ID
     assert botmux_runtime._SAFE_TOKEN is PUBLIC_OPAQUE_ID
+    assert presentation._ID_RE is COMPACT_TOKEN_PATTERN
+    assert integrity._PUBLIC_EVIDENCE_ID_PATTERN is COMPACT_TOKEN_PATTERN
     for rejected in ("x" * 201, "has space"):
         with pytest.raises(ValueError, match="must be a compact opaque id"):
             chat_action_store._opaque_id(rejected, field="action_id")