diff --git a/loopx/capabilities/decision_context/assembler.py b/loopx/capabilities/decision_context/assembler.py index e778f439d1..a398b20c6b 100644 --- a/loopx/capabilities/decision_context/assembler.py +++ b/loopx/capabilities/decision_context/assembler.py @@ -17,7 +17,11 @@ canonical_context_matches, opaque_provider_ref, ) -from .freshness import build_source_freshness_report, source_freshness_row +from .freshness import ( + _SUCCESS_STATUSES, + build_source_freshness_report, + source_freshness_row, +) from .packets import build_decision_evidence_packet from .sources import ( DecisionSourceExactRead, @@ -177,7 +181,7 @@ def _source_coverage( complete_count = sum( 1 for collected in sources - if collected.scan.status in {"completed", "no_change"} + if collected.scan.status in _SUCCESS_STATUSES and not collected.exact_read_failed ) priority_rows.append( @@ -193,7 +197,7 @@ def _source_coverage( incomplete_required_source_ids.extend( collected.source.source_id for collected in sources - if collected.scan.status not in {"completed", "no_change"} + if collected.scan.status not in _SUCCESS_STATUSES or collected.exact_read_failed ) @@ -485,7 +489,7 @@ def _assembly_source_freshness( observed_at=assembly_time, last_read_at=( assembly_time.isoformat() - if attempt in {"completed", "no_change"} + if attempt in _SUCCESS_STATUSES else None ), last_attempt_status=attempt, @@ -720,7 +724,7 @@ def assemble_decision_evidence( "provider": source.provider_id, "status": ( "healthy" - if scan.status in {"completed", "no_change"} + if scan.status in _SUCCESS_STATUSES and not exact_read_failed else "degraded" if scan.status == "completed" diff --git a/loopx/capabilities/decision_context/capture.py b/loopx/capabilities/decision_context/capture.py index 0afa785b53..fb947724c8 100644 --- a/loopx/capabilities/decision_context/capture.py +++ b/loopx/capabilities/decision_context/capture.py @@ -24,6 +24,7 @@ assemble_decision_evidence, ) from .freshness import ( + _SUCCESS_STATUSES, build_source_freshness_report, source_freshness_row, write_capture_host_health, @@ -33,7 +34,6 @@ from .runtime import _build_source_providers from .sources import DecisionSourceProvider, DecisionSourceSpec -_READ_SUCCESS_STATUSES = frozenset({"completed", "no_change"}) _READ_FAILURE_STATUSES = frozenset({"provider_failed", "failed", "unavailable"}) @@ -150,7 +150,7 @@ def _status( else: last_read_at = ( source["checked_at"] - if source["status"] in _READ_SUCCESS_STATUSES + if source["status"] in _SUCCESS_STATUSES else None ) failure_streak = ( @@ -478,7 +478,7 @@ def _execute_capture_tick( else: status = scan.status if ( - status in {"completed", "no_change"} + status in _SUCCESS_STATUSES and scan.cursor_after and scan.cursor_after != cursor ): @@ -497,7 +497,7 @@ def _execute_capture_tick( # advance last_success_at, so repeated failures cannot look fresh. last_success_at = row["last_success_at"] if row else None failure_streak = int(row["failure_streak"] or 0) if row else 0 - if status in _READ_SUCCESS_STATUSES: + if status in _SUCCESS_STATUSES: last_success_at, failure_streak = observed_at, 0 elif status in _READ_FAILURE_STATUSES: failure_streak += 1 diff --git a/tests/architecture/test_decision_context_success_status_owner.py b/tests/architecture/test_decision_context_success_status_owner.py new file mode 100644 index 0000000000..3c863ab1ea --- /dev/null +++ b/tests/architecture/test_decision_context_success_status_owner.py @@ -0,0 +1,230 @@ +"""Guard one owner for the decision-context successful-read status set (#4447). + +``_SUCCESS_STATUSES`` in ``freshness.py`` decides which scan/attempt statuses +count as a successful read (``completed`` / ``no_change``). The same two-member +set was restated across the package: ``capture.py`` bound it under a different +name (``_READ_SUCCESS_STATUSES``) *and* inlined it once, while ``assembler.py`` +inlined it four times with no name at all -- seven sites folding to one value. +Both consumers *already import* from ``freshness``, so borrowing the owner object +adds no new import edge and cannot introduce a cycle. Every site now references +the single owner and the restated name is retired. + +The by-name semantic inventory cannot see this: the owner and the retired +capture name are private ``_``-bindings (the inventory only catalogues +``UPPER_NAME`` carriers), they differ by name, and four of the seven sites had no +name to key on. This guard keys on the **folded value** instead, so a renamed or +anonymous inline restatement is caught, while the genuinely different neighbours +-- the four-member ``_SCAN_STATUSES`` vocabulary and the ordered four-status +report tuple -- are never swept. + +Two layers: a whole-tree folded-value census (only the owner states the set), and +behavioural parity through the owner's own ``source_freshness_row`` gate, so a +migration that silently changed which statuses count as success fails here. +""" + +from __future__ import annotations + +import ast +import pathlib +from datetime import datetime, timezone + +import pytest + +from loopx.capabilities.decision_context import ( + assembler as assembler_mod, + capture as capture_mod, + freshness as owner, + sources as sources_mod, +) + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +PACKAGE_ROOT = REPO_ROOT / "loopx" +OWNER_PATH = "loopx/capabilities/decision_context/freshness.py" +CAPTURE_PATH = "loopx/capabilities/decision_context/capture.py" +ASSEMBLER_PATH = "loopx/capabilities/decision_context/assembler.py" + +SUCCESS = frozenset({"completed", "no_change"}) +FAILURE = frozenset({"provider_failed", "failed", "unavailable"}) +SCAN_VOCABULARY = frozenset({"completed", "failed", "no_change", "unavailable"}) + + +def _members(node: ast.expr) -> frozenset[str] | None: + """Fold a set / frozenset / list / tuple of plain string literals to members.""" + + if isinstance(node, ast.Call): + fname = getattr(node.func, "id", None) or getattr(node.func, "attr", None) + if fname in {"frozenset", "set"} and len(node.args) == 1: + return _members(node.args[0]) + return None + if isinstance(node, (ast.Set, ast.List, ast.Tuple)): + values = [ + element.value + for element in node.elts + if isinstance(element, ast.Constant) and isinstance(element.value, str) + ] + if len(values) == len(node.elts) and values: + return frozenset(values) + return None + + +def _status_literal_sites(source: str) -> list[int]: + """Every collection literal in the source folding to SUCCESS, counted once. + + ``frozenset({...})`` is walked as both the call and its inner set; the inner + node is skipped so one restatement is one site, not two. + """ + + tree = ast.parse(source) + wrapped = { + id(node.args[0]) + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and (getattr(node.func, "id", None) or getattr(node.func, "attr", None)) + in {"frozenset", "set"} + and len(node.args) == 1 + } + return [ + node.lineno + for node in ast.walk(tree) + if id(node) not in wrapped and _members(node) == SUCCESS + ] + + +@pytest.fixture(scope="module") +def tree_offenders() -> dict[str, list[int]]: + offenders: dict[str, list[int]] = {} + for path in sorted(PACKAGE_ROOT.rglob("*.py")): + relative = path.relative_to(REPO_ROOT).as_posix() + if relative == OWNER_PATH: + continue + sites = _status_literal_sites(path.read_text(encoding="utf-8")) + if sites: + offenders[relative] = sites + return offenders + + +def test_only_the_owner_states_the_success_set(tree_offenders) -> None: + assert tree_offenders == {}, ( + f"a restatement of the successful-read statuses remains: {tree_offenders}" + ) + + +def test_the_owner_states_the_recorded_members_once() -> None: + assert frozenset(owner._SUCCESS_STATUSES) == SUCCESS + source = (REPO_ROOT / OWNER_PATH).read_text(encoding="utf-8") + assert len(_status_literal_sites(source)) == 1 + + +def test_capture_borrows_the_owner_object() -> None: + assert capture_mod._SUCCESS_STATUSES is owner._SUCCESS_STATUSES + + +def test_assembler_borrows_the_owner_object() -> None: + assert assembler_mod._SUCCESS_STATUSES is owner._SUCCESS_STATUSES + + +def test_capture_retired_its_restated_name() -> None: + assert not hasattr(capture_mod, "_READ_SUCCESS_STATUSES") + source = (REPO_ROOT / CAPTURE_PATH).read_text(encoding="utf-8") + assert _status_literal_sites(source) == [] + assert "_SUCCESS_STATUSES" in source + + +def test_assembler_has_no_inline_restatement() -> None: + source = (REPO_ROOT / ASSEMBLER_PATH).read_text(encoding="utf-8") + assert _status_literal_sites(source) == [] + assert "_SUCCESS_STATUSES" in source + + +def test_the_failure_neighbour_is_a_different_set() -> None: + # capture keeps a distinct failure set; single-sourcing success must not + # collapse or delete it. + assert frozenset(capture_mod._READ_FAILURE_STATUSES) == FAILURE + assert SUCCESS.isdisjoint(FAILURE) + + +def test_the_scan_vocabulary_is_a_different_set() -> None: + # The full status vocabulary is a superset, not the success set; it stays + # its own owner in sources.py and is deliberately left untouched. + assert frozenset(sources_mod._SCAN_STATUSES) == SCAN_VOCABULARY + assert SUCCESS < SCAN_VOCABULARY + assert SUCCESS != SCAN_VOCABULARY + + +# --- behavioural parity through the owner's own freshness gate -------------------------- + +OBSERVED_AT = datetime(2026, 7, 25, 17, 30, tzinfo=timezone.utc) + + +def _spec() -> sources_mod.DecisionSourceSpec: + return sources_mod.DecisionSourceSpec( + source_id="source:owner", + provider_id="source-provider", + source_kind="person", + priority="p0", + evidence_level="s2", + objectives=("adoption",), + scan_mode="incremental", + exact_read_policy="material_change", + freshness_seconds=3600, + private_locator="private-owner-locator", + ) + + +@pytest.mark.parametrize("status", sorted(SUCCESS)) +def test_a_successful_attempt_does_not_alert(status: str) -> None: + row = owner.source_freshness_row( + source=_spec(), + observed_at=OBSERVED_AT, + last_read_at=OBSERVED_AT.isoformat(), + last_attempt_status=status, + failure_streak=0, + ) + assert row["status"] == "fresh" + assert row["alert"] is False + assert row["alert_reasons"] == [] + + +@pytest.mark.parametrize("status", sorted(capture_mod._READ_FAILURE_STATUSES)) +def test_a_failed_attempt_alerts_with_the_status(status: str) -> None: + # The read time is fresh, yet a non-success attempt still raises an alert: + # this is exactly the owner gate the consumers now share by identity. + row = owner.source_freshness_row( + source=_spec(), + observed_at=OBSERVED_AT, + last_read_at=OBSERVED_AT.isoformat(), + last_attempt_status=status, + failure_streak=0, + ) + assert row["status"] == "fresh" + assert row["alert"] is True + assert row["alert_reasons"] == [f"last_attempt_{status}"] + + +# --- the census bites every restatement form and leaves the others ---------------------- + +POISON_CORPUS = ( + ("bare set restatement", 'X = {"completed", "no_change"}\n', 1), + ("frozenset call", 'X = frozenset({"completed", "no_change"})\n', 1), + ("set call over a list", 'X = set(["completed", "no_change"])\n', 1), + ("list literal", 'X = ["completed", "no_change"]\n', 1), + ("tuple literal", 'X = ("completed", "no_change")\n', 1), + ("reordered members fold the same", 'X = {"no_change", "completed"}\n', 1), + ("renamed binding is still caught", '_ANY = {"completed", "no_change"}\n', 1), + ("nested inside a function", 'def f():\n return {"completed", "no_change"}\n', 1), + ("importing the owner is not a restatement", "from .m import _SUCCESS_STATUSES\n", 0), + ("alias rebind is not a restatement", "_READ_SUCCESS_STATUSES = _SUCCESS_STATUSES\n", 0), + ("the full scan vocabulary is not this set", '_SCAN = {"completed", "failed", "no_change", "unavailable"}\n', 0), + ("the ordered report tuple is not this set", 'X = ("completed", "no_change", "failed", "unavailable")\n', 0), + ("a superset is not this set", 'X = {"completed", "no_change", "failed"}\n', 0), + ("a single status is not this set", 'X = {"completed"}\n', 0), +) + + +@pytest.mark.parametrize( + "label,source,expected", POISON_CORPUS, ids=[case[0] for case in POISON_CORPUS] +) +def test_the_census_names_every_form_and_leaves_the_others( + label: str, source: str, expected: int +) -> None: + assert len(_status_literal_sites(source)) == expected, label \ No newline at end of file