From dd7b8399e6822be3f4539a2ec80b1a7c3bbeffd8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 19:44:49 +0000 Subject: [PATCH 1/2] =?UTF-8?q?version=5Fskew:=20deep=20--pypi=20leg=20?= =?UTF-8?q?=E2=80=94=20flag=20floors=20naming=20yanked=20releases?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tick-path check compares floors against local git tags, so it cannot see the other way a floor goes bad: the release it names being yanked on PyPI afterwards (the 2026-07 shape, where every floor named the yanked 2026.7.6.649). The gap was acknowledged in the module docstring and owned nowhere. `python -m heart.checks.version_skew --pypi` now asks the PyPI JSON API whether each floor still names an installable (non-yanked) release and whether any installable release satisfies it. Statuses: UNSATISFIABLE (nothing installable >= floor — same defect class as the tag leg's UNSATISFIABLE, readiness RED), FLOOR_YANKED (floors are >= bounds, so a yanked floor with newer installable releases still resolves — readiness YELLOW, fix by bumping the floor), UNKNOWN (PyPI unreachable — STALE, never a false block), OK/BAD as before. Network, so never part of the tick: the probe is on-demand/nightly only and persists to its own version_skew_pypi.json sidecar so the tick's version_skew.json rewrite can never clobber its evidence (and vice versa). run_pypi() is side-effect-free like run(); one fetch per distinct package, not per workspace. Snapshot, readiness legs (+weights) and a dashboard section wired; 484 tests pass (15 new). Task: PyAutoMind draft/feature/pyautoheart/version_skew_yank_awareness.md Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01SKw8oaLZoD3cZnmETMDEkc --- heart/checks/version_skew.py | 123 ++++++++++++++++++++++++++++++++--- heart/dashboard.py | 17 +++++ heart/readiness.py | 35 ++++++++++ heart/state.py | 1 + tests/test_readiness.py | 38 +++++++++++ tests/test_version_skew.py | 85 ++++++++++++++++++++++++ 6 files changed, 289 insertions(+), 10 deletions(-) diff --git a/heart/checks/version_skew.py b/heart/checks/version_skew.py index a323be8..6cf57c9 100644 --- a/heart/checks/version_skew.py +++ b/heart/checks/version_skew.py @@ -29,11 +29,25 @@ - **UNKNOWN** — the library isn't checked out / carries no release tags, so the newest release can't be resolved; surfaced as caution, never a hard block. -Not covered here (deeper, non-tick checks own it): whether the floor names a -release that was later *yanked* on PyPI — that needs the PyPI API, not git tags. +The yank gap is owned by the **deep PyPI leg** (``--pypi``): git tags cannot see +a release being *yanked* on PyPI after the fact, so ``--pypi`` asks the PyPI +JSON API whether each floor still names an installable (non-yanked) release and +whether *any* installable release satisfies it. Network — never part of the +tick; run it on demand or from a nightly. Its statuses: + +- **UNSATISFIABLE** — no installable release >= floor exists on PyPI (every + candidate yanked): same defect class as the tag-based UNSATISFIABLE. +- **FLOOR_YANKED** — the floor version itself is yanked/absent but a newer + installable release satisfies it; floors are ``>=`` so installs still + resolve — warn, fix by bumping the floor. +- **OK** / **BAD** / **UNKNOWN** — as above; UNKNOWN covers PyPI unreachable + (offline is caution, never a false hard block). + An informational "floor lags far behind newest" signal is a possible future add. -The result lands at ``$HEART_STATE_DIR/version_skew.json``. +The tick result lands at ``$HEART_STATE_DIR/version_skew.json``; the ``--pypi`` +leg at ``version_skew_pypi.json`` (a sibling file, so the tick never clobbers +on-demand evidence). """ from __future__ import annotations @@ -149,20 +163,103 @@ def run(root: Path = PYAUTO_ROOT) -> dict[str, Any]: return {"workspaces": workspaces} +# --------------------------------------------------------------------------- # +# Deep PyPI leg (``--pypi``) — yank-awareness. Network; never run from the tick. +# --------------------------------------------------------------------------- # + +PYPI_URL = "https://pypi.org/pypi/{package}/json" +PYPI_TIMEOUT_S = 10 + + +def fetch_pypi_releases(package: str) -> dict[str, list] | None: + """The package's ``releases`` map from the PyPI JSON API, or None when the + API is unreachable/unparseable — offline must degrade to UNKNOWN, never a + false hard block.""" + import urllib.request + + try: + with urllib.request.urlopen( + PYPI_URL.format(package=package), timeout=PYPI_TIMEOUT_S + ) as resp: + data = json.load(resp) + except Exception: + return None + releases = data.get("releases") + return releases if isinstance(releases, dict) else None + + +def _installable(files: list) -> bool: + """A release is installable iff at least one of its files is not yanked + (PyPI marks yank per file; a fileless release installs nothing).""" + return any(isinstance(f, dict) and not f.get("yanked") for f in files or []) + + +def pypi_floor_status(floor: str | None, releases: dict[str, list] | None) -> str: + """OK / FLOOR_YANKED / UNSATISFIABLE / UNKNOWN / BAD for one floor vs PyPI. + + Floors are ``>=`` bounds, so a yanked floor with a newer installable + release still resolves at install time — that is FLOOR_YANKED (fix by + bumping the floor to an installable release), not a hard block. No + installable release >= floor at all is the same defect class as the + tag-based UNSATISFIABLE. + """ + if releases is None: + return "UNKNOWN" + ft = _tuple(floor or "") + if ft is None: + return "BAD" + installable = { + v.strip() + for v, files in releases.items() + if _TAG_RE.match(v.strip()) and _installable(files) + } + if not any(_tuple(v) >= ft for v in installable): + return "UNSATISFIABLE" + return "OK" if (floor or "").strip() in installable else "FLOOR_YANKED" + + +def run_pypi(root: Path = PYAUTO_ROOT) -> dict[str, Any]: + """Side-effect-free like run(): one PyPI fetch per distinct package, one + entry per floored workspace.""" + releases_by_package: dict[str, dict[str, list] | None] = {} + workspaces = [] + for workspace, (repo, pkg) in workspace_library().items(): + floor = read_workspace_floor(workspace, root) + if floor is None: + continue # no floor recorded → not a candidate + if pkg not in releases_by_package: + releases_by_package[pkg] = fetch_pypi_releases(pkg) + workspaces.append( + { + "workspace": workspace, + "library": repo, + "package": pkg, + "floor": floor, + "status": pypi_floor_status(floor, releases_by_package[pkg]), + } + ) + return {"workspaces": workspaces} + + def main(argv: list[str]) -> int: - result = run() + pypi = "--pypi" in argv + result = run_pypi() if pypi else run() sys.path.insert(0, str(HEART_HOME)) from heart import state - # Persist only here, at the tick/CLI entrypoint — run() is side-effect-free - # so library callers (and the test suite) can never clobber live state. - state.atomic_write_json(HEART_STATE_DIR / "version_skew.json", result) + # Persist only here, at the tick/CLI entrypoint — run()/run_pypi() are + # side-effect-free so library callers (and the test suite) can never + # clobber live state. The --pypi leg gets its own sidecar so the tick's + # version_skew.json rewrite never clobbers on-demand PyPI evidence. + name = "version_skew_pypi.json" if pypi else "version_skew.json" + state.atomic_write_json(HEART_STATE_DIR / name, result) from heart.heart_color import c_ok, c_warn, c_fail, c_info, c_meta, glyph_ok, glyph_warn, glyph_fail workspaces = result["workspaces"] unsatisfiable = [w for w in workspaces if w["status"] == "UNSATISFIABLE"] bad = [w for w in workspaces if w["status"] == "BAD"] + yanked = [w for w in workspaces if w["status"] == "FLOOR_YANKED"] unknown = [w for w in workspaces if w["status"] == "UNKNOWN"] blocking = unsatisfiable + bad # release-blocking statuses if blocking: @@ -173,13 +270,19 @@ def main(argv: list[str]) -> int: if bad: parts.append(c_warn(f"{len(bad)} bad")) label = " ".join(parts) - elif unknown: + elif yanked or unknown: glyph = glyph_warn() - label = c_warn(f"{len(unknown)} unknown") + parts = [] + if yanked: + parts.append(c_warn(f"{len(yanked)} floor yanked")) + if unknown: + parts.append(c_warn(f"{len(unknown)} unknown")) + label = " ".join(parts) else: glyph = glyph_ok() label = c_ok(f"{len(workspaces)} floors satisfiable") - print(f"{glyph} {c_info('version_skew')} {label} {c_meta(f'({len(workspaces)} floors)')}") + check_name = "version_skew --pypi" if pypi else "version_skew" + print(f"{glyph} {c_info(check_name)} {label} {c_meta(f'({len(workspaces)} floors)')}") return 0 diff --git a/heart/dashboard.py b/heart/dashboard.py index 3179426..84b7edc 100644 --- a/heart/dashboard.py +++ b/heart/dashboard.py @@ -545,6 +545,23 @@ def build_board( elif skew: sections.append(Section("version_skew", "Version skew", OK, "all floors satisfiable", [])) + # Version skew — PyPI yank leg (deep `version_skew --pypi`; the slice is + # absent until that on-demand probe has run, so no section = not yet run) -- + skew_pypi = (snapshot.get("version_skew_pypi") or {}).get("workspaces") or [] + pypi_off = [w for w in skew_pypi if isinstance(w, dict) and w.get("status") not in ("OK", None)] + pypi_blocking = [w for w in pypi_off if str(w.get("status")).upper() in ("UNSATISFIABLE", "BAD")] + if pypi_off: + st = FAIL if pypi_blocking else WARN + summary = f"{len(pypi_blocking)} blocking" if pypi_blocking else f"{len(pypi_off)} unresolved" + details = [ + f"{w.get('status')}: {w.get('workspace')} floor {w.get('floor')} " + f"({w.get('package')} on PyPI)" + for w in pypi_off[:8] + ] + sections.append(Section("version_skew_pypi", "Version skew (PyPI)", st, summary, details)) + elif skew_pypi: + sections.append(Section("version_skew_pypi", "Version skew (PyPI)", OK, "all floors installable", [])) + # Install verification --------------------------------------------------- vi = snapshot.get("verify_install") or {} if isinstance(vi, dict) and "ready" in vi: diff --git a/heart/readiness.py b/heart/readiness.py index 2dea3e9..5726fea 100644 --- a/heart/readiness.py +++ b/heart/readiness.py @@ -126,6 +126,10 @@ "parked": (5, 15), "skew_bad": (25, 50), "skew_unknown": (10, 30), + "skew_pypi_unsatisfiable": (25, 50), + "skew_pypi_bad": (25, 50), + "skew_pypi_floor_yanked": (8, 24), + "skew_pypi_unknown": (10, 30), "install_not_ready": (40, 40), "install_non_release": (10, 10), "install_stale": (10, 10), @@ -402,6 +406,37 @@ def scope_local(msg: str, key: str) -> None: stale.append(f"{w.get('workspace')}: newest {w.get('library')} release unknown") hit("skew_unknown") + # --- version skew, PyPI yank leg (deep `version_skew --pypi`; the slice is + # absent until that on-demand probe has run — absence is no signal) --- + skew_pypi = snapshot.get("version_skew_pypi") + if isinstance(skew_pypi, dict): + for w in skew_pypi.get("workspaces") or []: + if not isinstance(w, dict): + continue + status = str(w.get("status", "")).upper() + if status == "UNSATISFIABLE": + red.append( + f"{w.get('workspace')}: no installable {w.get('package')} release " + f"on PyPI satisfies floor {w.get('floor')} (all candidates yanked)" + ) + hit("skew_pypi_unsatisfiable") + elif status == "BAD": + red.append( + f"{w.get('workspace')}: unparseable floor {w.get('floor')} (PyPI leg)" + ) + hit("skew_pypi_bad") + elif status == "FLOOR_YANKED": + yellow.append( + f"{w.get('workspace')}: floor {w.get('floor')} names a yanked/" + f"unavailable {w.get('package')} release — bump the floor" + ) + hit("skew_pypi_floor_yanked") + elif status == "UNKNOWN": + stale.append( + f"{w.get('workspace')}: PyPI unreachable for {w.get('package')}" + ) + hit("skew_pypi_unknown") + # --- manifest drift (YELLOW — identity hygiene vs PyAutoMind/repos.yaml) --- manifest = snapshot.get("manifest_drift") if isinstance(manifest, dict) and manifest.get("available"): diff --git a/heart/state.py b/heart/state.py index 339118a..14af1d0 100644 --- a/heart/state.py +++ b/heart/state.py @@ -80,6 +80,7 @@ def aggregate() -> dict[str, Any]: "test_run": _read_json_or_default(HEART_STATE_DIR / "test_run.json", {}), "workspace_testmode_timing": _read_json_or_default(HEART_STATE_DIR / "workspace_testmode_timing.json", {}), "version_skew": _read_json_or_default(HEART_STATE_DIR / "version_skew.json", {}), + "version_skew_pypi": _read_json_or_default(HEART_STATE_DIR / "version_skew_pypi.json", {}), "manifest_drift": _read_json_or_default(HEART_STATE_DIR / "manifest_drift.json", {}), "verify_install": _read_json_or_default(HEART_STATE_DIR / "verify_install.json", {}), "url_check": _read_json_or_default(HEART_STATE_DIR / "url_check.json", {}), diff --git a/tests/test_readiness.py b/tests/test_readiness.py index 79f0988..32231af 100644 --- a/tests/test_readiness.py +++ b/tests/test_readiness.py @@ -178,6 +178,44 @@ def test_version_skew_unknown_is_stale_tier(): assert any("release unknown" in r for r in v["stale_reasons"]) +# --- version skew, PyPI yank leg (deep `version_skew --pypi`) ------------------ +# The slice is absent from make_snapshot(), so the all-green test already proves +# absence is no signal (the probe is on-demand, not part of the tick). + +def test_version_skew_pypi_unsatisfiable_is_red(): + snap = make_snapshot(version_skew_pypi={"workspaces": [ + {"workspace": "autolens_workspace", "library": "PyAutoLens", "package": "autolens", + "floor": "2026.7.6.649", "status": "UNSATISFIABLE"} + ]}) + v = compute(snap) + assert v["verdict"] == "red" + assert any("no installable" in r and "yanked" in r for r in v["red_reasons"]) + assert v["score"] == 75 + + +def test_version_skew_pypi_yanked_floor_is_yellow(): + # Floors are >= bounds: a yanked floor with newer installable releases + # still resolves at install time — warn (bump the floor), never block. + snap = make_snapshot(version_skew_pypi={"workspaces": [ + {"workspace": "autolens_workspace", "library": "PyAutoLens", "package": "autolens", + "floor": "2026.7.6.649", "status": "FLOOR_YANKED"} + ]}) + v = compute(snap) + assert v["verdict"] == "yellow" + assert any("yanked" in r and "bump the floor" in r for r in v["yellow_reasons"]) + + +def test_version_skew_pypi_unknown_is_stale_tier(): + # PyPI unreachable (offline box) must degrade to caution, never a false RED. + snap = make_snapshot(version_skew_pypi={"workspaces": [ + {"workspace": "autolens_workspace", "library": "PyAutoLens", "package": "autolens", + "floor": "2026.7.9.1", "status": "UNKNOWN"} + ]}) + v = compute(snap) + assert v["verdict"] == "stale" + assert any("PyPI unreachable" in r for r in v["stale_reasons"]) + + def test_install_verification_failed_is_red(): snap = make_snapshot(verify_install={ "ready": False, "ts": "2026-06-01T00:00:00+00:00", diff --git a/tests/test_version_skew.py b/tests/test_version_skew.py index a122750..2f41d31 100644 --- a/tests/test_version_skew.py +++ b/tests/test_version_skew.py @@ -121,6 +121,72 @@ def test_autolens_assistant_is_a_polled_workspace(): assert mapping["autolens_assistant"] == ("PyAutoLens", "autolens") +# --- deep PyPI yank leg (--pypi) ---------------------------------------------- + +def _files(*yanked): + return [{"yanked": y} for y in yanked] + + +# 2026.7.6.649 fully yanked (the real 2026-07 incident); two installable newer. +RELEASES = { + "2026.7.6.649": _files(True, True), + "2026.7.9.1": _files(False, False), + "2026.7.15.1": _files(False), +} + + +@pytest.mark.parametrize("floor,releases,expected", [ + ("2026.7.9.1", RELEASES, "OK"), # floor installable + ("2026.7.6.649", RELEASES, "FLOOR_YANKED"), # floor yanked, newer installable + ("2026.7.1.1", RELEASES, "FLOOR_YANKED"), # floor absent from PyPI, newer installable + ("2026.8.1.1", RELEASES, "UNSATISFIABLE"), # nothing >= floor exists + ("2026.7.9.1", {"2026.7.9.1": _files(True)}, "UNSATISFIABLE"), # everything >= floor yanked + ("2026.7.9.1", {"2026.7.9.1": []}, "UNSATISFIABLE"), # fileless release installs nothing + ("not.a.version", RELEASES, "BAD"), + ("2026.7.9.1", None, "UNKNOWN"), # PyPI unreachable → never a false block +]) +def test_pypi_floor_status(floor, releases, expected): + assert vs.pypi_floor_status(floor, releases) == expected + + +def test_run_pypi_one_fetch_per_package(tmp_path, monkeypatch): + # autolens_workspace and HowToLens both map to package `autolens` → the + # probe must fetch each distinct package once, not once per workspace. + for ws in ("autolens_workspace", "HowToLens"): + cfg = tmp_path / ws / "config" + cfg.mkdir(parents=True) + (cfg / "general.yaml").write_text("version:\n minimum_library_version: 2026.7.9.1\n") + calls = [] + monkeypatch.setattr(vs, "fetch_pypi_releases", lambda pkg: calls.append(pkg) or RELEASES) + result = vs.run_pypi(root=tmp_path) + assert calls == ["autolens"] + by_ws = {w["workspace"]: w for w in result["workspaces"]} + assert by_ws["autolens_workspace"]["status"] == "OK" + assert by_ws["HowToLens"]["package"] == "autolens" + + +def test_run_pypi_offline_is_unknown(tmp_path, monkeypatch): + ws = tmp_path / "autolens_workspace" / "config" + ws.mkdir(parents=True) + (ws / "general.yaml").write_text("version:\n minimum_library_version: 2026.7.9.1\n") + monkeypatch.setattr(vs, "fetch_pypi_releases", lambda pkg: None) + result = vs.run_pypi(root=tmp_path) + w = {x["workspace"]: x for x in result["workspaces"]}["autolens_workspace"] + assert w["status"] == "UNKNOWN" + + +def test_run_pypi_flags_yanked_floor(tmp_path, monkeypatch): + # The 2026-07 incident shape: floor names the yanked release while newer + # installable releases exist → FLOOR_YANKED (warn), not a hard block. + ws = tmp_path / "autolens_workspace" / "config" + ws.mkdir(parents=True) + (ws / "general.yaml").write_text("version:\n minimum_library_version: 2026.7.6.649\n") + monkeypatch.setattr(vs, "fetch_pypi_releases", lambda pkg: RELEASES) + result = vs.run_pypi(root=tmp_path) + w = {x["workspace"]: x for x in result["workspaces"]}["autolens_workspace"] + assert w["status"] == "FLOOR_YANKED" + + # --- state-dir isolation (the 2026-07-15 clobber incident's sibling) ----------- def test_run_writes_nothing_to_state_dir(tmp_path): @@ -144,3 +210,22 @@ def test_main_persists_result_to_state_dir(monkeypatch): assert vs.main(["version_skew"]) == 0 written = json.loads((Path(os.environ["HEART_STATE_DIR"]) / "version_skew.json").read_text()) assert written == {"workspaces": []} + + +def test_main_pypi_persists_to_sibling_file(monkeypatch): + """--pypi writes version_skew_pypi.json and never touches the tick's + version_skew.json — the tick must not clobber on-demand PyPI evidence and + vice versa.""" + import json + import os + from pathlib import Path + state_dir = Path(os.environ["HEART_STATE_DIR"]) + tick_file = state_dir / "version_skew.json" + tick_before = tick_file.read_text() if tick_file.is_file() else None + payload = {"workspaces": [{"workspace": "autolens_workspace", "status": "FLOOR_YANKED"}]} + monkeypatch.setattr(vs, "run_pypi", lambda root=vs.PYAUTO_ROOT: payload) + assert vs.main(["version_skew", "--pypi"]) == 0 + written = json.loads((state_dir / "version_skew_pypi.json").read_text()) + assert written == payload + tick_after = tick_file.read_text() if tick_file.is_file() else None + assert tick_after == tick_before From 34a3f00c108a718118a69a7c6e66fb781757f3cf Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 20:34:42 +0000 Subject: [PATCH 2/2] tests: use autolens_assistant, not HowToLens, in the one-fetch-per-package test The tenant firewall gate flagged HowToLens as a new instance fact in organ code; autolens_assistant is an already-present fact in this file and maps to the same package, so the test proves the same dedup behaviour. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01SKw8oaLZoD3cZnmETMDEkc --- tests/test_version_skew.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_version_skew.py b/tests/test_version_skew.py index 2f41d31..f07a8c2 100644 --- a/tests/test_version_skew.py +++ b/tests/test_version_skew.py @@ -150,9 +150,9 @@ def test_pypi_floor_status(floor, releases, expected): def test_run_pypi_one_fetch_per_package(tmp_path, monkeypatch): - # autolens_workspace and HowToLens both map to package `autolens` → the - # probe must fetch each distinct package once, not once per workspace. - for ws in ("autolens_workspace", "HowToLens"): + # autolens_workspace and autolens_assistant both map to package `autolens` + # → the probe must fetch each distinct package once, not once per workspace. + for ws in ("autolens_workspace", "autolens_assistant"): cfg = tmp_path / ws / "config" cfg.mkdir(parents=True) (cfg / "general.yaml").write_text("version:\n minimum_library_version: 2026.7.9.1\n") @@ -162,7 +162,7 @@ def test_run_pypi_one_fetch_per_package(tmp_path, monkeypatch): assert calls == ["autolens"] by_ws = {w["workspace"]: w for w in result["workspaces"]} assert by_ws["autolens_workspace"]["status"] == "OK" - assert by_ws["HowToLens"]["package"] == "autolens" + assert by_ws["autolens_assistant"]["package"] == "autolens" def test_run_pypi_offline_is_unknown(tmp_path, monkeypatch):