refactor(periodic-report): import the public-safe slug shape from its owner

loopx/capabilities/periodic_report/adapters.py

@@ -17,6 +17,7 @@
 )
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
 from ...control_plane.digest_envelope import sha256_envelope
+from ...public_safe_text import PUBLIC_SAFE_SLUG_PATTERN
 
 
 SOURCE_RESULT_SCHEMA = "periodic_report_source_result_v0"
@@ -26,7 +27,6 @@
 ARTIFACT_SCHEMA = "periodic_report_artifact_v0"
 SINK_RESULT_SCHEMA = "periodic_report_sink_result_v0"
 
-_TOKEN_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
 _ITEM_CONTENT_KINDS = {
     "capability_change",
     "coverage",
@@ -81,7 +81,7 @@ def _optional_text(
 
 def _token(value: object, label: str) -> str:
     token = _text(value, label, maximum=128).lower()
-    if not _TOKEN_RE.fullmatch(token):
+    if not PUBLIC_SAFE_SLUG_PATTERN.fullmatch(token):
         raise ValueError(f"{label} must be a lower-snake-like public token")
     return token
 

loopx/capabilities/periodic_report/archive.py

@@ -9,13 +9,13 @@
 
 import hashlib
 import json
-import re
 from collections.abc import Callable, Mapping, Sequence
 from typing import Any
 from urllib.parse import unquote, urlsplit
 
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
 from ...control_plane.digest_envelope import sha256_envelope
+from ...public_safe_text import PUBLIC_SAFE_SLUG_PATTERN
 from .adapters import ARTIFACT_SCHEMA, DOCUMENT_SCHEMA
 from .core import _normalize_trigger_receipt, _reject_raw_keys
 
@@ -25,8 +25,6 @@
 ARCHIVE_RECEIPT_SCHEMA = "periodic_report_archive_receipt_v0"
 MEMORY_REFERENCE_SCHEMA = "periodic_report_memory_reference_v0"
 
-_TOKEN_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
-
 ArchiveReadback = Callable[[str], Mapping[str, Any]]
 
 
@@ -62,7 +60,7 @@ def _content(value: object, label: str, *, maximum: int) -> str:
 
 def _token(value: object, label: str) -> str:
     token = _text(value, label, maximum=128).lower()
-    if not _TOKEN_RE.fullmatch(token):
+    if not PUBLIC_SAFE_SLUG_PATTERN.fullmatch(token):
         raise ValueError(f"{label} must be a lower-snake-like public token")
     return token
 

loopx/capabilities/periodic_report/audience.py

@@ -1,20 +1,19 @@
 from __future__ import annotations
 
 import json
-import re
 from collections.abc import Mapping, Sequence
 from typing import Any
 
 from .core import _reject_raw_keys
 from ...control_plane.digest_envelope import sha256_envelope
+from ...public_safe_text import PUBLIC_SAFE_SLUG_PATTERN
 
 AUDIENCE_POLICY_SCHEMA = "periodic_report_audience_policy_v0"
 AUDIENCE_RECIPIENT_SCHEMA = "periodic_report_audience_recipient_v0"
 AUDIENCE_ROUTING_RULE_SCHEMA = "periodic_report_audience_routing_rule_v0"
 ANNOUNCEMENT_PLAN_SCHEMA = "periodic_report_announcement_plan_v0"
 
 _DOCUMENT_SCHEMA = "periodic_report_document_v0"
-_TOKEN_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
 _VISIBILITIES = {"primary", "supporting"}
 _RULE_SELECTORS = {
     "source_ids",
@@ -48,7 +47,7 @@ def _text(value: object, label: str, *, maximum: int = 128) -> str:
 
 def _token(value: object, label: str) -> str:
     token = _text(value, label).lower()
-    if not _TOKEN_RE.fullmatch(token):
+    if not PUBLIC_SAFE_SLUG_PATTERN.fullmatch(token):
         raise ValueError(f"{label} must be a lower-snake-like public token")
     return token
 

loopx/capabilities/periodic_report/bindings.py

@@ -17,6 +17,7 @@
 )
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
 from ...control_plane.digest_envelope import sha256_envelope
+from ...public_safe_text import PUBLIC_SAFE_SLUG_PATTERN
 
 GENERATION_BUNDLE_SCHEMA = "periodic_report_generation_bundle_v0"
 GENERATION_RECEIPT_SCHEMA = "periodic_report_generation_receipt_v0"
@@ -25,7 +26,6 @@
 DELIVERY_INTENT_SCHEMA = "periodic_report_delivery_intent_v0"
 DELIVERY_RECEIPT_SCHEMA = "periodic_report_delivery_receipt_v0"
 
-_TOKEN_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
 _VERSION_RE = re.compile(r"^[0-9a-z][a-z0-9_.+-]{0,127}$")
 _DEPENDENCY_POLICIES = {"required", "optional", "disabled"}
 _PROVIDER_STATUSES = {"ready", "unavailable", "unknown"}
@@ -69,7 +69,7 @@ def _text(value: object, label: str, *, maximum: int = 500) -> str:
 
 def _token(value: object, label: str) -> str:
     token = _text(value, label, maximum=128).lower()
-    if not _TOKEN_RE.fullmatch(token):
+    if not PUBLIC_SAFE_SLUG_PATTERN.fullmatch(token):
         raise ValueError(f"{label} must be a lower-snake-like public token")
     return token
 

loopx/public_safe_text.py

@@ -356,11 +356,11 @@
 )
 
 
-# One definition for "this identifier is a lowercase public-safe slug". Four
-# periodic-report surfaces and one hand-off validator each compiled the same
-# shape locally; the four report files are being edited by another branch of
-# mine right now, so this slice converts the three free ones and the guard
-# declares the other four by file and count.
+# One definition for "this identifier is a lowercase public-safe slug". Seven
+# modules - four periodic-report surfaces, the issue-fix report and a hand-off
+# validator - compiled the same shape locally; every one of them now imports
+# this object, and tests/architecture/test_public_safe_slug_owner.py rejects a
+# new local copy.
 PUBLIC_SAFE_SLUG_PATTERN = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
 
 

tests/architecture/test_public_safe_slug_owner.py

@@ -4,19 +4,18 @@
 identifier a lowercase public-safe slug -- and on the base revision seven modules
 compiled it themselves: ``periodic_report/core.py``, ``issue_fix/periodic_report.py``
 and ``control_plane/handoff/review_batch.py``, plus four files inside
-``capabilities/periodic_report`` that this slice declares instead of converting.
+``capabilities/periodic_report``.
+
+Those four files were declared by file and count instead of being converted
+because another branch was rewriting their import blocks at the time. That
+branch is no longer in flight, so they are converted here and the declaration
+budget is empty: every module that states the shape must now be the owner or a
+module that imports the owner's object.
 
 Decisions are made on the folded value, not the spelling: a regex built at module
 level, inside a function, or by concatenating same-file string constants is
 reported, and a construction whose value cannot be folded has to be declared with
 its file and count.
-
-The four declared sites are ``adapters.py``, ``archive.py``, ``audience.py`` and
-``bindings.py``.  They are not converted here because their import blocks are
-being rewritten by an open branch of the same author, so a seventh edit would
-only create a conflict between two pending pull requests.  Converting one means
-deleting its entry, and a declared file whose count moves fails the guard either
-way.
 """
 
 from __future__ import annotations
@@ -29,24 +28,43 @@
 
 from loopx import public_safe_text
 from loopx.capabilities.issue_fix import periodic_report as issue_fix_report
-from loopx.capabilities.periodic_report import core
+from loopx.capabilities.periodic_report import (
+    adapters,
+    archive,
+    audience,
+    bindings,
+    core,
+)
 from loopx.control_plane.handoff import review_batch
 
 REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
 OWNER_PATH = "loopx/public_safe_text.py"
 SLUG_BODY = r"^[a-z][a-z0-9_.-]{0,127}$"
 SLUG_INNER = SLUG_BODY[1:-1]
 
-DECLARED_INDIVIDUAL_SITES: dict[str, int] = {
-    "loopx/capabilities/periodic_report/adapters.py": 1,
-    "loopx/capabilities/periodic_report/archive.py": 1,
-    "loopx/capabilities/periodic_report/audience.py": 1,
-    "loopx/capabilities/periodic_report/bindings.py": 1,
-}
+# Sites still allowed to state the shape locally, per file and count.  Empty:
+# every converted module imports the owner instead.
+DECLARED_INDIVIDUAL_SITES: dict[str, int] = {}
 # A construction whose value this scan cannot fold, per file.  None today.
 DECLARED_UNFOLDABLE_SITES: dict[str, int] = {}
 
 CONSUMERS: dict[str, tuple[ModuleType, str]] = {
+    "loopx/capabilities/periodic_report/adapters.py": (
+        adapters,
+        "PUBLIC_SAFE_SLUG_PATTERN",
+    ),
+    "loopx/capabilities/periodic_report/archive.py": (
+        archive,
+        "PUBLIC_SAFE_SLUG_PATTERN",
+    ),
+    "loopx/capabilities/periodic_report/audience.py": (
+        audience,
+        "PUBLIC_SAFE_SLUG_PATTERN",
+    ),
+    "loopx/capabilities/periodic_report/bindings.py": (
+        bindings,
+        "PUBLIC_SAFE_SLUG_PATTERN",
+    ),
     "loopx/capabilities/periodic_report/core.py": (core, "PUBLIC_SAFE_SLUG_PATTERN"),
     "loopx/capabilities/issue_fix/periodic_report.py": (
         issue_fix_report,
@@ -284,12 +302,32 @@ def test_no_other_module_compiles_the_slug_shape() -> None:
     assert offender_rows(collect(REPO_ROOT)) == []
 
 
-def test_declared_sites_are_exactly_the_four_in_flight_files() -> None:
+def modules_importing_the_owner() -> set[str]:
+    importing: set[str] = set()
+    for path in sorted((REPO_ROOT / "loopx").rglob("*.py")):
+        try:
+            tree = ast.parse(path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeDecodeError, SyntaxError):
+            continue
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and any(
+                alias.name == "PUBLIC_SAFE_SLUG_PATTERN" for alias in node.names
+            ):
+                importing.add(path.relative_to(REPO_ROOT).as_posix())
+    return importing
+
+
+def test_every_slug_site_is_the_owner_or_a_converted_consumer() -> None:
     rows = collect(REPO_ROOT)
-    declared_found = sorted(
-        str(row["file"]) for row in rows if str(row["file"]) in DECLARED_INDIVIDUAL_SITES
-    )
-    assert declared_found == sorted(DECLARED_INDIVIDUAL_SITES)
+    owner_or_consumer = {OWNER_PATH, *CONSUMERS}
+    stray = sorted({str(row["file"]) for row in rows} - owner_or_consumer)
+    assert DECLARED_INDIVIDUAL_SITES == {}
+    assert stray == []
+
+
+def test_every_module_that_imports_the_owner_is_a_declared_consumer() -> None:
+    importing = modules_importing_the_owner() - {OWNER_PATH}
+    assert sorted(importing) == sorted(CONSUMERS)
 
 
 def test_owner_states_the_shape_as_the_recorded_literal() -> None:
@@ -451,20 +489,18 @@ def test_unfoldable_probe_is_forced_into_the_open() -> None:
 
 
 def test_declared_counts_are_enforced_in_both_directions() -> None:
-    rows = collect(REPO_ROOT)
-    assert offender_rows(rows) == []
-    retired = [row for row in rows if str(row["file"]) !=
-               "loopx/capabilities/periodic_report/audience.py"]
-    assert offender_rows(retired) == [
+    declared = {"loopx/capabilities/periodic_report/audience.py": 1}
+    assert offender_rows([], declared=declared) == [
         "loopx/capabilities/periodic_report/audience.py declares 1 site(s), found 0"
     ]
-    doubled = rows + [
+    doubled = [
         {
             "file": "loopx/capabilities/periodic_report/audience.py",
-            "line": 99,
+            "line": line,
             "kind": "slug_shape",
         }
+        for line in (99, 100)
     ]
-    assert offender_rows(doubled) == [
+    assert offender_rows(doubled, declared=declared) == [
         "loopx/capabilities/periodic_report/audience.py declares 1 site(s), found 2"
     ]