009 — CPython warnings 3.14.5：我量的那個通道，是被我的儀器改變過的

{
  "name": "warnings-channel-recut",
  "what_it_is": "A notice channel whose memory is part of the declared structure, and whose observers must say whether they perturb it.",
  "examined": {
    "project": "CPython warnings",
    "version": "3.14.5",
    "license": "PSF-2.0",
    "why_this_one": "Archaeology 008 measured logging's deprecation alias through the warnings channel and reported 'the old name emits one DeprecationWarning'. That sentence is true once per call site per process, and the only reason my observer saw it every time is that the observer itself resets the channel's memory. I measured a channel my instrument had changed.",
    "measured": {
      "warnings_py_lines": 99,
      "_py_warnings_py_lines": 869,
      "_warnings": "C builtin; warnings.warn.__module__ == '_warnings'",
      "names_re_exported_by_the_facade": 47,
      "public_names": 9,
      "filters_are_the_same_object": "warnings.filters is _warnings.filters",
      "memory": "__warningregistry__ per module, keyed by (text, category, lineno)",
      "invalidation": "registry['version'] != _filters_version -> the registry is discarded"
    },
    "reproduced": {
      "baseline_no_instrument": "1 emission out of 5 calls from one site",
      "catch_warnings_only": "5 of 5 — control added 2026-08-09 after Pragma pointed out that the row below cannot isolate the cause",
      "catch_warnings_plus_simplefilter_always": "5 of 5",
      "what_the_control_establishes": "entering the context manager is sufficient; simplefilter is not what restores the notice, so the cause is the filter MUTATION rather than the filter setting",
      "why": "catch_warnings mutates the filter list, _filters_mutated() bumps _filters_version, and every registry whose version no longer matches is discarded"
    }
  },
  "the_finding": "Three layers, and the one you import is not the one that runs: a 99-line facade re-exporting 47 names from an 869-line pure-Python module, with a C builtin actually providing warn(). Underneath, the channel has memory — a notice fires once per (text, category, line) per module — and the standard way to observe it erases that memory. The observation and the observed are the same filter list.",
  "the_correction_it_forces": "Archaeology 008's permit reads 'one DeprecationWarning naming the replacement'. Measured without an instrument that resets the channel, the honest statement is 'one on the first call from a given site in a process, and none afterwards'. The contract was not wrong about what it saw; it was silent about what made it visible.",
  "channel": {
    "id": "notices",
    "memory": "per (source, text) — a notice is delivered once until the memory is cleared",
    "who_may_clear_it": "declared in SCL"
  },
  "observers": {
    "passive": {
      "perturbs_the_channel": false,
      "sees": "exactly what a caller in the same process would see",
      "cost": "a notice already delivered is invisible to it"
    },
    "resetting": {
      "perturbs_the_channel": true,
      "sees": "every notice the code attempts to send",
      "cost": "it reports a frequency no caller experiences"
    }
  },
  "sets": {
    "FMS": "this file: the channel, its memory, and the two observers with their declared perturbation",
    "SCL": "which observer this deployment permits, and who may clear the memory",
    "SMS": "the channel itself — deliver, remember, clear",
    "TMS": "two observers, each importing nothing",
    "DMS": "what was seen, under which observer, and what that observer cannot see"
  },
  "non_goals": [
    "Being the warnings module. One channel, two observers, no filter language.",
    "Claiming CPython is wrong. Once-per-site is the correct default for a notice channel; the finding is about measuring it, not about it."
  ],
  "evidence_drift_note": "This block said 'wrapped 3/3' until 2026-08-09 while the README, the board post and the island test all said 5/5. Same direction, two numbers in one public record — Metron flagged it. The figures now come from the same five-call run the test performs."
}

{
  "policy_version": "1.0",
  "_comment": [
    "A deployment that permits only the passive observer gets numbers a caller",
    "would recognise. One that permits the resetting observer gets numbers no",
    "caller experiences. Neither is wrong; reporting which one produced a",
    "figure is not optional."
  ],
  "permitted_observers": ["passive", "resetting"],
  "default_observer": "passive",
  "may_clear_memory": ["test-harness"]
}

"""Which observer, and who may clear the channel's memory."""
import json
import pathlib

_C = json.loads((pathlib.Path(__file__).parent / "policy.json").read_text(encoding="utf-8"))

permits = lambda name: name in _C["permitted_observers"]          # noqa: E731
default_observer = lambda: _C["default_observer"]                 # noqa: E731
may_clear = lambda actor: actor in _C["may_clear_memory"]         # noqa: E731

"""The channel, with its memory made explicit.

CPython keeps this in a module-global `__warningregistry__` keyed by
(text, category, lineno), invalidated whenever the filter list is mutated. That
is efficient and it is invisible: nothing in the caller's view says "this notice
has a memory, and something you did earlier reset it".

Here the memory is a field on the channel, and clearing it is an operation with
an actor.
"""


class Channel:
    def __init__(self):
        self.delivered = []        # what a listener actually received
        self.attempted = []        # every send, whether delivered or not
        self._seen = set()
        self.clears = []

    def send(self, source, text):
        """Deliver a notice once per (source, text) until the memory is cleared."""
        self.attempted.append((source, text))
        key = (source, text)
        if key in self._seen:
            return False
        self._seen.add(key)
        self.delivered.append((source, text))
        return True

    def clear_memory(self, actor, may_clear):
        """Forget what has been delivered. Refused unless policy allows it."""
        if not may_clear(actor):
            self.clears.append((actor, "refused"))
            return False
        self._seen.clear()
        self.clears.append((actor, "cleared"))
        return True

    def suppressed(self):
        return len(self.attempted) - len(self.delivered)

"""Watches without touching. Imports nothing.

Sees what a caller in the same process sees — which means a notice already
delivered before this observer started is invisible to it, and it will report
zero for a channel that is working exactly as designed.
"""

NAME = "passive"
PERTURBS = False


def observe(channel, run):
    before = len(channel.delivered)
    run()
    return {"delivered": channel.delivered[before:], "attempted_since": None}

"""Clears the memory first, so every send is delivered. Imports nothing.

This is what `catch_warnings` does, though not on purpose: it mutates the filter
list, which bumps the version every registry is compared against, so every
registry is discarded. The reset is a side effect of the isolation.

It sees more, and what it sees is a frequency no caller experiences.
"""

NAME = "resetting"
PERTURBS = True


def observe(channel, run, clear, times=1):
    """Clear before EACH observation, which is what one catch_warnings block per
    call actually does — the isolation is per block, so the memory is discarded
    once per observation rather than once per session."""
    before = len(channel.delivered)
    for _ in range(times):
        if not clear():
            return {"delivered": [], "refused": "policy did not permit clearing the memory"}
        run()
    return {"delivered": channel.delivered[before:], "attempted_since": None}

"""What was seen, under which observer, and what that observer cannot see."""


def render(record, results, channel, policy_note):
    out = ["", "== the same code, watched two ways"]
    for name, seen in results.items():
        spec = record["observers"][name]
        mark = "!! " if spec["perturbs_the_channel"] else "ok "
        out.append(f"  {mark} {name:<10} delivered {len(seen['delivered'])}"
                   f"   perturbs the channel: {spec['perturbs_the_channel']}")
        out.append(f"        sees   {spec['sees']}")
        out.append(f"        cost   {spec['cost']}")

    out.append("")
    out.append(f"  attempts made by the code : {len(channel.attempted)}")
    out.append(f"  notices actually delivered: {len(channel.delivered)}")
    out.append(f"  suppressed by the memory  : {channel.suppressed()}")
    out.append("")
    out.append(f"  {policy_note}")
    out.append("")
    out.append("  A figure from a perturbing observer is not wrong. It answers a different")
    out.append("  question — 'how often does the code try' rather than 'how often does anyone")
    out.append("  hear' — and a report that does not name the observer lets a reader take one")
    out.append("  for the other.")
    return "\n".join(out) + "\n"

"""The island test, and the measurement that corrects archaeology 008.

    python src/island_test.py

Section 4 is why this entry exists. Archaeology 008 reported that logging's
deprecated alias "emits one DeprecationWarning". It does — once per call site
per process. My observer saw it every time because the observer resets the
channel's memory as a side effect of isolating itself.
"""
import inspect
import io
import json
import pathlib
import re
import sys
import warnings

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from SCL import policy  # noqa: E402
from SMS.channel import Channel  # noqa: E402
from TMS.observers import passive, resetting  # noqa: E402

RECORD = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))
FAILURES = []


def check(label, ok, detail=""):
    print(f"  {'PASS' if ok else 'FAIL'}  {label}{' - ' + detail if detail else ''}")
    if not ok:
        FAILURES.append(label)


print("\n== 1. each observer is an island, and its perturbation is MEASURED")
for module in (passive, resetting):
    source = (HERE / "TMS" / "observers" / f"{module.NAME}.py").read_text(encoding="utf-8")
    reaches = re.findall(r"^\s*(?:from|import)\s+(\S+)", source, re.M)
    check(f"observers/{module.NAME} imports nothing", not reaches, ", ".join(reaches) or "none")


def measure_perturbation(run_observer):
    """Run an observer against a primed channel and watch what it does to it.

    Metron and Pragma both found, independently on 2026-08-09, that this
    section previously compared `module.PERTURBS` with the record's
    `perturbs_the_channel` — two declarations agreeing with each other. Setting
    both to False left the resetting observer clearing the memory twice and the
    check still passed. That is the same shape as "the permit was prose nobody
    read", one layer up: a self-declaration verified against a copy of itself.
    """
    ch = Channel()
    ch.send("primed.py:1", "already delivered")
    before = (len(ch.clears), len(ch.delivered))
    run_observer(ch)
    after = (len(ch.clears), len(ch.delivered))
    return {
        "cleared_memory": after[0] > before[0]
                          and any(state == "cleared" for _, state in ch.clears),
        "clear_events": [c for c in ch.clears],
    }


measured = {
    passive.NAME: measure_perturbation(
        lambda ch: passive.observe(ch, lambda: ch.send("primed.py:1", "already delivered"))),
    resetting.NAME: measure_perturbation(
        lambda ch: resetting.observe(ch, lambda: ch.send("primed.py:1", "already delivered"),
                                     lambda: ch.clear_memory("test-harness", policy.may_clear))),
}
for module in (passive, resetting):
    declared = RECORD["observers"][module.NAME]["perturbs_the_channel"]
    observed = measured[module.NAME]["cleared_memory"]
    check(f"observers/{module.NAME}: the record's claim matches what it actually did",
          declared == observed,
          f"record says {declared}, the run shows cleared_memory={observed} "
          f"({measured[module.NAME]['clear_events'] or 'no clear events'})")
    check(f"observers/{module.NAME}: and the module constant agrees too",
          module.PERTURBS == observed, f"module says {module.PERTURBS}, measured {observed}")

# The failing case: a declaration that lies is caught by the measurement, which
# is the whole point of measuring rather than comparing two labels.
lying = measure_perturbation(
    lambda ch: resetting.observe(ch, lambda: ch.send("primed.py:1", "x"),
                                 lambda: ch.clear_memory("test-harness", policy.may_clear)))
check("an observer declared passive while clearing memory would be caught",
      lying["cleared_memory"] is True,
      "the measurement sees the clear events regardless of what any file says")

print("\n== 2. the same code, two observers, different numbers")
ch = Channel()
send = lambda: ch.send("a.py:1", "notice")  # noqa: E731
seen_passive = passive.observe(ch, lambda: [send() for _ in range(5)])
seen_reset = resetting.observe(ch, send, lambda: ch.clear_memory("test-harness", policy.may_clear), times=5)
check("passive sees the notice once out of five", len(seen_passive["delivered"]) == 1,
      f"{len(seen_passive['delivered'])} of 5")
check("resetting sees it five times out of five", len(seen_reset["delivered"]) == 5,
      f"{len(seen_reset['delivered'])} of 5")
check("and the code did exactly the same thing both times", len(ch.attempted) == 10,
      f"{len(ch.attempted)} attempts, {len(ch.delivered)} delivered, {ch.suppressed()} suppressed")

print("\n== 3. the checks can fail")
ch2 = Channel()
refused = resetting.observe(ch2, lambda: ch2.send("b.py:1", "x"),
                            lambda: ch2.clear_memory("nobody", policy.may_clear), times=3)
check("an actor policy does not permit cannot clear the memory",
      refused.get("refused") is not None, refused.get("refused", ""))
check("and the refusal is recorded on the channel",
      ("nobody", "refused") in ch2.clears, str(ch2.clears))
ch3 = Channel()
ch3.send("c.py:1", "first")
ch3.send("c.py:1", "first")
ch3.send("c.py:2", "first")
check("the memory is keyed by source AND text, not by text alone",
      len(ch3.delivered) == 2, f"{len(ch3.delivered)} delivered from 3 sends")

print("\n== 4. measured against CPython warnings 3.14.5 itself")
check("this interpreter is the examined version", sys.version.split()[0] == "3.14.5",
      sys.version.split()[0])

facade = pathlib.Path(inspect.getsourcefile(warnings))
facade_lines = len(facade.read_text(encoding="utf-8").splitlines())
check("the module you import is a 99-line facade", facade_lines == 99, f"{facade_lines} lines")

import _py_warnings  # noqa: E402
impl = pathlib.Path(inspect.getsourcefile(_py_warnings))
impl_lines = len(impl.read_text(encoding="utf-8").splitlines())
check("the pure-Python implementation is elsewhere and far larger",
      impl_lines > 800, f"_py_warnings.py = {impl_lines} lines")
check("and warn() actually comes from the C module",
      warnings.warn.__module__ == "_warnings", warnings.warn.__module__)

import _warnings  # noqa: E402
check("the facade's filter list IS the C module's", warnings.filters is _warnings.filters,
      "the observation and the observed are the same object")

# The memory, measured without an instrument that resets it.
captured = []
warnings.showwarning = lambda message, category, filename, lineno, file=None, line=None: \
    captured.append((category.__name__, str(message)))
warnings.simplefilter("default")


def emit():
    warnings.warn("a deprecated thing", DeprecationWarning, stacklevel=2)


before = len(captured)
for _ in range(5):
    emit()
uninstrumented = len(captured) - before
check("five calls from one site emit ONE warning", uninstrumented == 1,
      f"{uninstrumented} of 5 — the channel remembers")

registry = sys.modules["__main__"].__dict__.get("__warningregistry__", {})
check("and the memory is visible as __warningregistry__",
      any(isinstance(k, tuple) for k in registry), str(registry)[:70])

instrumented = 0
for _ in range(5):
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        emit()
        instrumented += len(w)
check("the same five calls, each wrapped, emit FIVE", instrumented == 5,
      f"{instrumented} of 5 — catch_warnings mutates the filters and every registry is discarded")

# Pragma, 2026-08-09: the row above uses catch_warnings AND simplefilter, so it
# demonstrates the combination and cannot isolate the cause. A caller reading it
# could reasonably conclude simplefilter("always") is what does the work. The
# control below removes simplefilter entirely.
catch_only = 0
for _ in range(5):
    with warnings.catch_warnings(record=True) as w:
        emit()
        catch_only += len(w)
check("CONTROL: catch_warnings alone, no simplefilter, also emits FIVE",
      catch_only == 5,
      f"{catch_only} of 5 — so entering the context manager is sufficient; "
      "simplefilter is not what restores the notice")
check("which isolates the cause to the filter mutation, not the filter setting",
      catch_only == instrumented == 5 and uninstrumented == 1,
      "baseline 1, catch-only 5, catch+simplefilter 5")
check("so the instrument changed the channel it was measuring",
      uninstrumented == 1 and instrumented == 5,
      "1 vs 5 from identical code — this is archaeology 008's observer")

print("\n== 5. the correction this forces on archaeology 008")
prior = pathlib.Path(HERE / ".." / ".." / "008-cpython-logging-warn" / "src" / "FMS" / "architecture.json").resolve()
if prior.exists():
    prior_record = json.loads(prior.read_text(encoding="utf-8"))
    permit = prior_record["compatibility_aliases"][0]["equivalence"]["allowed_deltas"][0]
    check("archaeology 008's permit is still stated as 'one DeprecationWarning'",
          "one-deprecation-warning" in str(permit.get("predicate", "")), str(permit))
    check("and that is true only under a resetting observer",
          uninstrumented == 1,
          "a passive observer sees one on the FIRST call from a site and none after")
    observer = prior_record["observers"]["three-channel-v1"]
    check("008's observer now names this in what it cannot see",
          "memory" in observer["what_it_cannot_see"] or "registry" in observer["what_it_cannot_see"],
          observer["what_it_cannot_see"][:80])
else:
    check("archaeology 008 is on disk to be corrected", False, str(prior))

print("\n== 6. what this entry does not claim")
check("the record does not call CPython wrong",
      "not about it" in RECORD["non_goals"][1], RECORD["non_goals"][1][:70])
check("once-per-site is recorded as the correct default for a notice channel",
      "correct default" in RECORD["non_goals"][1])

print("")
if FAILURES:
    print(f"  {len(FAILURES)} check(s) failed: {', '.join(FAILURES)}")
    raise SystemExit(1)
print("  island test passed")

"""Send the same notice five times, watched two ways.

    python src/main.py
"""
import json
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).parent))

from DMS import report  # noqa: E402
from SCL import policy  # noqa: E402
from SMS.channel import Channel  # noqa: E402
from TMS.observers import passive, resetting  # noqa: E402

HERE = pathlib.Path(__file__).parent
RECORD = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))


def main():
    channel = Channel()

    def one_call():
        channel.send("ledger.py:41", "the 'warn' name is deprecated, use 'warning'")

    def five_calls():
        for _ in range(5):
            one_call()

    results = {}
    if policy.permits(passive.NAME):
        results[passive.NAME] = passive.observe(channel, five_calls)
    if policy.permits(resetting.NAME):
        # Five observations, each isolated — the shape of a test that wraps
        # every call in its own catch_warnings block.
        results[resetting.NAME] = resetting.observe(
            channel, one_call,
            lambda: channel.clear_memory("test-harness", policy.may_clear),
            times=5,
        )

    note = (f"SCL: default observer is {policy.default_observer()}; "
            f"may clear the memory: {'test-harness' if policy.may_clear('test-harness') else 'nobody'}")
    sys.stdout.write(report.render(RECORD, results, channel, note))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())