diff --git a/loopx/extensions/lark/event_collector.py b/loopx/extensions/lark/event_collector.py index 5ed433d67c..1d724884a7 100644 --- a/loopx/extensions/lark/event_collector.py +++ b/loopx/extensions/lark/event_collector.py @@ -18,6 +18,7 @@ inspect_lark_event_inbox, load_lark_event_inbox_config, ) +from .identity_shapes import LARK_CHAT_ID_PATTERN CONFIG_SCHEMA_VERSION_V0 = "lark_event_collector_config_v0" CONFIG_SCHEMA_VERSION = "lark_event_collector_config_v1" @@ -29,7 +30,6 @@ STATUS_SCHEMA_VERSION = "lark_event_collector_status_v0" INSTALL_SCHEMA_VERSION = "lark_event_collector_install_v0" SERVICE_RE = re.compile(r"^loopx-[a-z0-9][a-z0-9._-]{1,73}$") -CHAT_RE = re.compile(r"^oc_[A-Za-z0-9_-]+$") TIMEOUT_RE = re.compile(r"^[1-9][0-9]*(?:s|m|h)$") SUPPORTED_SUPERVISORS = {"launchd", "systemd"} SUPPORTED_EVENT_KEY = "im.message.receive_v1" @@ -231,7 +231,7 @@ def load_lark_event_collector_config( "public-safe token" ) chat_id = str(raw_route.get("chat_id") or "").strip() - if not CHAT_RE.fullmatch(chat_id): + if not LARK_CHAT_ID_PATTERN.fullmatch(chat_id): raise ValueError( f"collector route {index + 1} chat_id must be a Lark oc_ chat id" ) diff --git a/loopx/extensions/lark/event_collector_routes.py b/loopx/extensions/lark/event_collector_routes.py index 7329c1aca3..703fcbb24e 100644 --- a/loopx/extensions/lark/event_collector_routes.py +++ b/loopx/extensions/lark/event_collector_routes.py @@ -12,7 +12,6 @@ from ...file_lock import exclusive_file_lock from .event_collector import ( - CHAT_RE, CONFIG_SCHEMA_VERSION, MAX_ROUTE_COUNT, _project_config_path, @@ -23,6 +22,7 @@ ROUTE_KEY_PATTERN, load_lark_event_inbox_config, ) +from .identity_shapes import LARK_CHAT_ID_PATTERN ROUTE_RECONCILE_SCHEMA_VERSION = "lark_event_collector_route_reconcile_v0" @@ -38,7 +38,7 @@ def _route_reconcile_candidate( raise ValueError("collector route reconcile requires a v1 collector config") if not ROUTE_KEY_PATTERN.fullmatch(route_key): raise ValueError("route_key must be a lowercase public-safe token") - if not CHAT_RE.fullmatch(chat_id): + if not LARK_CHAT_ID_PATTERN.fullmatch(chat_id): raise ValueError("chat_id must be a Lark oc_ chat id") root = Path(config["project"]) inbox_ref, inbox_config_path = _relative_project_path( diff --git a/loopx/extensions/lark/event_inbox.py b/loopx/extensions/lark/event_inbox.py index b8b7da1130..1060d72cda 100644 --- a/loopx/extensions/lark/event_inbox.py +++ b/loopx/extensions/lark/event_inbox.py @@ -29,15 +29,17 @@ OPEN_ID_PATTERN, lark_provider_mention_identities, ) +from .identity_shapes import ( # noqa: F401 + LARK_CHAT_ID_PATTERN as CHAT_ID_PATTERN, + LARK_MESSAGE_ID_PATTERN as MESSAGE_ID_PATTERN, +) EVENT_SCHEMA_VERSION = "lark_event_inbox_event_v0" CONFIG_SCHEMA_VERSION = "lark_event_inbox_config_v0" PROCESSED_SCHEMA_VERSION = "lark_event_inbox_processed_v0" MATERIAL_REVIEW_LEDGER_SCHEMA_VERSION = "lark_material_review_ledger_v0" -MESSAGE_ID_PATTERN = re.compile(r"om_[A-Za-z0-9_-]+") EVENT_ID_PATTERN = re.compile(r"[A-Za-z0-9:_-]{1,200}") SAFE_PROFILE_PATTERN = re.compile(r"[A-Za-z0-9._-]{1,100}") -CHAT_ID_PATTERN = re.compile(r"oc_[A-Za-z0-9_-]+") REACTION_EMOJI_PATTERN = re.compile(r"[A-Za-z0-9_]{1,64}") REPLY_PLACEMENT_POLICIES = {"source_thread", "source_context"} REPLY_EDITORIAL_STYLES = {"concise", "bullet_points_preferred"} diff --git a/loopx/extensions/lark/goal_channel_delivery_contract.py b/loopx/extensions/lark/goal_channel_delivery_contract.py index 3da9acb42a..ea12b3e820 100644 --- a/loopx/extensions/lark/goal_channel_delivery_contract.py +++ b/loopx/extensions/lark/goal_channel_delivery_contract.py @@ -6,9 +6,9 @@ from collections.abc import Callable, Mapping from typing import Any +from .identity_shapes import LARK_CHAT_ID_PATTERN _GOAL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$") -_LARK_CHAT_ID_RE = re.compile(r"^oc_[A-Za-z0-9_-]+$") _LARK_APP_ID_RE = re.compile(r"^cli_[A-Za-z0-9_-]+$") _LARK_PROFILE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$") @@ -42,7 +42,7 @@ def goal_channel_delivery_route( if ( identity.get("mode") != "project_bot" or sender_identity != "bot" - or not _LARK_CHAT_ID_RE.fullmatch(chat_id) + or not LARK_CHAT_ID_PATTERN.fullmatch(chat_id) or not _LARK_PROFILE_RE.fullmatch(sender_profile) or sender_profile.lower() == "default" or not _LARK_APP_ID_RE.fullmatch(bot_app_id) diff --git a/loopx/extensions/lark/goal_channel_operation.py b/loopx/extensions/lark/goal_channel_operation.py index 41626c66b2..c32eee39e4 100644 --- a/loopx/extensions/lark/goal_channel_operation.py +++ b/loopx/extensions/lark/goal_channel_operation.py @@ -5,7 +5,6 @@ import hashlib import html import json -import re from collections.abc import Callable, Mapping from pathlib import Path from typing import Any @@ -36,16 +35,13 @@ GoalChannelMessageDeliverySession, resolve_bound_goal_channel, ) +from .identity_shapes import require_card_callback_identity from .presentation.kanban import CommandRunner, default_subprocess_runner OPERATION_CARD_ACTION_SCHEMA_VERSION = "loopx_operation_card_action_v0" OPERATION_CALLBACK_RECEIPT_SCHEMA_VERSION = "lark_operation_callback_receipt_v0" OPERATION_EXECUTOR_CAPABILITY_ID = "human-confirmed-operation-executor" -_EVENT_ID = re.compile(r"^[A-Za-z0-9._:-]{1,240}$") -_MESSAGE_ID = re.compile(r"^om_[A-Za-z0-9_-]+$") -_CHAT_ID = re.compile(r"^oc_[A-Za-z0-9_-]+$") -_OPEN_ID = re.compile(r"^ou_[A-Za-z0-9_-]+$") def _digest(value: object) -> str: @@ -990,14 +986,7 @@ def handle_goal_channel_operation_callback( or any(ord(character) < 32 for character in callback_token) ): raise ValueError("operation callback update token is invalid") - for field, pattern in ( - ("event_id", _EVENT_ID), - ("message_id", _MESSAGE_ID), - ("chat_id", _CHAT_ID), - ("operator_id", _OPEN_ID), - ): - if not pattern.fullmatch(str(event.get(field) or "")): - raise ValueError(f"operation callback {field} is invalid") + require_card_callback_identity(event, error_prefix="operation callback") if str(event.get("host") or "") != "im_message": raise ValueError("operation callback host is unsupported") store = ChatActionStore(action_store_root) diff --git a/loopx/extensions/lark/goal_channel_transport.py b/loopx/extensions/lark/goal_channel_transport.py index b4c79a21a1..e09f8a68ba 100644 --- a/loopx/extensions/lark/goal_channel_transport.py +++ b/loopx/extensions/lark/goal_channel_transport.py @@ -7,12 +7,16 @@ from enum import Enum from typing import Any +# Re-exported for the existing callers of this module; the shapes themselves are +# decided once, in ``identity_shapes``. +from .identity_shapes import ( # noqa: F401 + LARK_CHAT_ID_SEARCH as CHAT_ID_PATTERN, + LARK_MESSAGE_ID_SEARCH as MESSAGE_ID_PATTERN, + LARK_OPEN_ID_SEARCH as OPEN_ID_PATTERN, +) from .presentation.kanban import CommandRunner -CHAT_ID_PATTERN = re.compile(r"oc_[A-Za-z0-9_-]+") -MESSAGE_ID_PATTERN = re.compile(r"om_[A-Za-z0-9_-]+") APP_ID_PATTERN = re.compile(r"cli_[A-Za-z0-9_-]+") -OPEN_ID_PATTERN = re.compile(r"ou_[A-Za-z0-9_-]+") SAFE_PROFILE_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}") REQUIRED_GOAL_TOPIC_SCOPES = ("im:message", "im:message:readonly") REQUIRED_BOT_GROUP_HISTORY_SCOPES = ( diff --git a/loopx/extensions/lark/identity_shapes.py b/loopx/extensions/lark/identity_shapes.py new file mode 100644 index 0000000000..8aa2eb8632 --- /dev/null +++ b/loopx/extensions/lark/identity_shapes.py @@ -0,0 +1,49 @@ +"""One owner for the whole-value shapes of Lark platform identifiers. + +A card callback names four identifiers: the event that triggered it, the message +the card lives in, the chat that message belongs to, and the operator who +clicked. Two callback validators each decided independently what a valid one +looks like, so a shape fix in one of them silently leaves the other behind. + +Searching for an identifier inside larger text is the same decision about a +different question, so it is stated here too and nowhere else: +``goal_channel_transport`` and ``event_inbox`` each compiled the unanchored +spelling themselves while fifteen and thirteen modules respectively imported it +from them, which is how a shape fix could land in one and be missed in the +other. The unanchored forms are not interchangeable with the anchored ones -- +``goal_channel_contracts`` and ``goal_channel_notification`` ``search()`` for an +id inside payload text, where ``^`` and ``$`` would change the answer -- so both +spellings stay distinct and one module decides both. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from typing import Any + +LARK_EVENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,240}$") +LARK_MESSAGE_ID_PATTERN = re.compile(r"^om_[A-Za-z0-9_-]+$") +LARK_CHAT_ID_PATTERN = re.compile(r"^oc_[A-Za-z0-9_-]+$") +LARK_OPEN_ID_PATTERN = re.compile(r"^ou_[A-Za-z0-9_-]+$") + +LARK_MESSAGE_ID_SEARCH = re.compile(r"om_[A-Za-z0-9_-]+") +LARK_CHAT_ID_SEARCH = re.compile(r"oc_[A-Za-z0-9_-]+") +LARK_OPEN_ID_SEARCH = re.compile(r"ou_[A-Za-z0-9_-]+") + +CARD_CALLBACK_IDENTITY_SHAPES = ( + ("event_id", LARK_EVENT_ID_PATTERN), + ("message_id", LARK_MESSAGE_ID_PATTERN), + ("chat_id", LARK_CHAT_ID_PATTERN), + ("operator_id", LARK_OPEN_ID_PATTERN), +) + + +def require_card_callback_identity( + event: Mapping[str, Any], *, error_prefix: str +) -> None: + """Reject a card callback whose identity fields are not whole Lark ids.""" + + for field, pattern in CARD_CALLBACK_IDENTITY_SHAPES: + if not pattern.fullmatch(str(event.get(field) or "")): + raise ValueError(f"{error_prefix} {field} is invalid") diff --git a/loopx/extensions/lark/reviewer_notification.py b/loopx/extensions/lark/reviewer_notification.py index 86eea75dcb..5d015625dd 100644 --- a/loopx/extensions/lark/reviewer_notification.py +++ b/loopx/extensions/lark/reviewer_notification.py @@ -13,6 +13,11 @@ reviewer_notification_idempotency_key, ) from ...control_plane.runtime.public_safety import public_safe_compact_text +from .identity_shapes import ( + LARK_CHAT_ID_PATTERN, + LARK_MESSAGE_ID_PATTERN, + LARK_OPEN_ID_PATTERN, +) from .outbound import ( LarkMention, build_lark_mention_prefix, @@ -30,9 +35,6 @@ r"search:message|missing\s+scope[^\n]*(?:message|search)", re.IGNORECASE, ) -LARK_DESTINATION_PATTERN = re.compile(r"oc_[A-Za-z0-9_-]+") -LARK_MEMBER_PATTERN = re.compile(r"ou_[A-Za-z0-9_-]+") -LARK_MESSAGE_PATTERN = re.compile(r"om_[A-Za-z0-9_-]+") def _normalise_handle(value: Any) -> str | None: @@ -43,7 +45,7 @@ def _normalise_handle(value: Any) -> str | None: def _find_message_id(value: Any) -> str | None: if isinstance(value, Mapping): candidate = value.get("message_id") - if isinstance(candidate, str) and LARK_MESSAGE_PATTERN.fullmatch(candidate): + if isinstance(candidate, str) and LARK_MESSAGE_ID_PATTERN.fullmatch(candidate): return candidate for nested in value.values(): found = _find_message_id(nested) @@ -99,7 +101,7 @@ def visit(node: Any) -> None: for key, child in node.items(): if key in {"member_id", "open_id"} and isinstance(child, str): member_id = child.strip() - if LARK_MEMBER_PATTERN.fullmatch(member_id): + if LARK_OPEN_ID_PATTERN.fullmatch(member_id): member_ids.add(member_id) else: visit(child) @@ -179,7 +181,7 @@ def lark_reviewer_notification_sink( external_write_authority_asserted=execute, blocker="dedicated_bot_identity_required", ) - if not LARK_DESTINATION_PATTERN.fullmatch(destination_id): + if not LARK_CHAT_ID_PATTERN.fullmatch(destination_id): return build_reviewer_notification_sink_result( sink_kind=sink_kind, reviewer_handles=[], @@ -207,7 +209,7 @@ def lark_reviewer_notification_sink( identity.get("display_name") or handle, limit=80, ) - if not LARK_MEMBER_PATTERN.fullmatch(member_id): + if not LARK_OPEN_ID_PATTERN.fullmatch(member_id): return build_reviewer_notification_sink_result( sink_kind=sink_kind, reviewer_handles=[], diff --git a/loopx/extensions/lark/team_plan_confirmation.py b/loopx/extensions/lark/team_plan_confirmation.py index 99625129c5..8591929660 100644 --- a/loopx/extensions/lark/team_plan_confirmation.py +++ b/loopx/extensions/lark/team_plan_confirmation.py @@ -6,7 +6,6 @@ import hashlib import json import logging -import re import subprocess import threading from collections.abc import Callable, Mapping, Sequence @@ -46,6 +45,11 @@ goal_channel_target_for_name, read_goal_channel_targets, ) +from .identity_shapes import ( + LARK_CHAT_ID_PATTERN, + LARK_MESSAGE_ID_PATTERN, + require_card_callback_identity, +) from .manager_reply_delivery import ( TEAM_PLAN_DELIVERY_RECEIPT_SCHEMA_VERSION, validate_team_plan_delivery_receipt, @@ -60,10 +64,6 @@ TEAM_PLAN_CALLBACK_RECEIPT_SCHEMA_VERSION = "lark_team_plan_callback_receipt_v0" -_EVENT_ID = re.compile(r"^[A-Za-z0-9._:-]{1,240}$") -_MESSAGE_ID = re.compile(r"^om_[A-Za-z0-9_-]+$") -_CHAT_ID = re.compile(r"^oc_[A-Za-z0-9_-]+$") -_OPEN_ID = re.compile(r"^ou_[A-Za-z0-9_-]+$") _EVENT_DIAGNOSTIC_PREFIX = "[event] " ProcessFactory = Callable[[list[str]], Any] @@ -141,7 +141,7 @@ def active_profile_chat_ids(snapshot: Mapping[str, Any], profile: str) -> list[s same_app = bool(app_ids) and str(identity.get("bot_app_id") or "") in app_ids if ( same_app or str(identity.get("sender_profile") or "") == profile - ) and _CHAT_ID.fullmatch(chat_id): + ) and LARK_CHAT_ID_PATTERN.fullmatch(chat_id): chats.add(chat_id) return sorted(chats) @@ -459,7 +459,8 @@ def _deliver_one( "submitted_card": card, "authorized_principal": authorized_principal, } - if not _MESSAGE_ID.fullmatch(str(recorded.get("message_id") or "")) or any( + recorded_message_id = str(recorded.get("message_id") or "") + if not LARK_MESSAGE_ID_PATTERN.fullmatch(recorded_message_id) or any( recorded.get(key) != value for key, value in expected.items() ): raise ActionConflictError( @@ -691,14 +692,7 @@ def handle_team_plan_review_callback( or any(ord(character) < 32 for character in callback_token) ): raise ValueError("team plan callback update token is invalid") - for field, pattern in ( - ("event_id", _EVENT_ID), - ("message_id", _MESSAGE_ID), - ("chat_id", _CHAT_ID), - ("operator_id", _OPEN_ID), - ): - if not pattern.fullmatch(str(event.get(field) or "")): - raise ValueError(f"team plan callback {field} is invalid") + require_card_callback_identity(event, error_prefix="team plan callback") if str(event.get("host") or "") != "im_message": raise ValueError("team plan callback host is unsupported") store = ChatActionStore(action_store_root) diff --git a/tests/architecture/test_lark_identity_shape_owner.py b/tests/architecture/test_lark_identity_shape_owner.py new file mode 100644 index 0000000000..acea4b8a76 --- /dev/null +++ b/tests/architecture/test_lark_identity_shape_owner.py @@ -0,0 +1,843 @@ +"""Guard the single owner of the whole-value Lark identifier shapes. + +Three questions, in the order a reviewer will ask them: + +1. Is there a second definition of one of these identifier shapes? Decided on + the folded value of every regular-expression construction in the product + tree, not on spelling, so ``re.compile(PATTERN)``, an inline ``re.fullmatch`` + and a concatenated literal are all offenders. +2. Is every consumer wired to the owner? Decided by object identity plus an + actual reference in the module body, so a kept-alive import that no longer + makes the decision is still an offender. +3. Are the two *uses* of a shape kept apart? A whole-value check and a search + for an id inside larger text are different questions, so the owner states + both spellings: four anchored patterns and three unanchored ones. Either + spelling anywhere else is an offender, and so is converting a declared site + without retiring its declaration. +""" + +from __future__ import annotations + +import ast +import pathlib +from types import ModuleType + +import pytest + +from loopx.extensions.lark import ( + event_collector, + event_collector_routes, + event_inbox, + goal_channel_delivery_contract, + goal_channel_operation, + goal_channel_transport, + identity_shapes, + reviewer_notification, + team_plan_confirmation, +) + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +LARK_PACKAGE = REPO_ROOT / "loopx" / "extensions" / "lark" +OWNER_MODULE = "loopx/extensions/lark/identity_shapes.py" +TRANSPORT_HUB = "loopx/extensions/lark/goal_channel_transport.py" +INBOX_HUB = "loopx/extensions/lark/event_inbox.py" + +# The four identifier bodies as the owner states them, without anchors. The +# substring in the second column is what a module has to contain to be worth +# parsing at all; it is load-bearing, so a test below re-checks the mapping. +WHOLE_VALUE_BODIES = { + "event_id": r"[A-Za-z0-9._:-]{1,240}", + "message_id": r"om_[A-Za-z0-9_-]+", + "chat_id": r"oc_[A-Za-z0-9_-]+", + "operator_id": r"ou_[A-Za-z0-9_-]+", +} +# A module has to import the regex machinery before it can decide a shape, and +# both spellings this scan accepts (``re.X(...)`` and a name from +# ``from re import X``) require one of these two lines. Screening on the bodies +# themselves would be unsound: a pattern assembled from two literals contains no +# single body, which is one of the probe cases below. +PRESCREEN_TOKENS = ("import re", "from re import") +# Only these three bodies have a search spelling. An event id is never looked +# for inside larger text. +SEARCH_IDENTIFIERS = {"chat_id", "message_id", "operator_id"} +ANCHORED_EXPORTS = { + "event_id": "LARK_EVENT_ID_PATTERN", + "message_id": "LARK_MESSAGE_ID_PATTERN", + "chat_id": "LARK_CHAT_ID_PATTERN", + "operator_id": "LARK_OPEN_ID_PATTERN", +} +SEARCH_EXPORTS = { + "message_id": "LARK_MESSAGE_ID_SEARCH", + "chat_id": "LARK_CHAT_ID_SEARCH", + "operator_id": "LARK_OPEN_ID_SEARCH", +} + +# Sites that still decide an identifier shape for themselves, with the exact +# number of constructions allowed there. Converting one deletes its entry, and +# editing one past its budget is a review event. +# Regex constructions in a marker file whose pattern cannot be folded. There +# are none today, so the list is closed: a future indirect construction has to +# state the file, the count and the shape it answers, instead of hiding inside +# a scan that cannot see it. +DECLARED_UNFOLDABLE_SITES: dict[str, int] = { + # A mention lookup built from a runtime display name. It searches text for + # ``@handle`` and decides no identifier shape at all. + "loopx/extensions/lark/event_inbox.py": 1, + # Keyword routing matches a caller-supplied pattern against message content. + # The pattern is data at this boundary, and no identifier shape is decided. + "loopx/extensions/lark/goal_topic_routing.py": 1, +} +# A file only enters the unfoldable layer when it plausibly touches these ids. +# Without the gate the layer reports every regex built from data in the package +# (setup URLs, markdown headings) and the declaration becomes noise. +IDENTIFIER_RELEVANCE_TOKENS = ( + "oc_", + "om_", + "ou_", + "chat_id", + "message_id", + "operator_id", + "event_id", +) + +DECLARED_INDIVIDUAL_SITES: dict[str, int] = { + # ``re.fullmatch(r"[A-Za-z0-9._:-]{1,240}", ...)`` against an event id. The + # in-flight goal-channel claim work restructures this file, so the site is + # declared here instead of racing that branch. + "loopx/extensions/lark/event_collector_runtime.py": 1, +} + +SHAPE_CONSUMERS: dict[str, tuple[ModuleType, str]] = { + "loopx/extensions/lark/goal_channel_operation.py": ( + goal_channel_operation, + "require_card_callback_identity", + ), + "loopx/extensions/lark/team_plan_confirmation.py": ( + team_plan_confirmation, + "require_card_callback_identity", + ), + "loopx/extensions/lark/event_collector.py": (event_collector, "LARK_CHAT_ID_PATTERN"), + "loopx/extensions/lark/goal_channel_delivery_contract.py": ( + goal_channel_delivery_contract, + "LARK_CHAT_ID_PATTERN", + ), + "loopx/extensions/lark/event_collector_routes.py": ( + event_collector_routes, + "LARK_CHAT_ID_PATTERN", + ), + "loopx/extensions/lark/reviewer_notification.py": ( + reviewer_notification, + "LARK_OPEN_ID_PATTERN", + ), +} +HUB_REEXPORTS: dict[str, tuple[ModuleType, tuple[str, ...]]] = { + INBOX_HUB: (event_inbox, ("CHAT_ID_PATTERN", "MESSAGE_ID_PATTERN")), + TRANSPORT_HUB: ( + goal_channel_transport, + ("CHAT_ID_PATTERN", "MESSAGE_ID_PATTERN", "OPEN_ID_PATTERN"), + ), +} +_RE_MODULE = "re" +_REGEX_METHODS = { + "compile", + "fullmatch", + "match", + "search", + "findall", + "finditer", + "sub", + "subn", + "split", +} + + +FOLD_DEPTH_LIMIT = 4 + + +def fold_string(node: ast.AST, constants: dict[str, str], depth: int = 0) -> str | None: + """Fold a literal-valued expression, or return None when it is not literal. + + ``constants`` are the module-level string bindings of the file being read, + so a pattern assembled out of same-file parts is judged on its folded value. + Only module-level bindings are trusted, and only to a bounded depth: a + function-local re-binding of the same name is deliberately not folded, and + lands in the unfoldable declaration below instead of silently folding to the + wrong value. + """ + + if isinstance(node, ast.Constant): + return node.value if isinstance(node.value, str) else None + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for value in node.values: + if isinstance(value, ast.Constant) and isinstance(value.value, str): + parts.append(value.value) + continue + return None + return "".join(parts) + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = fold_string(node.left, constants, depth) + right = fold_string(node.right, constants, depth) + return None if left is None or right is None else left + right + if isinstance(node, ast.Name) and depth < FOLD_DEPTH_LIMIT: + return constants.get(node.id) + return None + + +def untrusted_names(tree: ast.Module) -> set[str]: + """Names this scan will not fold: anything re-bound or shadowed anywhere. + + A module constant whose name is also a parameter, a function, an import or a + second assignment anywhere in the file cannot be trusted to hold one value, + so a pattern built from it becomes an unfoldable site that has to be + declared. Folding it anyway is how a second owner hides. + """ + + module_targets: dict[str, int] = {} + for node in tree.body: + if isinstance(node, ast.Assign) and len(node.targets) == 1: + if isinstance(node.targets[0], ast.Name): + name = node.targets[0].id + module_targets[name] = module_targets.get(name, 0) + 1 + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + name = node.target.id + module_targets[name] = module_targets.get(name, 0) + 1 + + untrusted = {name for name, count in module_targets.items() if count > 1} + for node in ast.walk(tree): + if isinstance(node, ast.arg): + untrusted.add(node.arg) + elif isinstance(node, (ast.Import, ast.ImportFrom)): + untrusted.update(alias.asname or alias.name for alias in node.names) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + untrusted.add(node.name) + elif isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): + if node.id not in module_targets: + untrusted.add(node.id) + return untrusted + + +def module_level_string_constants(tree: ast.Module) -> dict[str, str]: + """Name -> folded value for module-level string constants, resolved bounded.""" + + trusted = untrusted_names(tree) + pending: list[tuple[str, ast.AST]] = [] + constants: dict[str, str] = {} + for node in tree.body: + if isinstance(node, ast.Assign) and len(node.targets) == 1: + target = node.targets[0] + if ( + isinstance(target, ast.Name) + and isinstance(target.id, str) + and target.id not in trusted + ): + pending.append((target.id, node.value)) + for _ in range(FOLD_DEPTH_LIMIT): + unresolved: list[tuple[str, ast.AST]] = [] + for name, value in pending: + folded = fold_string(value, {k: v for k, v in constants.items() if k != name}) + if folded is None: + unresolved.append((name, value)) + else: + constants[name] = folded + pending = unresolved + if not pending: + break + return constants + + +def regex_pattern_argument(node: ast.Call) -> ast.AST | None: + """Return this call's pattern argument when it builds or applies a regex.""" + + func = node.func + if isinstance(func, ast.Attribute): + holder = func.value + if not (isinstance(holder, ast.Name) and holder.id == _RE_MODULE): + return None + return func.attr if func.attr in _REGEX_METHODS else None + if isinstance(func, ast.Name): + # Covers ``from re import fullmatch`` style call sites. + return func.id if func.id in _REGEX_METHODS else None + return None + + +def regex_constructions(tree: ast.Module) -> list[ast.Call]: + calls: list[ast.Call] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call) and node.args and regex_pattern_argument(node): + calls.append(node) + return calls + + +def shape_rows_in_source(source: str, name: str) -> list[dict[str, object]]: + """Every whole-value identifier shape decided inside ``source``.""" + + if not any(token in source for token in PRESCREEN_TOKENS): + return [] + tree = ast.parse(source) + constants = module_level_string_constants(tree) + rows: list[dict[str, object]] = [] + for node in regex_constructions(tree): + pattern = fold_string(node.args[0], constants) + if pattern is None: + continue + body = pattern.removeprefix("^").removesuffix("$") + identifier = next( + (key for key, shape in WHOLE_VALUE_BODIES.items() if shape == body), None + ) + if identifier is None: + continue + rows.append( + { + "file": name, + "line": getattr(node, "lineno", 0), + "identifier": identifier, + "anchored": body != pattern, + } + ) + return sorted(rows, key=lambda row: (str(row["file"]), int(row["line"]))) + + +def unfoldable_rows_in_source(source: str, name: str) -> list[dict[str, object]]: + """Regex constructions in ``source`` whose pattern value cannot be folded.""" + + if not any(token in source for token in PRESCREEN_TOKENS): + return [] + if not any(token in source for token in IDENTIFIER_RELEVANCE_TOKENS): + return [] + tree = ast.parse(source) + constants = module_level_string_constants(tree) + return [ + {"file": name, "line": node.lineno, "unfoldable": True} + for node in regex_constructions(tree) + if fold_string(node.args[0], constants) is None + ] + + +def collect_source_rows(root: pathlib.Path) -> list[tuple[str, str]]: + pairs: list[tuple[str, str]] = [] + for path in sorted(root.rglob("*.py")): + if "__pycache__" in path.parts: + continue + try: + source = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + pairs.append((path.relative_to(REPO_ROOT).as_posix(), source)) + return pairs + + +def collect_shape_definitions(root: pathlib.Path) -> list[dict[str, object]]: + return [ + row + for name, source in collect_source_rows(root) + for row in shape_rows_in_source(source, name) + ] + + +def collect_unfoldable_definitions(root: pathlib.Path) -> list[dict[str, object]]: + return [ + row + for name, source in collect_source_rows(root) + for row in unfoldable_rows_in_source(source, name) + ] + + +def offender_rows( + definitions: list[dict[str, object]], + *, + declared: dict[str, int] | None = None, +) -> list[str]: + """Violations of the one-owner rule, as readable strings.""" + + budget = DECLARED_INDIVIDUAL_SITES if declared is None else declared + rows: list[str] = [] + counted: dict[str, int] = {} + for item in definitions: + file = str(item["file"]) + identifier = str(item["identifier"]) + anchored = bool(item["anchored"]) + if file in budget: + counted[file] = counted.get(file, 0) + 1 + continue + spelling = "whole-value" if anchored else "search" + if file != OWNER_MODULE: + rows.append(f"{file} restates the {identifier} {spelling} shape") + for file, allowed in budget.items(): + found = counted.get(file, 0) + if found != allowed: + rows.append( + f"{file} declares {allowed} individual shape site(s), found {found}" + ) + return sorted(set(rows)) + + +def unfoldable_offender_rows( + definitions: list[dict[str, object]], + *, + declared: dict[str, int] | None = None, +) -> list[str]: + budget = DECLARED_UNFOLDABLE_SITES if declared is None else declared + counted: dict[str, int] = {} + rows: list[str] = [] + for item in definitions: + file = str(item["file"]) + if file in budget: + counted[file] = counted.get(file, 0) + 1 + continue + rows.append( + f"{file}:{item['line']} builds a Lark identifier regex from a value " + "this scan cannot fold; declare it or route the site to the owner" + ) + for file, allowed in budget.items(): + found = counted.get(file, 0) + if found != allowed: + rows.append( + f"{file} declares {allowed} unfoldable site(s), found {found}" + ) + return sorted(set(rows)) + + +def module_level_bindings(tree: ast.Module) -> set[str]: + names: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Assign): + names.update( + target.id for target in node.targets if isinstance(target, ast.Name) + ) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + names.add(node.target.id) + return names + + +def references(tree: ast.Module, name: str) -> bool: + """True when ``name`` is read outside its own import clause. + + An import binds through ``ast.alias``, never ``ast.Name``, so a plain Name + walk already ignores it. + """ + + return any( + isinstance(node, ast.Name) and node.id == name for node in ast.walk(tree) + ) + + +def parse_module(module: ModuleType) -> ast.Module: + path = pathlib.Path(str(module.__file__)).resolve() + return ast.parse(path.read_text(encoding="utf-8")) + + +# -------------------------------------------------------------------------- +# 1. one owner per shape family, across the whole product tree +# -------------------------------------------------------------------------- + + +def test_no_module_restates_a_lark_identifier_shape() -> None: + assert offender_rows(collect_shape_definitions(REPO_ROOT / "loopx")) == [] + + +def test_no_lark_module_hides_an_unfoldable_identifier_construction() -> None: + # The unfoldable layer is bounded to the package that owns these shapes and + # to modules whose text mentions an identifier. Spread wider it reports + # dozens of regexes built from data (credential alternations, setup URLs, + # markdown headings) that answer a different question, and a declaration list + # that wide is noise nobody keeps current. + assert unfoldable_offender_rows(collect_unfoldable_definitions(LARK_PACKAGE)) == [] + + +def test_unfoldable_probes_are_forced_into_the_open() -> None: + probes = ( + 'import re\n\n\ndef check(prefix, chat_id):\n' + " return re.compile(prefix + chat_id)\n", + 'import re\n\nCHAT_ID = "^oc_[A-Za-z0-9_-]+$"\n\n\ndef pick(CHAT_ID):\n' + " return re.compile(CHAT_ID)\n", + ) + for source in probes: + rows = unfoldable_rows_in_source(source, "loopx/extensions/lark/probe.py") + assert rows, source + assert shape_rows_in_source(source, "loopx/extensions/lark/probe.py") == [] + assert unfoldable_offender_rows(rows, declared={}) != [] + + +def test_a_foldable_probe_is_not_also_reported_as_unfoldable() -> None: + source = 'import re\n\nHEAD = "^oc_"\nTAIL = "[A-Za-z0-9_-]+$"\n' + source += "CHAT = re.compile(HEAD + TAIL)\n" + name = "loopx/extensions/lark/probe.py" + assert shape_rows_in_source(name and source, name) + assert unfoldable_rows_in_source(source, name) == [] + + +def test_prescreen_still_leaves_the_scans_two_call_forms_reachable() -> None: + # The prescreen is load-bearing: a token list that misses a call form makes + # the scan blind to it. Both accepted forms are probed here. + for source in ( + 'import re\n\nX = re.compile(r"^oc_[A-Za-z0-9_-]+$")\n', + 'from re import fullmatch\n\nX = fullmatch(r"^om_[A-Za-z0-9_-]+$", "om_x")\n', + ): + assert shape_rows_in_source(source, "loopx/extensions/lark/probe.py") + + def _bodies() -> list[str]: + return list(WHOLE_VALUE_BODIES.values()) + + assert _bodies()[0] in WHOLE_VALUE_BODIES.values() + + +def test_owner_states_each_shape_as_the_recorded_whole_value_body() -> None: + # Stated against literals on purpose: if the owner renames or re-forms a + # shape, this fails instead of the scan quietly following the new spelling. + assert identity_shapes.LARK_EVENT_ID_PATTERN.pattern == ( + "^" + WHOLE_VALUE_BODIES["event_id"] + "$" + ) + assert identity_shapes.LARK_MESSAGE_ID_PATTERN.pattern == ( + "^" + WHOLE_VALUE_BODIES["message_id"] + "$" + ) + assert identity_shapes.LARK_CHAT_ID_PATTERN.pattern == ( + "^" + WHOLE_VALUE_BODIES["chat_id"] + "$" + ) + assert identity_shapes.LARK_OPEN_ID_PATTERN.pattern == ( + "^" + WHOLE_VALUE_BODIES["operator_id"] + "$" + ) + + +def test_the_owner_is_the_only_module_that_defines_any_of_them() -> None: + rows = collect_shape_definitions(REPO_ROOT / "loopx") + assert {str(row["file"]) for row in rows} == {OWNER_MODULE, *DECLARED_INDIVIDUAL_SITES} + + +def test_owner_states_both_spellings_and_nothing_else() -> None: + rows = [item for item in collect_shape_definitions(LARK_PACKAGE) if item["file"] == OWNER_MODULE] + anchored = {str(row["identifier"]) for row in rows if row["anchored"]} + searched = {str(row["identifier"]) for row in rows if not row["anchored"]} + assert anchored == set(ANCHORED_EXPORTS) + assert searched == SEARCH_IDENTIFIERS + assert len(rows) == len(ANCHORED_EXPORTS) + len(SEARCH_EXPORTS) + for export in ANCHORED_EXPORTS.values(): + pattern = getattr(identity_shapes, export) + body = pattern.pattern[1:-1] + assert pattern.pattern.startswith("^") and pattern.pattern.endswith("$") + assert body in WHOLE_VALUE_BODIES.values(), export + for export in SEARCH_EXPORTS.values(): + pattern = getattr(identity_shapes, export) + assert pattern.pattern in WHOLE_VALUE_BODIES.values(), export + + +@pytest.mark.parametrize("hub", [TRANSPORT_HUB, INBOX_HUB]) +def test_a_former_hub_defines_no_shape_of_its_own(hub: str) -> None: + rows = [item for item in collect_shape_definitions(LARK_PACKAGE) if item["file"] == hub] + assert rows == [], f"{hub} still decides an identifier shape locally" + + +# -------------------------------------------------------------------------- +# 2. consumers hold the owner's object and actually apply it +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("path", "entry"), sorted(SHAPE_CONSUMERS.items()), ids=sorted(SHAPE_CONSUMERS) +) +def test_shape_consumer_is_wired_to_the_owner(path: str, entry: tuple) -> None: + module, attribute = entry + tree = parse_module(module) + assert references(tree, attribute), f"{path} never applies {attribute}" + assert attribute not in module_level_bindings(tree), ( + f"{path} binds {attribute} locally instead of importing the owner" + ) + assert getattr(module, attribute) is getattr(identity_shapes, attribute) + + +def test_each_hub_reexports_the_spelling_its_callers_need() -> None: + # The transport's two search() call sites need the unanchored form; every + # inbox caller applies fullmatch, so the inbox hands on the whole-value one. + assert goal_channel_transport.CHAT_ID_PATTERN is identity_shapes.LARK_CHAT_ID_SEARCH + assert goal_channel_transport.MESSAGE_ID_PATTERN is ( + identity_shapes.LARK_MESSAGE_ID_SEARCH + ) + assert goal_channel_transport.OPEN_ID_PATTERN is identity_shapes.LARK_OPEN_ID_SEARCH + assert event_inbox.CHAT_ID_PATTERN is identity_shapes.LARK_CHAT_ID_PATTERN + assert event_inbox.MESSAGE_ID_PATTERN is identity_shapes.LARK_MESSAGE_ID_PATTERN + assert ( + goal_channel_transport.CHAT_ID_PATTERN + is not identity_shapes.LARK_CHAT_ID_PATTERN + ) + + +def test_hub_reexports_are_named_by_alias_import_not_local_compile() -> None: + for path, (module, attributes) in sorted(HUB_REEXPORTS.items()): + rows = shape_rows_in_source( + pathlib.Path(str(module.__file__)).resolve().read_text("utf-8"), path + ) + assert rows == [], path + + +def test_inbox_callers_still_receive_one_object_through_the_chain() -> None: + for name in ("manager_reply_delivery", "turn_start_sync", "inbox_reply"): + module = __import__(f"loopx.extensions.lark.{name}", fromlist=["MESSAGE_ID_PATTERN"]) + assert module.MESSAGE_ID_PATTERN is identity_shapes.LARK_MESSAGE_ID_PATTERN, name + + +def test_callback_validators_do_not_restate_the_identity_table() -> None: + for module in (goal_channel_operation, team_plan_confirmation): + tree = parse_module(module) + tables = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Tuple) + and any( + isinstance(element, ast.Constant) and element.value == "operator_id" + for element in node.elts + ) + ] + assert tables == [], f"{module.__name__} restates the callback table" + + +def test_neither_callback_validator_recompiles_a_shape() -> None: + for module in (goal_channel_operation, team_plan_confirmation): + source = pathlib.Path(str(module.__file__)).resolve().read_text("utf-8") + assert shape_rows_in_source(source, module.__name__) == [] + + +# -------------------------------------------------------------------------- +# 3. the owner's decision +# -------------------------------------------------------------------------- + +VALID_IDS = { + "event_id": "evt:1", + "message_id": "om_control_public_fixture", + "chat_id": "oc_abc-123", + "operator_id": "ou_ABC_9", +} + + +def _event(**overrides: object) -> dict[str, object]: + event: dict[str, object] = dict(VALID_IDS) + event.update(overrides) + return event + + +def test_callback_table_names_the_four_fields_in_validation_order() -> None: + assert identity_shapes.CARD_CALLBACK_IDENTITY_SHAPES == ( + ("event_id", identity_shapes.LARK_EVENT_ID_PATTERN), + ("message_id", identity_shapes.LARK_MESSAGE_ID_PATTERN), + ("chat_id", identity_shapes.LARK_CHAT_ID_PATTERN), + ("operator_id", identity_shapes.LARK_OPEN_ID_PATTERN), + ) + + +def test_require_identity_accepts_a_whole_identifier_per_field() -> None: + for field, value in VALID_IDS.items(): + identity_shapes.require_card_callback_identity( + _event(**{field: value}), error_prefix="operation callback" + ) + + +def test_longest_accepted_event_id_is_240_characters() -> None: + identity_shapes.require_card_callback_identity( + _event(event_id="a" * 240), error_prefix="operation callback" + ) + with pytest.raises(ValueError): + identity_shapes.require_card_callback_identity( + _event(event_id="a" * 241), error_prefix="operation callback" + ) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("event_id", ""), + ("event_id", "id with space"), + ("event_id", "id/slash"), + ("message_id", "om_"), + ("message_id", "prefix_om_real"), + ("message_id", "oc_not_a_message"), + ("message_id", "om_ dot"), + ("chat_id", "oc_"), + ("chat_id", "oc_not+plus"), + ("operator_id", "ou_"), + ("operator_id", "om_not_an_operator"), + ], +) +def test_require_identity_rejects_values_that_are_not_whole_ids( + field: str, value: str +) -> None: + with pytest.raises(ValueError) as caught: + identity_shapes.require_card_callback_identity( + _event(**{field: value}), error_prefix="team plan callback" + ) + assert caught.value.args[0] == f"team plan callback {field} is invalid" + + +def test_missing_identity_field_is_rejected() -> None: + event = dict(VALID_IDS) + del event["chat_id"] + with pytest.raises(ValueError) as caught: + identity_shapes.require_card_callback_identity( + event, error_prefix="operation callback" + ) + assert caught.value.args[0] == "operation callback chat_id is invalid" + + +def test_first_offending_field_is_reported_and_validation_is_ordered() -> None: + with pytest.raises(ValueError) as caught: + identity_shapes.require_card_callback_identity( + _event(message_id="bad", chat_id="bad", operator_id="bad"), + error_prefix="operation callback", + ) + assert caught.value.args[0] == "operation callback message_id is invalid" + + +@pytest.mark.parametrize( + "value", + ["om_ok1", "om_ok1\n", "\nom_ok1", "om_a\nb", "", "om_", "om_\u00e9", "om1\n\n"], +) +def test_anchored_and_unanchored_fullmatch_agree_on_every_tricky_value( + value: str, +) -> None: + """Why this convergence cannot change a product answer. + + Every site this slice replaced applies ``fullmatch``, where the anchors are + redundant: ``re.fullmatch`` already requires the whole string, so the owner's + anchored spelling and the transport's unanchored extraction spelling accept + and reject exactly the same values. Only the ``search()`` sites, which stay + with the transport, read a difference. + """ + + anchored = bool(identity_shapes.LARK_MESSAGE_ID_PATTERN.fullmatch(value)) + unanchored = bool(identity_shapes.LARK_MESSAGE_ID_SEARCH.fullmatch(value)) + assert anchored == unanchored + accepted = True + try: + identity_shapes.require_card_callback_identity( + _event(message_id=value), error_prefix="operation callback" + ) + except ValueError: + accepted = False + assert accepted == anchored + + +# -------------------------------------------------------------------------- +# 4. the scan sees the spellings a future edit will actually use +# -------------------------------------------------------------------------- + +OFFENDING_SNIPPETS: tuple[tuple[str, str], ...] = ( + ( + "module level anchored compile", + 'import re\n\nCHAT = re.compile(r"^oc_[A-Za-z0-9_-]+$")\n', + ), + ( + "module level unanchored compile", + 'import re\n\nCHAT = re.compile(r"oc_[A-Za-z0-9_-]+")\n', + ), + ( + "literal concatenated from a same-file constant", + 'import re\n\nHEAD = "^oc_"\nTAIL = "[A-Za-z0-9_-]+$"\n' + "CHAT = re.compile(HEAD + TAIL)\n", + ), + ( + "inline fullmatch against the event id body", + 'import re\n\n\ndef check(value):\n' + ' return re.fullmatch(r"[A-Za-z0-9._:-]{1,240}", value)\n', + ), + ( + "fullmatch imported straight from re", + 'from re import fullmatch\n\n' + 'CHECKED = fullmatch(r"^om_[A-Za-z0-9_-]+$", "om_x")\n', + ), + ( + "non-raw literal with the same body", + 'import re\n\nCHAT = re.compile("^oc_[A-Za-z0-9_-]+$")\n', + ), + ( + "concatenated through a module-level constant chain", + 'import re\n\n_PREFIX = "^ou_"\n_BODY = "[A-Za-z0-9_-]+$"\n' + "OPERATOR = re.compile(_PREFIX + _BODY)\n", + ), + ( + "operator id restated under an unrelated name", + 'import re\n\nOPERATOR = re.compile(r"^ou_[A-Za-z0-9_-]+$")\n', + ), + ( + "event id inside a function body, not at module level", + 'import re\n\n\ndef pick(value):\n' + ' pattern = re.compile(r"^ou_[A-Za-z0-9_-]+$")\n' + " return pattern.match(value)\n", + ), +) +NON_OFFENDING_SNIPPETS: tuple[tuple[str, str], ...] = ( + ( + "a goal id shape, which is a different decision", + 'import re\n\nGOAL = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$")\n', + ), + ( + "a chat id embedded in a larger route grammar", + 'import re\n\nROUTE = re.compile(r"^route:oc_[A-Za-z0-9_-]+:v1$")\n', + ), + ( + "an app id, which this slice does not own", + 'import re\n\nAPP = re.compile(r"^cli_[A-Za-z0-9_-]+$")\n', + ), + ( + "a pattern built from runtime data, which cannot be folded", + 'import re\n\nCHAT = re.compile(prefix + "[A-Za-z0-9_-]+")\n', + ), + ( + "a module constant rebound locally before use, which is not trusted", + 'import re\n\nCHAT_ID = "^oc_[A-Za-z0-9_-]+$"\n\n\ndef pick(CHAT_ID):\n' + " return re.compile(CHAT_ID)\n", + ), + ( + "an id-shaped data value matched by another module\'s pattern", + 'import re\n\nMATCHED = re.compile(r"^x+$").fullmatch("oc_abc")\n', + ), +) + + +@pytest.mark.parametrize( + ("label", "source"), OFFENDING_SNIPPETS, ids=[row[0] for row in OFFENDING_SNIPPETS] +) +def test_the_scan_reports_a_restated_shape(label: str, source: str) -> None: + rows = shape_rows_in_source(source, "loopx/extensions/lark/probe.py") + assert rows, label + assert offender_rows(rows, declared={}) != [], label + + +@pytest.mark.parametrize( + ("label", "source"), + NON_OFFENDING_SNIPPETS, + ids=[row[0] for row in NON_OFFENDING_SNIPPETS], +) +def test_the_scan_leaves_a_different_decision_alone( + label: str, source: str +) -> None: + assert shape_rows_in_source(source, "loopx/extensions/lark/probe.py") == [] + + +def test_declared_site_count_is_enforced_in_both_directions() -> None: + rows = collect_shape_definitions(LARK_PACKAGE) + declared_file = "loopx/extensions/lark/event_collector_runtime.py" + assert any(item["file"] == declared_file for item in rows) + assert offender_rows(rows) == [] + retired = [item for item in rows if item["file"] != declared_file] + assert offender_rows(retired) == [ + f"{declared_file} declares 1 individual shape site(s), found 0" + ] + + +def test_a_converted_declared_site_is_not_silently_reintroduced() -> None: + declared_file = "loopx/extensions/lark/event_collector_runtime.py" + converted = [ + { + "file": declared_file, + "line": 145, + "identifier": "event_id", + "anchored": False, + }, + { + "file": "loopx/extensions/lark/other_module.py", + "line": 9, + "identifier": "event_id", + "anchored": True, + }, + ] + assert offender_rows(converted) == [ + "loopx/extensions/lark/other_module.py restates the event_id whole-value shape" + ]