From ccacedf19fc21961c01bd565e625cddec57e821b Mon Sep 17 00:00:00 2001 From: song Date: Thu, 17 Sep 2026 12:10:07 +0800 Subject: [PATCH] Add an Appendix E incident retrodiction harness The RFC's Appendix E collects 15 lessons from real escapes, but nothing answers the question they were written for: does the guard as it exists today reject the defect form each lesson describes? Recorded lessons without a reproducing probe decay into folklore, and the appendix's own E10 lesson says that a corpus exercised only locally is not a test. `loopx/semantics/incident_v0.json` pairs each lesson with a probe, the rule it expects, and an `expectation` authored from the lesson text. `scripts/ semantic_incident_retrodiction.py` drives the real pre-merge guard for each one, plus five negative controls: legal variations the guard must accept, which is what stops a fail-always check from scoring as coverage. A retrodiction asks whether today's guard rejects the defect form, never whether a historical revision passed; the guard did not exist then, so that answer would be no for all 15 and would measure nothing. Measured at 3937f6b99: would_have_caught=12/15, negative_control_false_positives=0/5 E02 uncovered no source outside loopx/ is scanned (declared root) E12 uncovered dropping 83 sources under loopx/extensions/ was unnoticed by every reach-consuming check; upper-bound budgets cannot see a narrowed reach, so the declared scope is unverified E14 uncovered 11 open decisions, no check enforces the inputs they name E12 and E14 are unstated gaps rather than declared boundaries. Recording them is the point: the ledger is a measurement, and the two gaps are the candidate work, not a silent pass. Probes fault-inject an input and then call the unmodified check, which requires the guard's real globals dict. `runpy.run_path` returns a copy, so an injection there is a silent no-op that reports "uncovered" without measuring anything; the module is imported instead, and every injection re-reads its own effect before its verdict is trusted. The registry-file probe additionally reads the committed registry back through the patched path before mutating. The negative controls pin the other direction, including that the equality anchors accept the committed registry and that the identifier-counting ratchet does not count a longer identifier containing the field token. tests/architecture/test_semantic_incident_retrodiction.py makes the ledger a committed check and freezes the baseline, so a lesson that was caught stays caught. Signed-off-by: song --- loopx/semantics/incident_v0.json | 166 +++++ scripts/semantic_incident_retrodiction.py | 566 ++++++++++++++++++ .../test_semantic_incident_retrodiction.py | 57 ++ 3 files changed, 789 insertions(+) create mode 100644 loopx/semantics/incident_v0.json create mode 100644 scripts/semantic_incident_retrodiction.py create mode 100644 tests/architecture/test_semantic_incident_retrodiction.py diff --git a/loopx/semantics/incident_v0.json b/loopx/semantics/incident_v0.json new file mode 100644 index 0000000000..213e3564ad --- /dev/null +++ b/loopx/semantics/incident_v0.json @@ -0,0 +1,166 @@ +{ + "schema_version": "loopx_semantic_incident_ledger_v0", + "meaning": "The incident and review lessons recorded in the semantic vocabulary RFC's Appendix E, each paired with the probe that retrodicts it against today's guard. A retrodiction asks whether the guard as it exists now rejects the defect form the lesson describes; it does not ask whether a historical revision passed, because the guard did not exist then and that answer is always no. `expectation` is authored from the lesson text, never from probe output; the harness reports where the measurement disagrees with it.", + "source": "docs/architecture/rfcs/semantic-vocabulary-convergence-v0.md#appendix-e", + "measured_at": "3937f6b99", + "incidents": [ + { + "id": "E01", + "class": "cross_runtime_hand_sync", + "defect": "Parallel constant lists across runtimes pass review until the day one side changes; a parity check must exist before the second copy is accepted.", + "expected_rule": "A registered cross-runtime vocabulary compares both owner modules against the registry value set.", + "probe": "cross_runtime_value_drift", + "expectation": "caught" + }, + { + "id": "E02", + "class": "fixture_private_copy", + "defect": "A private extraction of a shared constant looks harmless in a large module and is the most common way a schema version forks; test fixtures are second, and HANDOFF_MODES was copied into an e2e fixture.", + "expected_rule": "A copy outside the scanned roots is either visible to the scan or named as a declared boundary.", + "probe": "fixture_copy_visible", + "expectation": "uncovered" + }, + { + "id": "E03", + "class": "malformed_literal_ignored", + "defect": "A literal scan that only accepts [a-z_] silently skipped a _v2 spelling; malformed values must be reported, not ignored.", + "expected_rule": "Quoted literals are captured whatever their shape and the shape is validated separately.", + "probe": "malformed_value_reported", + "expectation": "caught" + }, + { + "id": "E04", + "class": "blind_guard_green_baseline", + "defect": "A scan pattern written from Python examples matched no TypeScript at all, and a guard that is green on a baseline containing violations proves only that it is blind.", + "expected_rule": "Both runtimes the registry claims to cover are anchored, so dropping one is a failure.", + "probe": "typescript_blind_guard", + "expectation": "caught" + }, + { + "id": "E05", + "class": "registry_weakens_validator", + "defect": "When the registry is both the specification and the validator's input, a data edit can weaken the validator.", + "expected_rule": "Recognised forms live in code, coverage counts are floored, and owners must be module::Symbol.", + "probe": "registry_weakness_rejected", + "expectation": "caught" + }, + { + "id": "E06", + "class": "name_only_ratchet", + "defect": "A ratchet on names alone lets an already-conflicting name gain a third spelling.", + "expected_rule": "Definition counts and name counts are both budgeted.", + "probe": "definitions_ratcheted", + "expectation": "caught" + }, + { + "id": "E07", + "class": "fleet_discovered_not_commit_time", + "defect": "A smoke the fleet discovers is not a commit-time check; if the answer is 'after merge', the invariant is a report, not a gate.", + "expected_rule": "The guard is selected by the pre-merge planning path for a diff that only touches the guarded code.", + "probe": "commit_time_selection", + "expectation": "caught" + }, + { + "id": "E08", + "class": "anchor_less_or_equal", + "defect": "An anchor compared with <= pins only the value it held when written, so every tightening below it is unprotected.", + "expected_rule": "Budget and floor anchors compare with equality, so neither raising nor lowering the registry value passes.", + "probe": "anchor_equality", + "expectation": "caught" + }, + { + "id": "E09", + "class": "field_carries_multiple_vocabularies", + "defect": "One field name can carry several vocabularies inside one envelope; a scan that sees the field cannot see the slot.", + "expected_rule": "Slots are recorded as a registered relation with an anchored composed domain.", + "probe": "slot_relation", + "expectation": "caught" + }, + { + "id": "E10", + "class": "committed_snapshot_coupling", + "defect": "A committed snapshot of the whole tree makes the guard's input depend on other people's merges.", + "expected_rule": "The inventory is computed at check time and no snapshot is tracked.", + "probe": "no_committed_snapshot", + "expectation": "caught" + }, + { + "id": "E11", + "class": "no_way_to_say_different_things", + "defect": "A name-keyed collision rule needs a way to say 'these are different things that share a name', or the honest fix and a rename lower the same number.", + "expected_rule": "A declaration names every defining module, so a rename cannot remove the debt.", + "probe": "scope_declaration", + "expectation": "caught" + }, + { + "id": "E12", + "class": "document_scope_exceeds_scanner", + "defect": "When a document widens its scope faster than the code, the two must be reconciled; a scope claim the scanner does not implement is a false invariant.", + "expected_rule": "The scan's implemented reach is measured at check time and compared with the declared claim.", + "probe": "scan_reach_disclosed", + "expectation": "uncovered" + }, + { + "id": "E13", + "class": "budgets_only_go_down", + "defect": "Budgets that only go down describe a direction; write the target table before the second milestone, or nobody can say when the work is done.", + "expected_rule": "The RFC carries a target table whose rows name the surface, the target, and the milestone that reaches it.", + "probe": "target_table", + "expectation": "caught" + }, + { + "id": "E14", + "class": "dangling_decision_reference", + "defect": "A decision that waits on 'a check' the RFC never defines is a dangling reference dressed as prudence.", + "expected_rule": "Every open decision names the invariant and the milestone that delivers its input.", + "probe": "dangling_decision", + "expectation": "uncovered" + }, + { + "id": "E15", + "class": "role_word_not_defined", + "defect": "Using a role word nineteen times is not defining it; until the roles are a table with a check per role, 'who writes this value' has no single answer.", + "expected_rule": "Roles are a declared table with a hierarchy and a producer check that fails closed.", + "probe": "roles_table", + "expectation": "caught" + } + ], + "negative_controls": [ + { + "id": "NC01", + "class": "single_value_not_a_vocabulary", + "meaning": "A closed set with one value is not a twin or a fork, so the collision rule must stay silent.", + "probe": "single_value_ignored" + }, + { + "id": "NC02", + "class": "unrelated_names_not_auto_merged", + "meaning": "Two names sharing a generic value set are a review candidate, never an automatic same_concept registration.", + "probe": "unrelated_not_autoregistered" + }, + { + "id": "NC03", + "class": "equality_anchor_accepts_current", + "meaning": "The equality anchors accept the registry as committed, so the comparison is not fail-always.", + "probe": "equality_anchor_accepts_current" + }, + { + "id": "NC04", + "class": "owner_shape_accepts_current", + "meaning": "The registry's own owner declarations pass the shape and value checks, so those checks are not fail-always.", + "probe": "owner_shape_accepts_current" + }, + { + "id": "NC05", + "class": "substring_not_counted", + "meaning": "The identifier-counting ratchet must not count a longer identifier that merely contains the field name.", + "probe": "substring_not_counted" + } + ], + "caught_baseline": { + "meaning": "Frozen at authoring time so the retrodiction cannot silently regress; it records what the guard caught then, not a target.", + "would_have_caught": 12, + "total": 15, + "negative_control_false_positives": 0 + } +} diff --git a/scripts/semantic_incident_retrodiction.py b/scripts/semantic_incident_retrodiction.py new file mode 100644 index 0000000000..17e910300f --- /dev/null +++ b/scripts/semantic_incident_retrodiction.py @@ -0,0 +1,566 @@ +#!/usr/bin/env python3 +"""Retrodict the RFC Appendix E incident lessons against today's guard. + +``loopx/semantics/incident_v0.json`` records the lessons the RFC's Appendix E +collected from real escapes, each paired with a probe. A probe asks one +question: **does the guard as it exists now reject the defect form this lesson +describes?** It never asks whether a historical revision passed, because the +guard did not exist then and that answer is always no for every lesson. + +Each probe drives the real pre-merge guard rather than a reimplementation, so a +probe cannot stay green after the check it exercises is removed. Negative +controls run the other direction: they are legal variations the guard must +accept, which is what stops a fail-always check from scoring as coverage. + +The ledger's ``expectation`` per lesson is authored from the lesson text. Where +the measurement disagrees, the mismatch is the finding and is printed last. +""" + +from __future__ import annotations + +import copy +import importlib.util +import json +import re +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from loopx.semantics.inventory import DEFAULT_ROOT, merge_candidate_groups # noqa: E402 + +SMOKE_PATH = REPO_ROOT / "examples" / "semantic-vocabulary-drift-smoke.py" +LEDGER_PATH = REPO_ROOT / "loopx" / "semantics" / "incident_v0.json" +LEDGER_SCHEMA_VERSION = "loopx_semantic_incident_ledger_v0" +LEDGER_KEYS = {"schema_version", "meaning", "source", "measured_at", "incidents", "negative_controls", "caught_baseline"} +INCIDENT_KEYS = {"id", "class", "defect", "expected_rule", "probe", "expectation"} +CONTROL_KEYS = {"id", "class", "meaning", "probe"} +VERDICTS = {"caught", "uncovered"} + + +class LedgerError(AssertionError): + pass + + +def require(condition: bool, message: str) -> None: + if not condition: + raise LedgerError(message) + + +def appendix_e_lessons(rfc_path: Path) -> int: + """Count the lessons the RFC's Appendix E records. + + The ledger is complete only if it probes every lesson the RFC collected, so + a lesson added to the appendix without a probe fails the harness instead of + being remembered by hand. + """ + text = rfc_path.read_text(encoding="utf-8") + require("## Appendix E" in text, "the RFC has no Appendix E; the ledger's source moved") + section = text.split("## Appendix E", 1)[1] + tail = section.split("## Appendix F", 1)[0] if "## Appendix F" in section else section + return sum(1 for line in tail.splitlines() if line.startswith("- ")) + + +def load_ledger() -> dict[str, Any]: + ledger = json.loads(LEDGER_PATH.read_text(encoding="utf-8")) + require(set(ledger) == LEDGER_KEYS, f"ledger keys must be exactly {sorted(LEDGER_KEYS)}") + require(ledger["schema_version"] == LEDGER_SCHEMA_VERSION, "ledger schema_version drift") + require(ledger["meaning"].strip() and ledger["source"].strip(), "ledger needs a meaning and a source") + require(LEDGER_PATH.parent.joinpath("vocabulary_v0.json").is_file(), "ledger must sit beside the vocabulary registry") + ids = [incident["id"] for incident in ledger["incidents"]] + require(len(ids) == len(set(ids)), "incident ids repeat") + for incident in ledger["incidents"]: + require(set(incident) == INCIDENT_KEYS, f"{incident.get('id')}: incident keys must be exactly {sorted(INCIDENT_KEYS)}") + require(incident["expectation"] in VERDICTS, f"{incident['id']}: expectation must be one of {sorted(VERDICTS)}") + for key in ("class", "defect", "expected_rule", "probe"): + require(incident[key].strip(), f"{incident['id']}: {key} must be non-empty") + control_ids = [control["id"] for control in ledger["negative_controls"]] + require(len(control_ids) == len(set(control_ids)), "negative control ids repeat") + for control in ledger["negative_controls"]: + require(set(control) == CONTROL_KEYS, f"{control.get('id')}: control keys must be exactly {sorted(CONTROL_KEYS)}") + baseline = ledger["caught_baseline"] + require(set(baseline) == {"meaning", "would_have_caught", "total", "negative_control_false_positives"}, + "caught_baseline keys must be meaning, would_have_caught, total, and negative_control_false_positives") + require(baseline["total"] == len(ledger["incidents"]), "caught_baseline.total must equal the incident count") + return ledger + + +class Guard: + """The pre-merge guard's own module, executed so its globals are patchable. + + Some probes fault-inject an input: they replace ``source`` with a reader + that returns a drifted module, or ``load_sources`` with one that returns a + narrowed tree, and then call the unmodified check function. That only works + against the module's real globals dict, which ``runpy.run_path`` does not + give -- it returns a copy, so an injection there is a silent no-op and the + probe would report 'uncovered' without measuring anything. Importing the + file as a module keeps the injection live, and every injection re-reads its + own effect before the verdict is trusted. + """ + + def __init__(self) -> None: + spec = importlib.util.spec_from_file_location("_semantic_drift_smoke_probe", SMOKE_PATH) + require(spec is not None and spec.loader is not None, "the guard module could not be loaded") + self.module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = self.module + spec.loader.exec_module(self.module) + self.ns = self.module.__dict__ + self.registry = self.ns["load_registry"]() + self.sources = self.ns["load_sources"](self.ns["REPO_ROOT"]) + self.inventory = self.ns["build_inventory"](self.ns["REPO_ROOT"], sources=self.sources) + + def rejects(self, call: Any) -> tuple[bool, str]: + try: + call() + except Exception as error: # noqa: BLE001 - any rejection is the signal + first = str(error).splitlines()[0][:140] + return True, f"{type(error).__name__}: {first}" + return False, "accepted" + + def with_registry_file(self, mutated: dict[str, Any], call: Any) -> tuple[bool, str]: + """Run a loader-level check against mutated registry content. + + The patch is proven live before the verdict is used: the committed + registry is written to the same temporary path and must read back + equal. A no-op patch fails the self-check instead of passing silently. + """ + handle = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False, encoding="utf-8") + handle.close() + temporary = Path(handle.name) + original = self.ns["REGISTRY_PATH"] + try: + temporary.write_text(json.dumps(self.registry), encoding="utf-8") + self.ns["REGISTRY_PATH"] = temporary + require(self.ns["load_registry"]() == self.registry, "registry injection is not live; the loader read the committed file") + temporary.write_text(json.dumps(mutated), encoding="utf-8") + return self.rejects(call) + finally: + self.ns["REGISTRY_PATH"] = original + temporary.unlink(missing_ok=True) + + def semantic_forks(self, registry: dict[str, Any] | None = None) -> int: + return self.ns["check_scope_declarations"](registry or self.registry, self.inventory) + + +# --- incident probes ----------------------------------------------------------------- + + +def probe_cross_runtime_value_drift(guard: Guard) -> tuple[bool, str]: + """E01: change the implementation on one runtime and see whether the other is compared.""" + target = None + for name, vocabulary in guard.registry["vocabularies"].items(): + owner = vocabulary["owners"]["typescript"] + if owner and "::" in owner and owner.split("::")[0].endswith(".ts"): + target = (name, owner) + break + require(target is not None, "no registered cross-runtime pair to mutate; the probe cannot measure anything") + name, owner = target + module, _symbol = owner.split("::") + victim = guard.ns["owner_values"](owner)[-1] + original = guard.ns["source"] + applied = [] + + def drifted(path: str): + file = original(path) + if path != module: + return file + applied.append(path) + if victim not in file.text: + return file + return guard.ns["SourceFile"](path=file.path, suffix=file.suffix, text=file.text.replace(victim, f"{victim}_drift", 1)) + + guard.ns["source"] = drifted + try: + require("drift" in guard.ns["owner_values"](owner)[-1], "implementation injection is not live") + rejected, detail = guard.rejects(lambda: guard.ns["check_owned_vocabularies"](guard.registry, guard.inventory)) + finally: + guard.ns["source"] = original + require(applied and set(applied) == {module}, f"the drifted module was never read (read={sorted(set(applied))})") + return rejected, f"{name} {owner}: {victim} -> {victim}_drift rejected={rejected} ({detail})" + + +def probe_fixture_copy_visible(guard: Guard) -> tuple[bool, str]: + """E02: a copy outside the scanned roots is invisible unless the boundary is declared.""" + roots = sorted({source.path.split("/")[0] for source in guard.sources}) + outside = [source.path for source in guard.sources if not source.path.startswith("loopx/")] + declared = guard.ns["LITERAL_SCAN_ROOTS"] + return bool(outside), f"scanned_roots={roots} declared_roots={declared} sources_outside={len(outside)}" + + +def probe_malformed_value_reported(guard: Guard) -> tuple[bool, str]: + """E03: a digit-bearing value is captured, and a malformed one is reported rather than skipped.""" + shape = guard.ns["VALUE_SHAPE"] + digit_accepted = shape.match("quota_skip_v2") is not None + malformed_reported = shape.match("Quota-Skip") is None + synthetic = guard.ns["SourceFile"]( + path="loopx/_probe.py", suffix=".py", text='packet = {"effective_action": "quota_skip_v2"}\n' + ) + rejected, detail = guard.rejects( + lambda: guard.ns["collect_literal_uses"](guard.ns["REPO_ROOT"], "effective_action", [synthetic]) + ) + captured = "quota_skip_v2" in guard.ns["collect_literal_uses"](guard.ns["REPO_ROOT"], "effective_action", [synthetic]) + return digit_accepted and malformed_reported and captured, ( + f"shape_accepts_digit={digit_accepted} shape_rejects_malformed={malformed_reported} " + f"scanner_captures_digit_value={captured} scanner_error={detail if rejected else 'none'}" + ) + + +def probe_typescript_blind_guard(guard: Guard) -> tuple[bool, str]: + """E04: dropping a runtime the registry claims to cover must fail.""" + mutated = copy.deepcopy(guard.registry) + for vocabulary in mutated["vocabularies"].values(): + if "literal_scan" in vocabulary: + vocabulary["literal_scan"]["suffixes"] = [ + suffix for suffix in vocabulary["literal_scan"]["suffixes"] if suffix != ".ts" + ] + rejected, detail = guard.rejects(lambda: guard.ns["check_coverage_floor"](mutated)) + typescript_sources = sum(1 for source in guard.sources if source.suffix == ".ts") + return rejected, f"dropping .ts rejected={rejected}; ts_sources_scanned={typescript_sources} ({detail})" + + +def probe_registry_weakness_rejected(guard: Guard) -> tuple[bool, str]: + """E05: a data edit must not be able to weaken the validator.""" + lowered = copy.deepcopy(guard.registry) + lowered["coverage_floor"]["vocabularies"] = 0 + floor_rejected, _ = guard.rejects(lambda: guard.ns["check_coverage_floor"](lowered)) + + bad_owner = copy.deepcopy(guard.registry) + vocabulary_name = next(name for name, item in bad_owner["vocabularies"].items() if item["owners"]["python"]) + bad_owner["vocabularies"][vocabulary_name]["owners"]["python"] = "not_a_module_symbol" + owner_rejected, owner_detail = guard.with_registry_file(bad_owner, guard.ns["load_registry"]) + + narrowed_scan = copy.deepcopy(guard.registry) + for vocabulary in narrowed_scan["vocabularies"].values(): + if "literal_scan" in vocabulary: + vocabulary["literal_scan"]["roots"] = ["loopx/control_plane"] + roots_rejected, _ = guard.rejects(lambda: guard.ns["check_coverage_floor"](narrowed_scan)) + return floor_rejected and owner_rejected and roots_rejected, ( + f"floor_lowered_rejected={floor_rejected} owner_shape_rejected={owner_rejected} " + f"scan_roots_narrowed_rejected={roots_rejected} ({owner_detail})" + ) + + +def probe_definitions_ratcheted(guard: Guard) -> tuple[bool, str]: + """E06: budgets must count definitions, not only names.""" + budgets = sorted(key for key in guard.ns["RATCHET_KEYS"] if key.endswith("_definitions")) + anchored = sorted(key for key in guard.ns["BUDGET_ANCHOR"] if key.endswith("_definitions")) + return len(budgets) >= 3 and budgets == anchored, f"definition_budgets={budgets} anchored={len(anchored)}" + + +def probe_commit_time_selection(guard: Guard) -> tuple[bool, str]: + """E07: the guard must be selected before merge, not discovered by the fleet.""" + planner = (REPO_ROOT / "loopx" / "canary" / "planner.py").read_text(encoding="utf-8") + premerge = (REPO_ROOT / "loopx" / "canary" / "premerge.py").read_text(encoding="utf-8") + command = "examples/semantic-vocabulary-drift-smoke.py" + planned = command in planner + inherited = "INHERITED_BASELINE_COMMANDS" in premerge and "planner" in premerge + guarded_roots = [hint for hint in ("loopx/", "scripts/", "examples/") if f'"{hint}"' in planner] + return planned and inherited, ( + f"named_in_planner={planned} premerge_consumes_planner={inherited} planner_trigger_hints={guarded_roots}" + ) + + +def probe_anchor_equality(guard: Guard) -> tuple[bool, str]: + """E08: neither direction of header drift may pass an equality anchor.""" + key = "multi_value_twins" + forks = guard.semantic_forks() + check = guard.ns["evaluate_inventory_budget_findings"] + summary = guard.inventory["summary"] + + def run(offset: int) -> bool: + mutated = copy.deepcopy(guard.registry) + mutated["inventory_ratchets"][key] = guard.ns["BUDGET_ANCHOR"][key] + offset + rejected, _ = guard.rejects(lambda: check(summary, forks, mutated["inventory_ratchets"])) + return rejected + + lowered, raised = run(-1), run(+1) + floor_lowered, _ = guard.rejects( + lambda: guard.ns["check_coverage_floor"]({**copy.deepcopy(guard.registry), "coverage_floor": { + **guard.registry["coverage_floor"], "vocabularies": guard.registry["coverage_floor"]["vocabularies"] - 1}}) + ) + return lowered and raised and floor_lowered, ( + f"budget_lowered_rejected={lowered} budget_raised_rejected={raised} coverage_floor_lowered_rejected={floor_lowered}" + ) + + +def probe_slot_relation(guard: Guard) -> tuple[bool, str]: + """E09: a field carrying several vocabularies records the slots as a relation.""" + groups = guard.registry["relations"].get("shared_field_names", []) + entry = next((group for group in groups if group.get("field") == "effective_action"), None) + slots = [slot.get("slot") for slot in (entry or {}).get("slots", [])] + reanchored = copy.deepcopy(guard.registry) + for group in reanchored["relations"]["shared_field_names"]: + for slot in group.get("slots", []): + if "slot" in slot: + slot["slot"] = "somewhere.else" + rejected, detail = guard.rejects(lambda: guard.ns["check_relations"](reanchored)) + return bool(entry) and "should_run.effective_action" in slots and rejected, ( + f"registered_slots={slots} unanchored_slot_rejected={rejected} ({detail})" + ) + + +def probe_no_committed_snapshot(guard: Guard) -> tuple[bool, str]: + """E10: the guard's input is computed at check time, not read from a snapshot.""" + tracked = subprocess.run( + ["git", "ls-files", "--", "loopx/semantics/inventory_v0.json"], + cwd=REPO_ROOT, capture_output=True, text=True, check=False, + ).stdout.strip() + text = SMOKE_PATH.read_text(encoding="utf-8") + computed = "build_inventory(REPO_ROOT" in text + return not tracked and computed, f"tracked_snapshot={tracked!r} inventory_computed_at_check_time={computed}" + + +def probe_scope_declaration(guard: Guard) -> tuple[bool, str]: + """E11: a rename cannot remove debt, and every defining module must be named.""" + declarations = guard.registry["scope_declarations"] + if not declarations: + return False, "no scope declarations are recorded" + name = next(iter(declarations)) + truncated = copy.deepcopy(guard.registry) + truncated["scope_declarations"][name]["contexts"] = [] + missing_rejected, _ = guard.rejects(lambda: guard.ns["check_scope_declarations"](truncated, guard.inventory)) + + renamed = copy.deepcopy(guard.registry) + renamed["scope_declarations"][f"{name}_renamed"] = renamed["scope_declarations"].pop(name) + rename_rejected, rename_detail = guard.rejects(lambda: guard.ns["check_scope_declarations"](renamed, guard.inventory)) + return missing_rejected and rename_rejected, ( + f"declared={sorted(declarations)} missing_context_rejected={missing_rejected} " + f"rename_rejected={rename_rejected} ({rename_detail})" + ) + + +def probe_scan_reach_disclosed(guard: Guard) -> tuple[bool, str]: + """E12: narrow the scanner's real reach and see whether any check notices. + + A document can widen its scope claim without the scanner widening with it. + The probe drops a subtree that holds no registered owner symbol and no + scanned literal field, then runs every reach-consuming check. Each of them + bounds its count from above, so a narrowed reach that still satisfies the + floors is invisible and the scope claim is unverified. + """ + dropped_prefix = "loopx/extensions/" + full = guard.ns["load_sources"](guard.ns["REPO_ROOT"]) + kept = [source for source in full if not source.path.startswith(dropped_prefix)] + dropped = len(full) - len(kept) + require(dropped > 0, f"{dropped_prefix} holds no sources; the probe cannot narrow anything") + original = guard.ns["load_sources"] + + def narrowed(repo_root: str, root: str | None = None): + scope = root if root is not None else DEFAULT_ROOT + return [source for source in kept if source.path.startswith(scope)] + + guard.ns["load_sources"] = narrowed + try: + require(len(guard.ns["load_sources"](guard.ns["REPO_ROOT"])) == len(kept), "reach injection is not live") + outcomes = {} + outcomes["literal_vocabularies"] = guard.rejects( + lambda: guard.ns["check_literal_vocabularies"](guard.registry, kept) + )[0] + outcomes["retirement_budgets"] = guard.rejects( + lambda: guard.ns["check_retirement_budgets"](guard.registry, kept) + )[0] + outcomes["dual_runtime_twins"] = guard.rejects(lambda: guard.ns["check_dual_runtime_twins"](guard.registry))[0] + finally: + guard.ns["load_sources"] = original + return any(outcomes.values()), ( + f"dropped_sources={dropped} ({dropped_prefix}) reaching_checks_that_noticed={sorted(k for k, v in outcomes.items() if v)}" + ) + + +def probe_target_table(guard: Guard) -> tuple[bool, str]: + """E13: the RFC must publish a target table, or the work has no completion test.""" + rfc = (REPO_ROOT / guard.registry["rfc"]).read_text(encoding="utf-8") + marker = "Target when this RFC closes" + if marker not in rfc: + return False, "no target table in the RFC" + body = rfc.split(marker, 1)[1].splitlines()[1:] + rows = 0 + for line in body: + if not line.startswith("|"): + if rows: + break + continue + if set(line) <= set("| -"): + continue + rows += 1 + return rows >= 8, f"target_table_rows={rows}" + + +def probe_dangling_decision(guard: Guard) -> tuple[bool, str]: + """E14: a decision waiting on an undefined check is a dangling reference.""" + rfc = (REPO_ROOT / guard.registry["rfc"]).read_text(encoding="utf-8") + if "## 12. Open decisions" not in rfc: + return False, "no open decisions section" + section = rfc.split("## 12. Open decisions", 1)[1].split("## Appendix A", 1)[0] + items = re.findall(r"^\d+\.\s\*\*(.+?)\*\*", section, re.M) + guard_text = "\n".join( + path.read_text(encoding="utf-8", errors="replace") + for path in sorted((REPO_ROOT / "loopx" / "canary").glob("*.py")) + ) + enforcing = "open_decision" in guard_text + return enforcing, f"open_decision_items={len(items)} check_enforcing_named_inputs={enforcing}" + + +def probe_roles_table(guard: Guard) -> tuple[bool, str]: + """E15: a role word is defined only when it is a table with a check per role.""" + model = guard.registry["formal_model"] + roles = sorted(model["roles"]) + hierarchy = model["role_hierarchy"] + invariants = {item["id"] for item in model["invariants"]} + role_invariants = sorted(invariants & {"F1_producer_closedness", "F3_consumer_domain_closedness"}) + anchor = sorted(guard.ns["PRODUCER_VOCABULARY_ANCHOR"]) + return bool(roles) and bool(hierarchy) and bool(role_invariants) and bool(anchor), ( + f"roles={roles} hierarchy={sorted(hierarchy)} role_invariants={role_invariants} producer_vocabulary_anchor={anchor}" + ) + + +# --- negative controls: legal variations the guard must accept ----------------------- + + +def control_single_value_ignored(guard: Guard) -> tuple[bool, str]: + """A one-value set must not be reported as a collision candidate.""" + fake = {section: [] for section in ("python_enums", "python_closed_sets", "python_literal_aliases", "typescript_const_arrays")} + fake["python_closed_sets"] = [ + {"name": "ONLY_ONE", "module": "loopx/_probe_a.py", "values": ["solo"]}, + {"name": "ALSO_ONE", "module": "loopx/_probe_b.py", "values": ["solo"]}, + ] + groups = merge_candidate_groups(fake) + return bool(groups), f"groups_for_single_value_sets={groups}" + + +def control_unrelated_not_autoregistered(guard: Guard) -> tuple[bool, str]: + """Two names sharing a generic value set must not become a same_concept relation.""" + pairs = [group["members"] for group in guard.registry["relations"]["same_concept"]] + offending = [ + pair for pair in pairs + if any("confidence_levels" in member for member in pair) + and any("edge_case_complexities" in member for member in pair) + ] + return bool(offending), f"same_concept_groups={len(pairs)} offending={offending}" + + +def control_equality_anchor_accepts_current(guard: Guard) -> tuple[bool, str]: + """The equality anchors must accept the committed registry.""" + check = guard.ns["evaluate_inventory_budget_findings"] + rejected, detail = guard.rejects( + lambda: check(guard.inventory["summary"], guard.semantic_forks(), guard.registry["inventory_ratchets"]) + ) + return rejected, f"committed_registry_rejected={rejected} ({detail})" + + +def control_owner_shape_accepts_current(guard: Guard) -> tuple[bool, str]: + """The registry's own owners must pass the shape and value checks.""" + rejected, detail = guard.rejects(lambda: guard.ns["check_owned_vocabularies"](guard.registry, guard.inventory)) + return rejected, f"committed_registry_rejected={rejected} ({detail})" + + +def control_substring_not_counted(guard: Guard) -> tuple[bool, str]: + """A longer identifier containing the field token must not be counted.""" + field = next(iter(guard.registry["retirement_ledger"]["should_run_legacy_decision_fields"]["fields"])) + counter = guard.ns["count_identifier_modules"] + exact = guard.ns["SourceFile"](path="loopx/_probe.py", suffix=".py", text=f"value = {field}\n") + extended = guard.ns["SourceFile"](path="loopx/_probe.py", suffix=".py", text=f"value = {field}_extra\n") + counted_exact = counter(field, ".py", [exact]) + counted_extended = counter(field, ".py", [extended]) + return counted_exact != 1 or counted_extended != 0, ( + f"field={field} exact_count={counted_exact} longer_identifier_count={counted_extended}" + ) + + +INCIDENT_PROBES = { + "cross_runtime_value_drift": probe_cross_runtime_value_drift, + "fixture_copy_visible": probe_fixture_copy_visible, + "malformed_value_reported": probe_malformed_value_reported, + "typescript_blind_guard": probe_typescript_blind_guard, + "registry_weakness_rejected": probe_registry_weakness_rejected, + "definitions_ratcheted": probe_definitions_ratcheted, + "commit_time_selection": probe_commit_time_selection, + "anchor_equality": probe_anchor_equality, + "slot_relation": probe_slot_relation, + "no_committed_snapshot": probe_no_committed_snapshot, + "scope_declaration": probe_scope_declaration, + "scan_reach_disclosed": probe_scan_reach_disclosed, + "target_table": probe_target_table, + "dangling_decision": probe_dangling_decision, + "roles_table": probe_roles_table, +} + +CONTROL_PROBES = { + "single_value_ignored": control_single_value_ignored, + "unrelated_not_autoregistered": control_unrelated_not_autoregistered, + "equality_anchor_accepts_current": control_equality_anchor_accepts_current, + "owner_shape_accepts_current": control_owner_shape_accepts_current, + "substring_not_counted": control_substring_not_counted, +} + + +def main() -> int: + ledger = load_ledger() + guard = Guard() + lessons = appendix_e_lessons(REPO_ROOT / guard.registry["rfc"]) + require(lessons == len(ledger["incidents"]), ( + f"Appendix E records {lessons} lessons but the ledger probes {len(ledger['incidents'])}; " + "every lesson needs a probe and every probe needs a lesson" + )) + for ledger_key, probes in (("incidents", INCIDENT_PROBES), ("negative_controls", CONTROL_PROBES)): + for entry in ledger[ledger_key]: + require(entry["probe"] in probes, f"{entry['id']}: unknown probe {entry['probe']!r}") + + caught, mismatches, details = 0, [], [] + for incident in ledger["incidents"]: + observed, detail = INCIDENT_PROBES[incident["probe"]](guard) + verdict = "caught" if observed else "uncovered" + caught += observed + details.append((incident["id"], incident["class"], verdict, detail)) + if verdict != incident["expectation"]: + mismatches.append(f"{incident['id']} {incident['class']}: expected {incident['expectation']}, measured {verdict}") + + false_positives = [] + for control in ledger["negative_controls"]: + violated, detail = CONTROL_PROBES[control["probe"]](guard) + if violated: + false_positives.append(f"{control['id']} {control['class']}: rejected a legal variation ({detail})") + + total = len(ledger["incidents"]) + print("semantic-incident-retrodiction: " + ("ok" if not mismatches else "mismatches")) + print(f" appendix_e_lessons={lessons} ledger_incidents={total}") + print(f" would_have_caught={caught}/{total}") + print(f" negative_control_false_positives={len(false_positives)}/{len(ledger['negative_controls'])}") + baseline = ledger["caught_baseline"] + print(f" caught_baseline={baseline['would_have_caught']}/{baseline['total']}") + for incident_id, incident_class, verdict, detail in details: + print(f" {incident_id} {verdict:9s} {incident_class}: {detail}") + for mismatch in mismatches: + print(f" expectation-mismatch: {mismatch}") + if '--report' in sys.argv[1:]: + for incident in ledger["incidents"]: + print(f" {incident['id']} defect: {incident['defect']}") + print(f" {incident['id']} rule: {incident['expected_rule']}") + if '--json' in sys.argv[1:]: + print(json.dumps({ + "would_have_caught": caught, + "total": total, + "negative_control_false_positives": len(false_positives), + "mismatches": mismatches, + "verdicts": {incident_id: verdict for incident_id, _c, verdict, _d in details}, + }, indent=2, sort_keys=True)) + + require(caught >= baseline["would_have_caught"], + f"retrodiction regressed: {caught}/{total} below the frozen baseline {baseline['would_have_caught']}/{total}; " + "a lesson that was caught must stay caught") + require(len(false_positives) <= baseline["negative_control_false_positives"], + "negative controls regressed: " + "; ".join(false_positives)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except LedgerError as error: + print(f"semantic-incident-retrodiction: FAILED: {error}", file=sys.stderr) + raise SystemExit(1) from error diff --git a/tests/architecture/test_semantic_incident_retrodiction.py b/tests/architecture/test_semantic_incident_retrodiction.py new file mode 100644 index 0000000000..11166a5e98 --- /dev/null +++ b/tests/architecture/test_semantic_incident_retrodiction.py @@ -0,0 +1,57 @@ +"""The Appendix E retrodiction is a committed check, not a report. + +The RFC's own incident list recorded that a mutation corpus which is only +exercised locally is not a test. These tests hold the retrodiction harness to +the same standard: the ledger is well formed, and the count of lessons today's +guard rejects cannot fall below the frozen baseline. +""" + +from __future__ import annotations + +import json +import re +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "scripts" / "semantic_incident_retrodiction.py" +LEDGER_PATH = REPO_ROOT / "loopx" / "semantics" / "incident_v0.json" + + +def load_ledger() -> dict: + return json.loads(LEDGER_PATH.read_text(encoding="utf-8")) + + +def test_ledger_is_well_formed() -> None: + ledger = load_ledger() + incidents = ledger["incidents"] + controls = ledger["negative_controls"] + + assert len(incidents) >= 15, "every Appendix E lesson needs a probe" + assert len({incident["id"] for incident in incidents}) == len(incidents), "incident ids repeat" + assert len({control["id"] for control in controls}) == len(controls), "control ids repeat" + for incident in incidents: + assert incident["expectation"] in {"caught", "uncovered"}, incident["id"] + assert incident["defect"].strip() and incident["expected_rule"].strip(), incident["id"] + assert ledger["caught_baseline"]["total"] == len(incidents) + assert ledger["caught_baseline"]["negative_control_false_positives"] == 0 + + +def test_retrodiction_holds_its_frozen_baseline() -> None: + result = subprocess.run( + [sys.executable, str(SCRIPT)], cwd=REPO_ROOT, capture_output=True, text=True, check=False + ) + assert result.returncode == 0, result.stdout + result.stderr + + caught = re.search(r"would_have_caught=(\d+)/(\d+)", result.stdout) + controls = re.search(r"negative_control_false_positives=(\d+)/", result.stdout) + assert caught and controls, result.stdout + + ledger = load_ledger() + baseline = ledger["caught_baseline"] + assert (int(caught.group(1)), int(caught.group(2))) == (baseline["would_have_caught"], baseline["total"]), ( + "the ledger baseline and the measurement disagree; a lesson that was caught must stay caught" + ) + assert int(controls.group(1)) <= baseline["negative_control_false_positives"], result.stdout + assert "appendix_e_lessons=15" in result.stdout, "the ledger no longer covers the RFC's lessons"