From 31582eb9015c014420928b0672d2e59640f036c9 Mon Sep 17 00:00:00 2001 From: Twiink <154496350+Twiink@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:07:25 +0800 Subject: [PATCH 1/2] refactor(issue-fix): import the reviewer-notification time shape from its owner Two issue-fix surfaces validated the same reviewer-notification ``allowed_local_time`` window with their own compile of the HH:MM shape the domain pack already owns, including the same start == end and outside_window rules. A later edit to one copy could accept a window the other rejects, and nothing tied the three call sites together. Observable result: the window verdict now comes from one pattern object for the domain pack, the notification sinks/before-send surface and the reward-memory delivery policy. The new guard judges folded values rather than spellings, pins every delegating module to the owner's object, and reconciles the modules that actually import the owner against the declared consumer list, so deleting an entry cannot silently shrink coverage. Validation: guard 22 passed; reviewer/reward-memory selection 343 passed; focused run 84 passed; ruff check clean; configured mypy scope 17 files clean. Mutations: local copy back in reward_memory -> 2 red; inline literal in the notification call site -> 2 red; dropping one consumer entry -> 1 red (reconciliation test). Refs #4447 Signed-off-by: Twiink <154496350+Twiink@users.noreply.github.com> --- .../issue_fix/reviewer_notification.py | 14 +- loopx/capabilities/issue_fix/reward_memory.py | 8 +- .../test_issue_fix_local_time_owner.py | 317 ++++++++++++++++++ 3 files changed, 329 insertions(+), 10 deletions(-) create mode 100644 tests/architecture/test_issue_fix_local_time_owner.py diff --git a/loopx/capabilities/issue_fix/reviewer_notification.py b/loopx/capabilities/issue_fix/reviewer_notification.py index f3d620b71e..56cb27d643 100644 --- a/loopx/capabilities/issue_fix/reviewer_notification.py +++ b/loopx/capabilities/issue_fix/reviewer_notification.py @@ -16,6 +16,7 @@ reviewer_notification_before_send_gate, ) from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN +from ...domain_packs.issue_fix import REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN ISSUE_FIX_REVIEWER_NOTIFICATION_SINKS_INPUT_SCHEMA_VERSION = ( @@ -34,7 +35,6 @@ "issue_fix_reviewer_notification_queue_receipt_v0" ) SAFE_LOCAL_KEY_PATTERN = re.compile(r"[A-Za-z0-9._-]{1,100}") -LOCAL_TIME_PATTERN = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d") CommandRunner = Callable[[Sequence[str]], Mapping[str, Any]] NotificationSinkAdapter = Callable[..., dict[str, Any]] @@ -334,8 +334,8 @@ def _delivery_window_decision( start_text = str(allowed_local_time.get("start") or "").strip() end_text = str(allowed_local_time.get("end") or "").strip() if ( - not LOCAL_TIME_PATTERN.fullmatch(start_text) - or not LOCAL_TIME_PATTERN.fullmatch(end_text) + not REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch(start_text) + or not REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch(end_text) or start_text == end_text or policy.get("outside_window") != "queue_without_send" ): @@ -545,8 +545,12 @@ def validate_issue_fix_reviewer_notification_sinks_result( or receipt.get("summary_policy_status") not in {"reward_memory_verified", "sink_config"} or not isinstance(window, Mapping) - or not LOCAL_TIME_PATTERN.fullmatch(str(window.get("start") or "")) - or not LOCAL_TIME_PATTERN.fullmatch(str(window.get("end") or "")) + or not REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch( + str(window.get("start") or "") + ) + or not REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch( + str(window.get("end") or "") + ) ): errors.append("queued receipt is invalid") continue diff --git a/loopx/capabilities/issue_fix/reward_memory.py b/loopx/capabilities/issue_fix/reward_memory.py index 5c594351bd..b7891025fd 100644 --- a/loopx/capabilities/issue_fix/reward_memory.py +++ b/loopx/capabilities/issue_fix/reward_memory.py @@ -2,13 +2,13 @@ import hashlib import json -import re from collections.abc import Mapping from datetime import time from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from ...control_plane.runtime.public_safety import public_safe_compact_text +from ...domain_packs.issue_fix import REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN from ..context_providers.base import ContextProvider from ..reward_memory.application import RewardMemoryApplier from ..reward_memory.candidate_review import build_issue_fix_reward_memory_candidate @@ -48,8 +48,6 @@ "issue_fix_reward_memory_execution_evidence_v0" ) -_LOCAL_TIME_PATTERN = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d") - _EVENT_FIELDS = { "schema_version", "issue_ref", @@ -613,8 +611,8 @@ def _reviewer_notification_delivery_policy(value: object) -> dict[str, Any]: end = str(allowed_local_time.get("end") or "").strip() if ( not timezone_name - or not _LOCAL_TIME_PATTERN.fullmatch(start) - or not _LOCAL_TIME_PATTERN.fullmatch(end) + or not REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch(start) + or not REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch(end) or start == end or value.get("outside_window") != "queue_without_send" ): diff --git a/tests/architecture/test_issue_fix_local_time_owner.py b/tests/architecture/test_issue_fix_local_time_owner.py new file mode 100644 index 0000000000..d3c4076780 --- /dev/null +++ b/tests/architecture/test_issue_fix_local_time_owner.py @@ -0,0 +1,317 @@ +"""One owner decides what a reviewer-notification local time looks like. + +The shape ``(?:[01]\\d|2[0-3]):[0-5]\\d`` answers one question -- is this an +``HH:MM`` value inside a reviewer-notification delivery window -- and three +modules stated it: the issue-fix domain pack plus two capability surfaces that +validate the same ``allowed_local_time`` window. The two surfaces now import the +domain pack's object instead of compiling the shape themselves. + +Layer 1 judges folded values rather than spellings, so a copy built with ``+``, +imported from ``re``, or written inline at a matcher is reported too. Layer 2 +pins the delegating modules to the owner's object and checks that they apply it +without keeping a local binding. Layer 3 records the owner's verdict at the +boundaries. A construction whose value this scan cannot fold has to be declared; +the declaration budget is empty today. +""" + +from __future__ import annotations + +import ast +import pathlib +from types import ModuleType + +import pytest + +from loopx.capabilities.issue_fix import reward_memory, reviewer_notification +from loopx.domain_packs import issue_fix as issue_fix_pack + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +OWNER_PATH = "loopx/domain_packs/issue_fix.py" +OWNER_NAME = "REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN" +LOCAL_TIME_BODY = r"(?:[01]\d|2[0-3]):[0-5]\d" +LOCAL_NAMES = (OWNER_NAME, "LOCAL_TIME_PATTERN", "_LOCAL_TIME_PATTERN") + +DECLARED_INDIVIDUAL_SITES: dict[str, int] = {} +DECLARED_UNFOLDABLE_SITES: dict[str, int] = {} + +CONSUMERS: dict[str, ModuleType] = { + "loopx/capabilities/issue_fix/reviewer_notification.py": reviewer_notification, + "loopx/capabilities/issue_fix/reward_memory.py": reward_memory, +} +MATCHERS = {"compile", "fullmatch", "match", "search"} + + +def fold_string(node: ast.AST, constants: dict[str, str], depth: int = 0) -> str | None: + if isinstance(node, ast.Constant): + return node.value if isinstance(node.value, str) else None + if isinstance(node, ast.JoinedStr): + parts = [ + value.value + for value in node.values + if isinstance(value, ast.Constant) and isinstance(value.value, str) + ] + return "".join(parts) if len(parts) == len(node.values) else None + 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 < 4: + return constants.get(node.id) + return None + + +def module_constants(tree: ast.Module) -> dict[str, str]: + 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): + pending.append((target.id, node.value)) + for _ in range(4): + 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 matcher_calls(tree: ast.Module) -> list[ast.Call]: + calls: list[ast.Call] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not node.args: + continue + func = node.func + if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name): + if func.value.id == "re" and func.attr in MATCHERS: + calls.append(node) + elif isinstance(func, ast.Name) and func.id in MATCHERS: + calls.append(node) + return calls + + +def time_rows(source: str, name: str) -> list[dict[str, object]]: + if ":" not in source: + return [] + tree = ast.parse(source) + constants = module_constants(tree) + rows: list[dict[str, object]] = [] + for node in matcher_calls(tree): + if fold_string(node.args[0], constants) == LOCAL_TIME_BODY: + rows.append({"file": name, "line": node.lineno, "kind": "local_time"}) + return rows + + +def unfoldable_rows(source: str, name: str) -> list[dict[str, object]]: + if ":" not in source or "[0-3]" not in source: + return [] + tree = ast.parse(source) + constants = module_constants(tree) + return [ + {"file": name, "line": node.lineno, "kind": "unfoldable"} + for node in matcher_calls(tree) + if fold_string(node.args[0], constants) is None + and "[0-3]" in ast.unparse(node.args[0]) + ] + + +def collect(root: pathlib.Path) -> list[dict[str, object]]: + rows: list[dict[str, object]] = [] + for path in sorted((root / "loopx").rglob("*.py")): + if "__pycache__" in path.parts: + continue + try: + source = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + rows.extend(time_rows(source, path.relative_to(root).as_posix())) + return rows + + +def offenders( + rows: list[dict[str, object]], + *, + declared: dict[str, int] | None = None, + owner: str = OWNER_PATH, +) -> list[str]: + budget = DECLARED_INDIVIDUAL_SITES if declared is None else declared + found: list[str] = [] + counted: dict[str, int] = {} + for row in rows: + file = str(row["file"]) + if file in budget: + counted[file] = counted.get(file, 0) + 1 + continue + if file == owner: + continue + found.append(f"{file}:{row['line']} restates the local-time shape") + for file, allowed in budget.items(): + if counted.get(file, 0) != allowed: + found.append( + f"{file} declares {allowed} site(s), found {counted.get(file, 0)}" + ) + return sorted(set(found)) + + +def module_bindings(tree: ast.Module) -> set[str]: + names: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Assign): + names.update(t.id for t in node.targets if isinstance(t, ast.Name)) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + names.add(node.target.id) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + names.add(node.name) + return names + + +def referenced(tree: ast.Module, name: str) -> bool: + return any( + isinstance(node, ast.Name) and node.id == name for node in ast.walk(tree) + ) + + +def parse(module: ModuleType) -> ast.Module: + path = pathlib.Path(str(module.__file__)).resolve() + return ast.parse(path.read_text(encoding="utf-8")) + + +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 == OWNER_NAME for alias in node.names + ): + importing.add(path.relative_to(REPO_ROOT).as_posix()) + return importing + + +def test_no_other_module_states_the_local_time_shape() -> None: + assert offenders(collect(REPO_ROOT)) == [] + + +def test_every_stated_site_is_the_owner_or_a_declared_consumer() -> None: + rows = collect(REPO_ROOT) + stray = sorted({str(row["file"]) for row in rows} - {OWNER_PATH, *CONSUMERS}) + 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) + + +@pytest.mark.parametrize("path", sorted(CONSUMERS), ids=sorted(CONSUMERS)) +def test_consumer_holds_the_owner_object_and_applies_it(path: str) -> None: + module = CONSUMERS[path] + assert ( + getattr(module, OWNER_NAME) + is issue_fix_pack.REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN + ) + tree = parse(module) + assert referenced(tree, OWNER_NAME), ( + f"{path} imports the shape but never applies it" + ) + local = module_bindings(tree) & set(LOCAL_NAMES) - {OWNER_NAME} + assert local == set(), f"{path} still binds {sorted(local)}" + assert time_rows(pathlib.Path(str(module.__file__)).read_text("utf-8"), path) == [] + + +def test_the_owner_states_the_shape_as_the_recorded_literal() -> None: + assert ( + issue_fix_pack.REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.pattern + == LOCAL_TIME_BODY + ) + + +@pytest.mark.parametrize("value", ["00:00", "09:30", "23:59"]) +def test_owner_accepts_a_window_time(value: str) -> None: + assert issue_fix_pack.REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch(value) + + +@pytest.mark.parametrize( + ("value", "why"), + [ + ("24:00", "hour out of range"), + ("09:60", "minute out of range"), + ("9:30", "hour needs two digits"), + ("09:30:00", "seconds are a different contract"), + ("09:30 ", "surrounding space"), + ("", "empty"), + ], +) +def test_owner_rejects_a_non_window_time(value: str, why: str) -> None: + assert ( + issue_fix_pack.REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN.fullmatch(value) is None + ), why + + +OFFENDING = ( + ( + "module level compile", + 'import re\n\nWINDOW = re.compile(r"(?:[01]\\d|2[0-3]):[0-5]\\d")\n', + ), + ( + "inline at the matcher", + 'import re\n\n\ndef check(value):\n return re.fullmatch(r"(?:[01]\\d|2[0-3]):[0-5]\\d", value)\n', + ), + ( + "concatenated from same-file constants", + 'import re\n\nHOUR = "(?:[01]\\\\d|2[0-3])"\nMINUTE = ":[0-5]\\\\d"\n' + "WINDOW = re.compile(HOUR + MINUTE)\n", + ), + ( + "imported matcher", + 'from re import compile\n\nWINDOW = compile(r"(?:[01]\\d|2[0-3]):[0-5]\\d")\n', + ), +) +NON_OFFENDING = ( + ( + "a seconds-aware time is a different decision", + 'import re\n\nCLOCK = re.compile(r"(?:[01]\\d|2[0-3]):[0-5]\\d:[0-5]\\d")\n', + ), + ( + "a date shape is unrelated", + 'import re\n\nDATE = re.compile(r"\\d{4}-\\d{2}-\\d{2}")\n', + ), +) + + +@pytest.mark.parametrize( + ("label", "source"), OFFENDING, ids=[row[0] for row in OFFENDING] +) +def test_the_scan_reports_a_restated_local_time(label: str, source: str) -> None: + rows = time_rows(source, "loopx/capabilities/probe.py") + assert rows, label + assert offenders(rows, declared={}) != [], label + + +@pytest.mark.parametrize( + ("label", "source"), NON_OFFENDING, ids=[row[0] for row in NON_OFFENDING] +) +def test_the_scan_leaves_a_different_decision_alone(label: str, source: str) -> None: + assert time_rows(source, "loopx/capabilities/probe.py") == [], label + + +def test_unfoldable_probe_is_forced_into_the_open() -> None: + source = "import re\n\n\ndef build(prefix):\n" + source += ' return re.compile(prefix + "(?:[01]\\\\d|2[0-3]):[0-5]\\\\d")\n' + name = "loopx/capabilities/probe.py" + assert time_rows(source, name) == [] + rows = unfoldable_rows(source, name) + assert rows, "the unfoldable probe must be seen" + assert DECLARED_UNFOLDABLE_SITES == {} From 17830d081bc78edb0d5d92838dbdc4b48f6a82de Mon Sep 17 00:00:00 2001 From: Twiink <154496350+Twiink@users.noreply.github.com> Date: Tue, 20 Oct 2026 17:15:21 +0800 Subject: [PATCH 2/2] test(architecture): route the unfoldable local-time probe through the repo scan Review P3: `collect()` only collected folded-value rows, so `unfoldable_rows()` was reached only by a synthetic probe and the empty declaration budget never participated in a repository-level decision. A real `build(prefix)` construction added under `loopx/` would therefore not have been reported. Add `collect_unfoldable()` plus `unfoldable_offender_rows()` and assert the real tree has no offending site; the probe now writes itself into a temporary `loopx/` tree and is rejected through the same collector the repository uses. Validation: guard 23 passed; ruff check clean. Mutation: making the collector return nothing turns the repository-scan probe red. Signed-off-by: Twiink <154496350+Twiink@users.noreply.github.com> --- .../test_issue_fix_local_time_owner.py | 55 +++++++++++++++++-- 1 file changed, 50 insertions(+), 5 deletions(-) diff --git a/tests/architecture/test_issue_fix_local_time_owner.py b/tests/architecture/test_issue_fix_local_time_owner.py index d3c4076780..14d7a4398f 100644 --- a/tests/architecture/test_issue_fix_local_time_owner.py +++ b/tests/architecture/test_issue_fix_local_time_owner.py @@ -136,6 +136,19 @@ def collect(root: pathlib.Path) -> list[dict[str, object]]: return rows +def collect_unfoldable(root: pathlib.Path) -> list[dict[str, object]]: + rows: list[dict[str, object]] = [] + for path in sorted((root / "loopx").rglob("*.py")): + if "__pycache__" in path.parts: + continue + try: + source = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + rows.extend(unfoldable_rows(source, path.relative_to(root).as_posix())) + return rows + + def offenders( rows: list[dict[str, object]], *, @@ -161,6 +174,29 @@ def offenders( return sorted(set(found)) +def unfoldable_offender_rows( + rows: list[dict[str, object]], *, declared: dict[str, int] | None = None +) -> list[str]: + budget = DECLARED_UNFOLDABLE_SITES if declared is None else declared + found: list[str] = [] + counted: dict[str, int] = {} + for row in rows: + file = str(row["file"]) + if file in budget: + counted[file] = counted.get(file, 0) + 1 + continue + found.append( + f"{file}:{row['line']} builds a local-time regex from a value this " + "scan cannot fold; declare it or route the site to the owner" + ) + for file, allowed in budget.items(): + if counted.get(file, 0) != allowed: + found.append( + f"{file} declares {allowed} unfoldable site(s), found {counted.get(file, 0)}" + ) + return sorted(set(found)) + + def module_bindings(tree: ast.Module) -> set[str]: names: set[str] = set() for node in tree.body: @@ -203,6 +239,10 @@ def test_no_other_module_states_the_local_time_shape() -> None: assert offenders(collect(REPO_ROOT)) == [] +def test_no_module_hides_an_unfoldable_local_time_construction() -> None: + assert unfoldable_offender_rows(collect_unfoldable(REPO_ROOT)) == [] + + def test_every_stated_site_is_the_owner_or_a_declared_consumer() -> None: rows = collect(REPO_ROOT) stray = sorted({str(row["file"]) for row in rows} - {OWNER_PATH, *CONSUMERS}) @@ -307,11 +347,16 @@ def test_the_scan_leaves_a_different_decision_alone(label: str, source: str) -> assert time_rows(source, "loopx/capabilities/probe.py") == [], label -def test_unfoldable_probe_is_forced_into_the_open() -> None: +def test_unfoldable_probe_is_rejected_through_the_repository_scan( + tmp_path: pathlib.Path, +) -> None: source = "import re\n\n\ndef build(prefix):\n" source += ' return re.compile(prefix + "(?:[01]\\\\d|2[0-3]):[0-5]\\\\d")\n' - name = "loopx/capabilities/probe.py" - assert time_rows(source, name) == [] - rows = unfoldable_rows(source, name) - assert rows, "the unfoldable probe must be seen" + probe_root = tmp_path / "loopx" / "capabilities" + probe_root.mkdir(parents=True) + (probe_root / "probe.py").write_text(source, encoding="utf-8") + assert time_rows(source, "loopx/capabilities/probe.py") == [] + rows = collect_unfoldable(tmp_path) + assert [row["file"] for row in rows] == ["loopx/capabilities/probe.py"] + assert unfoldable_offender_rows(rows)[0].startswith("loopx/capabilities/probe.py:") assert DECLARED_UNFOLDABLE_SITES == {}