diff --git a/loopx/capabilities/periodic_report/adapters.py b/loopx/capabilities/periodic_report/adapters.py index afaba04f43..5477b07360 100644 --- a/loopx/capabilities/periodic_report/adapters.py +++ b/loopx/capabilities/periodic_report/adapters.py @@ -2,7 +2,6 @@ import hashlib import json -import re from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone @@ -18,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 +from .field_shapes import REPORT_LANGUAGE_PATTERN, SUPPORTING_CONTENT_KINDS SOURCE_RESULT_SCHEMA = "periodic_report_source_result_v0" @@ -39,9 +39,7 @@ "runtime", } _ITEM_VISIBILITIES = {"primary", "supporting"} -_SUPPORTING_CONTENT_KINDS = {"coverage", "delivery_receipt", "runtime"} _HIGHLIGHT_TONES = {"attention", "neutral", "positive"} -_LANGUAGE_RE = re.compile(r"^[A-Za-z]{2,8}(?:-[A-Za-z0-9]{1,8})*$") SourceCollector = Callable[[Mapping[str, Any]], Mapping[str, Any]] ArtifactRenderer = Callable[[Mapping[str, Any]], Mapping[str, Any]] ArtifactSink = Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] @@ -226,7 +224,7 @@ def _normalize_item(raw: object, *, label: str) -> dict[str, Any]: if visibility not in _ITEM_VISIBILITIES: raise ValueError(f"{label}.visibility is invalid") normalized["visibility"] = visibility - if content_kind in _SUPPORTING_CONTENT_KINDS and visibility != "supporting": + if content_kind in SUPPORTING_CONTENT_KINDS and visibility != "supporting": raise ValueError( f"{label}.visibility must be supporting for {content_kind} content" ) @@ -268,7 +266,7 @@ def _normalize_editorial(raw: object) -> dict[str, Any] | None: editorial.get("language"), "editorial.language", maximum=64 ) if language: - if not _LANGUAGE_RE.fullmatch(language): + if not REPORT_LANGUAGE_PATTERN.fullmatch(language): raise ValueError("editorial.language must be a BCP-47-like language tag") normalized["language"] = language diff --git a/loopx/capabilities/periodic_report/cadence_journal.py b/loopx/capabilities/periodic_report/cadence_journal.py index a6c7503581..6149915149 100644 --- a/loopx/capabilities/periodic_report/cadence_journal.py +++ b/loopx/capabilities/periodic_report/cadence_journal.py @@ -12,6 +12,7 @@ from pathlib import Path from typing import Any +from .field_shapes import REPORT_IDENTITY_PATTERN from ...file_lock import LockAcquisitionPolicy, exclusive_file_lock from ...registry import atomic_write_json from .cadence import report_cadence_next_due_at, report_cadence_window @@ -20,7 +21,6 @@ CADENCE_WINDOW_SCHEMA = "periodic_report_cadence_window_v0" JOURNAL_SCHEMA = "periodic_report_cadence_journal_v0" -_ID = re.compile(r"^[a-z][a-z0-9_.:-]{2,127}$") def _digest(value: object) -> str: @@ -30,7 +30,7 @@ def _digest(value: object) -> str: def cadence_journal_path(root: Path, goal_id: str) -> Path: - if not _ID.fullmatch(goal_id): + if not REPORT_IDENTITY_PATTERN.fullmatch(goal_id): raise ValueError("cadence Goal identity is invalid") return root / "goals" / goal_id / "periodic_reports" / "cadence.json" @@ -44,7 +44,7 @@ def validate_cadence_window(raw: object) -> dict[str, Any]: raise ValueError("cadence window contract is invalid") value = dict(raw) if value["schema_version"] != CADENCE_WINDOW_SCHEMA or any( - not isinstance(value[key], str) or not _ID.fullmatch(value[key]) + not isinstance(value[key], str) or not REPORT_IDENTITY_PATTERN.fullmatch(value[key]) for key in ("goal_id", "agent_id") ): raise ValueError("cadence window identity is invalid") @@ -150,7 +150,7 @@ def admit_cadence_window( A configuration change may supersede an unprepared interval under the same lock used by consumption, preserving its predecessor as immutable history. """ - if not _ID.fullmatch(agent_id): + if not REPORT_IDENTITY_PATTERN.fullmatch(agent_id): raise ValueError("cadence Agent identity is invalid") path = cadence_journal_path(runtime_root, goal_id) if subscription_resolver is None and ( diff --git a/loopx/capabilities/periodic_report/field_shapes.py b/loopx/capabilities/periodic_report/field_shapes.py new file mode 100644 index 0000000000..4685b723e2 --- /dev/null +++ b/loopx/capabilities/periodic_report/field_shapes.py @@ -0,0 +1,28 @@ +"""One owner for the field rules a periodic report is validated against. Refs #4447 Track A. + +Three decisions were each stated twice on the base revision. ``adapters.py`` and +``project_progress.py`` both compiled the language-tag rule, and both listed the +content kinds that are supporting by nature -- the two set literals even spelled +their members in a different order, which a value-keyed duplicate scan reads as two +different values. The identity rule that gates ``goal_id``, ``agent_id`` and ``failure_code`` was +compiled three times under two different binding names -- ``pending_intent.py`` and +``request_action.py`` called it ``_IDENTITY_RE``, ``cadence_journal.py`` called it +``_ID`` -- which is exactly what a name-keyed duplicate scan cannot see. + +Every one of those sites only asks membership or ``fullmatch``, so importing the +owner here is behaviour-preserving, while a widening in one copy no longer leaves +the other behind silently. + +``incremental.py`` keeps its own ``_IDENTITY_RE``. It is a *different* rule -- +case-insensitive and up to 160 characters versus lowercase-led 3..128 -- so it is +bounded-context reuse of one name, not a copy, and it is pinned as distinct here +rather than merged away by renaming. +""" + +from __future__ import annotations + +import re + +SUPPORTING_CONTENT_KINDS = frozenset({"coverage", "delivery_receipt", "runtime"}) +REPORT_IDENTITY_PATTERN = re.compile(r"^[a-z][a-z0-9_.:-]{2,127}$") +REPORT_LANGUAGE_PATTERN = re.compile(r"^[A-Za-z]{2,8}(?:-[A-Za-z0-9]{1,8})*$") diff --git a/loopx/capabilities/periodic_report/pending_intent.py b/loopx/capabilities/periodic_report/pending_intent.py index cc1ee93b00..f354d2c03c 100644 --- a/loopx/capabilities/periodic_report/pending_intent.py +++ b/loopx/capabilities/periodic_report/pending_intent.py @@ -18,6 +18,7 @@ from ...control_plane.effect_runtime import effect_runtime_result from ...control_plane.digest_envelope import sha256_envelope from ...history import load_registry +from .field_shapes import REPORT_IDENTITY_PATTERN from .todo_source import read_report_todo_source from ...registry import ( atomic_write_json, @@ -56,7 +57,6 @@ HOOK_ID = "periodic_report.pending_intent" CAPABILITY_ID = "periodic-report" _DISPATCH_RE = re.compile(r"^pwh_[0-9a-f]{64}\.json$") -_IDENTITY_RE = re.compile(r"^[a-z][a-z0-9_.:-]{2,127}$") _CHINESE_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]") _ANALYSIS_SECTION_CONTRACT = ( ("overview", "全景判断"), @@ -333,7 +333,7 @@ def pending_periodic_report_intents( ) -> list[dict[str, Any]]: """Read only exact, eligible, unconsumed intents for one Goal/Agent.""" - if not _IDENTITY_RE.fullmatch(goal_id) or not _IDENTITY_RE.fullmatch(agent_id): + if not REPORT_IDENTITY_PATTERN.fullmatch(goal_id) or not REPORT_IDENTITY_PATTERN.fullmatch(agent_id): return [] pending: list[dict[str, Any]] = [] for intent in periodic_report_request_intents( diff --git a/loopx/capabilities/periodic_report/project_progress.py b/loopx/capabilities/periodic_report/project_progress.py index 59568f5ca8..dd02c3a400 100644 --- a/loopx/capabilities/periodic_report/project_progress.py +++ b/loopx/capabilities/periodic_report/project_progress.py @@ -2,10 +2,10 @@ import hashlib import json -import re from collections.abc import Mapping, Sequence from typing import Any +from .field_shapes import REPORT_LANGUAGE_PATTERN, SUPPORTING_CONTENT_KINDS from .adapters import ( PeriodicReportSourceAdapter, build_periodic_report_source_result, @@ -44,8 +44,6 @@ "supporting_evidence": "支撑证据", }, } -_LANGUAGE_RE = re.compile(r"^[A-Za-z]{2,8}(?:-[A-Za-z0-9]{1,8})*$") -_SUPPORTING_CONTENT_KINDS = {"coverage", "runtime", "delivery_receipt"} _MAX_PRIMARY_ITEMS = 8 _MAX_SUPPORTING_ITEMS = 16 @@ -80,7 +78,7 @@ def _snapshot_ref(projection: Mapping[str, Any], *, language: str) -> str: def _language(value: object) -> str: language = str(value or "en").strip() - if not _LANGUAGE_RE.fullmatch(language): + if not REPORT_LANGUAGE_PATTERN.fullmatch(language): raise ValueError( "project_progress.language must be a BCP-47-like language tag" ) @@ -126,7 +124,7 @@ def build_project_progress_periodic_report_source( if section_spec is None: raise ValueError(f"{label}.content_kind is invalid") visibility = str(item.get("visibility") or "primary").strip().lower() - if content_kind in _SUPPORTING_CONTENT_KINDS: + if content_kind in SUPPORTING_CONTENT_KINDS: supporting_count += 1 elif visibility == "supporting": supporting_count += 1 diff --git a/loopx/capabilities/periodic_report/request_action.py b/loopx/capabilities/periodic_report/request_action.py index b4013bd1d5..1a2786b2a1 100644 --- a/loopx/capabilities/periodic_report/request_action.py +++ b/loopx/capabilities/periodic_report/request_action.py @@ -16,6 +16,7 @@ from ...file_lock import exclusive_file_lock from ...history import load_registry from ...registry import atomic_write_json, find_registry_goal +from .field_shapes import REPORT_IDENTITY_PATTERN from .machine_defaults import resolve_goal_periodic_report_subscription from .machine_store import read_periodic_report_machine_defaults from .presets import build_periodic_report_preset_activation @@ -32,7 +33,6 @@ REQUEST_SETTLE_PORT = "periodic_report.request.settle_source" REQUEST_ADAPTER_PHASE = "capability_action" _REQUEST_FILE_RE = re.compile(r"^prq_[0-9a-f]{64}\.json$") -_IDENTITY_RE = re.compile(r"^[a-z][a-z0-9_.:-]{2,127}$") SourceBinder = Callable[..., Mapping[str, Any]] SourceSettler = Callable[..., Mapping[str, Any]] @@ -414,7 +414,7 @@ def record_periodic_report_request( source_adapter_id: str | None, execute: bool, ) -> dict[str, Any]: - if not _IDENTITY_RE.fullmatch(goal_id) or not _IDENTITY_RE.fullmatch(agent_id): + if not REPORT_IDENTITY_PATTERN.fullmatch(goal_id) or not REPORT_IDENTITY_PATTERN.fullmatch(agent_id): raise ValueError("periodic-report request Goal/Agent identity is invalid") opaque_ref = _source_ref(source_ref) selected_adapter_id = str(source_adapter_id or "").strip() @@ -674,7 +674,7 @@ def settle_periodic_report_request( or result.get("raw_content_returned") is not False or ( result.get("status") == SOURCE_SETTLEMENT_TERMINAL_STATUS - and not _IDENTITY_RE.fullmatch(str(result.get("failure_code") or "")) + and not REPORT_IDENTITY_PATTERN.fullmatch(str(result.get("failure_code") or "")) ) ): raise ValueError("periodic-report source settlement receipt is invalid") diff --git a/tests/architecture/test_periodic_report_field_shapes_owner.py b/tests/architecture/test_periodic_report_field_shapes_owner.py new file mode 100644 index 0000000000..1810a140c6 --- /dev/null +++ b/tests/architecture/test_periodic_report_field_shapes_owner.py @@ -0,0 +1,338 @@ +"""Guard the single owner of the periodic-report field rules. Refs #4447 Track A. + +`field_shapes` decides three things the package used to decide twice: the identity +rule that gates ``goal_id`` / ``agent_id`` / ``failure_code``, the BCP-47-like +language tag, and the content kinds that are supporting by nature. Two modules +compiled the language rule, two compiled the identity rule, and two listed the +supporting kinds -- and the two kind-literals differed only in member order, which +a value-keyed duplicate scan reads as two unrelated values. + +Two layers matter here because they catch different regressions. A name-keyed scan, +like the semantic inventory, is blind to a copy that renames the binding, and +`re.compile` constants never enter the inventory budgets at all, so nothing in +`vocabulary_v0.json` would notice a second spelling either. This guard therefore +compares compiled and set *values*, not names -- and that is how the third identity +site, bound as `_ID`, was found after the two `_IDENTITY_RE` copies were already +migrated. The entry-function cases then prove each face +really reads the owner instead of keeping an equivalent local rule. + +``incremental.py`` is declared, not migrated. Its ``_IDENTITY_RE`` is a different +rule (case-insensitive, up to 160 characters, versus lowercase-led 3..128), so it is +bounded-context reuse of one name and merging it would change what the incremental +reader accepts. The declaration pins both halves: it stays distinct, and it stays +the only such site. +""" + +from __future__ import annotations + +import ast +import pathlib +import re +from types import ModuleType + +import pytest + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +PACKAGE = REPO_ROOT / "loopx" / "capabilities" / "periodic_report" +OWNER_PATH = "loopx/capabilities/periodic_report/field_shapes.py" + +CONSUMERS: dict[str, tuple[str, ...]] = { + "loopx/capabilities/periodic_report/adapters.py": ( + "REPORT_LANGUAGE_PATTERN", + "SUPPORTING_CONTENT_KINDS", + ), + "loopx/capabilities/periodic_report/project_progress.py": ( + "REPORT_LANGUAGE_PATTERN", + "SUPPORTING_CONTENT_KINDS", + ), + "loopx/capabilities/periodic_report/cadence_journal.py": ("REPORT_IDENTITY_PATTERN",), +"loopx/capabilities/periodic_report/pending_intent.py": ("REPORT_IDENTITY_PATTERN",), + "loopx/capabilities/periodic_report/request_action.py": ("REPORT_IDENTITY_PATTERN",), +} +# Bounded-context reuse of one name: a distinct rule, declared rather than merged. +DECLARED_DISTINCT_SITES: dict[str, int] = { + "loopx/capabilities/periodic_report/incremental.py": 1, +} +RETIRED_LOCAL_NAMES = ("_IDENTITY_RE", "_LANGUAGE_RE", "_SUPPORTING_CONTENT_KINDS", "_ID") + +def _rel(path: pathlib.PurePath) -> str: + """Normalize a scanned site to the "/" spelling the tables in this file use. + + `str(Path.relative_to(root))` yields backslashes on Windows, which would make + these ownership assertions reject a correct tree on that platform. `as_posix()` + is stable across both spellings, and a Windows-style relative path is pinned + below so the substitution cannot be dropped again. + """ + + return path.as_posix() + + +def _package_sources() -> list[pathlib.Path]: + return sorted(p for p in PACKAGE.glob("*.py") if "__pycache__" not in p.parts) + + +def _module_assignments(path: pathlib.Path) -> list[tuple[str, int, ast.expr]]: + found: list[tuple[str, int, ast.expr]] = [] + for node in ast.parse(path.read_text(encoding="utf-8")).body: + if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance( + node.targets[0], ast.Name + ): + found.append((node.targets[0].id, node.lineno, node.value)) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + if node.value is not None: + found.append((node.target.id, node.lineno, node.value)) + return found + + +def _compiled_pattern(node: ast.expr) -> str | None: + """Fold a compiled pattern to its source text, however the call is spelled.""" + + if not isinstance(node, ast.Call) or not node.args: + return None + func = node.func + name = getattr(func, "attr", None) or getattr(func, "id", None) + if name != "compile": + return None + first = node.args[0] + if isinstance(first, ast.Constant) and isinstance(first.value, str): + return first.value + return ast.unparse(first) + + +def _set_members(node: ast.expr) -> tuple[str, ...] | None: + if isinstance(node, ast.Call) and node.args: + node = node.args[0] + if not isinstance(node, (ast.Set, ast.List, ast.Tuple)): + return None + return tuple(sorted(ast.unparse(element) for element in node.elts)) + + +def _pattern_definitions(pattern: re.Pattern[str]) -> list[str]: + """Sites that compile this pattern, at any depth and under any binding name. + + A module-level scan misses both failure modes this decision actually has: a + copy renamed to something else, and a copy inlined inside a function. + """ + + sites: set[str] = set() + for path in _package_sources(): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and _compiled_pattern(node) == pattern.pattern: + sites.add(_rel(path.relative_to(REPO_ROOT))) + return sorted(sites) + + +def _set_definitions(members: frozenset[str]) -> list[str]: + """Sites listing these members at any depth, in any order.""" + + wanted = tuple(sorted(f"'{value}'" for value in members)) + sites: set[str] = set() + for path in _package_sources(): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if _set_members(node) == wanted: + sites.add(_rel(path.relative_to(REPO_ROOT))) + return sorted(sites) + + +def _consumer(rel: str) -> ModuleType: + stem = pathlib.Path(rel).stem + return __import__(f"loopx.capabilities.periodic_report.{stem}", fromlist=["_"]) + + +@pytest.fixture(scope="module") +def owner() -> ModuleType: + from loopx.capabilities.periodic_report import field_shapes + + return field_shapes + + +def test_scanned_sites_use_one_path_spelling(owner: ModuleType) -> None: + """A Windows-style relative path must fold to the same site string as POSIX.""" + + import ntpath + import posixpath + + windows_relative = pathlib.PureWindowsPath(ntpath.join("loopx", "x.py")) + posix_relative = pathlib.PurePosixPath(posixpath.join("loopx", "x.py")) + assert str(pathlib.PureWindowsPath(ntpath.join("loopx", "x.py"))).count("\\") == 1 + assert _rel(windows_relative) == _rel(posix_relative) == "loopx/x.py" + sites = _pattern_definitions(owner.REPORT_IDENTITY_PATTERN) + _set_definitions( + owner.SUPPORTING_CONTENT_KINDS + ) + assert all("\\" not in site for site in sites), sites + assert OWNER_PATH in sites + + +def test_each_pattern_is_compiled_once_in_the_package(owner: ModuleType) -> None: + for attribute in ("REPORT_IDENTITY_PATTERN", "REPORT_LANGUAGE_PATTERN"): + pattern: re.Pattern[str] = getattr(owner, attribute) + sites = _pattern_definitions(pattern) + assert sites == [OWNER_PATH], ( + f"{attribute} is compiled in {sites}; only {OWNER_PATH} may state it" + ) + + +def test_supporting_kinds_are_listed_once_including_reordered_copies( + owner: ModuleType, +) -> None: + sites = _set_definitions(owner.SUPPORTING_CONTENT_KINDS) + assert sites == [OWNER_PATH], ( + f"SUPPORTING_CONTENT_KINDS members are listed in {sites}; " + "a copy that only reorders the members counts as a second owner too" + ) + + +@pytest.mark.parametrize("rel", sorted(CONSUMERS), ids=sorted(CONSUMERS)) +def test_consumers_hold_the_owner_objects_not_copies(rel: str, owner: ModuleType) -> None: + module = _consumer(rel) + for attribute in CONSUMERS[rel]: + assert getattr(module, attribute) is getattr(owner, attribute), ( + f"{pathlib.Path(rel).stem}.{attribute} is a second binding, not the owner's object" + ) + + +@pytest.mark.parametrize("rel", sorted(CONSUMERS), ids=sorted(CONSUMERS)) +def test_each_consumer_uses_the_owner_name_beyond_the_import(rel: str) -> None: + """An import that is never used would leave the local decision in place.""" + + lines = (PACKAGE / pathlib.Path(rel).name).read_text(encoding="utf-8").splitlines() + for attribute in CONSUMERS[rel]: + used = [ + line + for line in lines + if attribute in line and not line.lstrip().startswith(("from ", "import ")) + ] + assert used, f"{rel} imports {attribute} but never applies it" + + +@pytest.mark.parametrize("rel", sorted(CONSUMERS), ids=sorted(CONSUMERS)) +def test_retired_local_names_are_not_reintroduced(rel: str) -> None: + for name, _, _ in _module_assignments(PACKAGE / pathlib.Path(rel).name): + assert name not in RETIRED_LOCAL_NAMES, ( + f"{rel} defines {name} again; import the owner's name instead" + ) + + +def test_declared_distinct_site_stays_distinct_and_stays_declared( + owner: ModuleType, +) -> None: + from loopx.capabilities.periodic_report import incremental + + assert incremental._IDENTITY_RE.pattern != owner.REPORT_IDENTITY_PATTERN.pattern, ( + "incremental's identity rule now equals the owner's: migrate it and drop the declaration" + ) + declared = sorted( + _rel(path.relative_to(REPO_ROOT)) + for path in _package_sources() + for name, _, _ in _module_assignments(path) + if name == "_IDENTITY_RE" + ) + assert declared == sorted(DECLARED_DISTINCT_SITES), ( + f"_IDENTITY_RE is declared in {declared}; the declaration is {sorted(DECLARED_DISTINCT_SITES)}" + ) + + +def test_language_face_in_project_progress_reads_the_owner(owner: ModuleType) -> None: + from loopx.capabilities.periodic_report import project_progress + + assert project_progress._language("EN-US") == "en-us" + for rejected in ("en!", "e", "en-", "language with spaces"): + assert not owner.REPORT_LANGUAGE_PATTERN.fullmatch(rejected) + with pytest.raises(ValueError, match="BCP-47-like"): + project_progress._language(rejected) + + +def test_language_face_in_adapters_reads_the_owner(owner: ModuleType) -> None: + from loopx.capabilities.periodic_report import adapters + + assert adapters._normalize_editorial({"language": "zh-Hans"}) == {"language": "zh-Hans"} + assert adapters._normalize_editorial({"language": "en", "kicker": "k"}) == { + "kicker": "k", + "language": "en", + } + with pytest.raises(ValueError, match="BCP-47-like"): + adapters._normalize_editorial({"language": "zh_Hans"}) + + +def test_identity_face_in_cadence_journal_rejects_what_the_owner_rejects( + tmp_path: pathlib.Path, + owner: ModuleType, +) -> None: + from loopx.capabilities.periodic_report import cadence_journal + + # The site this guard found only under a different binding name: ``_ID``. + assert cadence_journal.cadence_journal_path(tmp_path, "goal-1").is_relative_to(tmp_path) + with pytest.raises(ValueError, match="Goal identity is invalid"): + cadence_journal.cadence_journal_path(tmp_path, "Goal-1") + assert not owner.REPORT_IDENTITY_PATTERN.fullmatch("Goal-1") + + +def test_supporting_kind_membership_is_enforced_through_the_adapters_face( + owner: ModuleType, +) -> None: + from loopx.capabilities.periodic_report import adapters + + for kind in sorted(owner.SUPPORTING_CONTENT_KINDS): + with pytest.raises(ValueError, match="must be supporting"): + adapters._normalize_item( + { + "item_id": "item1", + "title": "t", + "summary": "s", + "content_kind": kind, + "visibility": "primary", + }, + label="item", + ) + accepted = adapters._normalize_item( + { + "item_id": "item1", + "title": "t", + "summary": "s", + "content_kind": sorted(owner.SUPPORTING_CONTENT_KINDS)[0], + "visibility": "supporting", + }, + label="item", + ) + assert accepted["visibility"] == "supporting" + + +def test_identity_face_rejects_what_the_owner_rejects( + tmp_path: pathlib.Path, + owner: ModuleType, +) -> None: + from loopx.capabilities.periodic_report import pending_intent, request_action + + assert pending_intent.pending_periodic_report_intents( + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path, + goal_id="Uppercase-led-identity", + agent_id="agent-a", + ) == [] + with pytest.raises(ValueError, match="identity is invalid"): + request_action.record_periodic_report_request( + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path, + goal_id="Uppercase-led-identity", + agent_id="agent-a", + source_ref="source-ref", + request_ports=None, + source_adapter_id=None, + execute=False, + ) + assert not owner.REPORT_IDENTITY_PATTERN.fullmatch("Uppercase-led-identity") + + +def test_incremental_face_keeps_its_own_wider_rule(owner: ModuleType) -> None: + from loopx.capabilities.periodic_report import incremental + + accepted_by_incremental = "Abc1" + assert ( + incremental._identity_text(accepted_by_incremental, "surface") + == accepted_by_incremental + ) + assert not owner.REPORT_IDENTITY_PATTERN.fullmatch(accepted_by_incremental), ( + "the two rules no longer differ; remove the declaration above and migrate it" + )