version_skew: deep --pypi leg — flag floors naming yanked releases

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
 
 

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:

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"):

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", {}),

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",

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 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")
+    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["autolens_assistant"]["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