Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions health_agent/capabilities.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -75,8 +75,8 @@ continuous_checks:
gate_role: "YELLOW if workspace validation not passing / stale / unknown (standing debt, advisory)"
- id: version_skew
impl: heart/checks/version_skew.py
measures: "each workspace's pinned version vs the installed library"
gate_role: "RED if AHEAD / MISMATCH (general.yaml vs version.txt) / BAD; YELLOW if BEHIND / UNKNOWN"
measures: "each workspace's compatibility floor (minimum_library_version) vs the newest library release tag"
gate_role: "RED if UNSATISFIABLE (floor exceeds newest release) / BAD; STALE if UNKNOWN (newest release unresolvable)"
- id: manifest_drift
impl: heart/checks/manifest_drift.py
measures: "repo identity (name/owner) in Heart/Build/labels lists and checkout origins vs the PyAutoMind/repos.yaml body map"
Expand Down
214 changes: 100 additions & 114 deletions heart/checks/version_skew.py
Original file line number Diff line number Diff line change
@@ -1,29 +1,37 @@
"""heart/checks/version_skew.py — workspace pin vs installed library version.

PyAutoHands's ``verify_workspace_versions.sh`` blocks a release if any
workspace's pinned version is AHEAD of its installed library — but that only
runs at ``pre_build`` time, so day-to-day the skew is invisible. This check
makes it continuous.

For each polled workspace that carries a pin, it resolves:

- the **pinned** version: ``config/general.yaml`` → ``version.workspace_version``,
falling back to a ``version.txt`` at the workspace root (same precedence as
``verify_workspace_versions.sh``);
- the **installed** library version: by regex-reading ``__version__`` from the
matching library's ``<pkg>/__init__.py`` source — never importing the heavy
library (keeps the tick cheap).

Versions are ``YYYY.M.D.B`` tuples; the comparison yields MATCH / AHEAD /
BEHIND / BAD. Two further statuses mirror the hard-fail conditions of
``verify_workspace_versions.sh`` so Heart is the single authoritative readiness
gate:

- **MISMATCH** — ``config/general.yaml`` and ``version.txt`` both exist and
disagree (a release-blocking inconsistency in ``verify_workspace_versions.sh``).
- **UNKNOWN** — the library ``__init__.py`` could not be read (the workspace is
pinned but the library isn't checked out); surfaced as caution, never a hard
block, matching the script's ``SKIP (cannot import …)`` behaviour.
"""heart/checks/version_skew.py — workspace compatibility floor vs newest release.

Under the pre-2026-07 model this check compared a workspace's recorded pin
(``config/general.yaml`` → ``version.workspace_version`` / ``version.txt``)
against the library ``__init__.py`` ``__version__`` stamp. That model is gone:
releases no longer commit the stamp or the pin back to ``main`` (PyAutoConf#119 /
PyAutoBuild#121 — the daily "Update version to X" commits were the CI-storm /
cron-pause engine), so *both* artifacts are frozen and the old check was inert —
permanently MATCH on stale values, unfailable by any release.

The live invariant now is the **compatibility floor**: each workspace records
``config/general.yaml`` → ``version.minimum_library_version`` — the oldest
library release whose API its scripts require — and users must be able to install
a release that satisfies it. So this check compares:

- the **floor**: ``version.minimum_library_version`` for the workspace;
- the **newest release**: the highest ``YYYY.M.D.B`` git tag on the mapped
library checkout (read with ``git tag`` — no library import, no network, cheap
enough for the <30s tick).

Statuses:

- **UNSATISFIABLE** — floor > newest release: no released version satisfies the
floor, so a fresh ``pip install`` cannot pair with the workspace. Release-
blocking (this is the invariant nothing guarded before: the old leg was
unfailable by releases).
- **OK** — floor <= newest release.
- **BAD** — floor or newest release is unparseable.
- **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.
An informational "floor lags far behind newest" signal is a possible future add.

The result lands at ``$HEART_STATE_DIR/version_skew.json``.
"""
Expand All @@ -33,6 +41,7 @@
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any
Expand All @@ -48,7 +57,8 @@
or Path.home() / ".pyauto-heart"
)

_VERSION_RE = re.compile(r'^__version__\s*=\s*["\']([^"\']+)["\']', re.MULTILINE)
_TAG_RE = re.compile(r"^\d{4}\.\d+\.\d+\.\d+$")


def workspace_library(config_path: Path | str = CONFIG_PATH) -> dict[str, tuple[str, str]]:
"""workspace name -> (library repo dir, package dir), from the policy
Expand All @@ -59,48 +69,47 @@ def workspace_library(config_path: Path | str = CONFIG_PATH) -> dict[str, tuple[
return {ws: (spec["library"], spec["package"]) for ws, spec in block.items()}


def read_library_version(repo: str, pkg: str, root: Path = PYAUTO_ROOT) -> str | None:
"""Regex-read ``__version__`` from ``<repo>/<pkg>/__init__.py`` (no import)."""
init = root / repo / pkg / "__init__.py"
def read_workspace_floor(workspace: str, root: Path = PYAUTO_ROOT) -> str | None:
"""The compatibility floor: ``config/general.yaml`` →
``version.minimum_library_version``. Returns None when absent (the
workspace is not a floor candidate)."""
general = root / workspace / "config" / "general.yaml"
if not general.is_file():
return None
try:
m = _VERSION_RE.search(init.read_text())
data = yaml.safe_load(general.read_text()) or {}
except yaml.YAMLError:
return None
floor = (data.get("version") or {}).get("minimum_library_version")
return str(floor).strip() if floor else None


def _newest_version(tags: list[str]) -> str | None:
"""Highest ``YYYY.M.D.B`` tag by numeric tuple, or None if none match."""
versions = [t.strip() for t in tags if _TAG_RE.match(t.strip())]
if not versions:
return None
return max(versions, key=lambda v: tuple(int(p) for p in v.split(".")))


def newest_release_tag(repo: str, root: Path = PYAUTO_ROOT) -> str | None:
"""Newest ``YYYY.M.D.B`` release tag on the library checkout (``git tag``),
or None when the repo isn't a checkout or carries no release tags. No
network — reads local tags only, so it stays inside the tick budget."""
repo_dir = root / repo
if not (repo_dir / ".git").exists():
return None
try:
proc = subprocess.run(
["git", "-C", str(repo_dir), "tag"],
capture_output=True,
text=True,
)
except OSError:
return None
return m.group(1) if m else None


def read_workspace_pin_sources(
workspace: str, root: Path = PYAUTO_ROOT
) -> tuple[str | None, str | None]:
"""Return ``(general_yaml_version, version_txt)`` — either may be None.

Kept separate from :func:`read_workspace_pin` so callers that need to detect
a general.yaml ↔ version.txt disagreement (MISMATCH) can see both sources.
"""
ws = root / workspace
yaml_v: str | None = None
general = ws / "config" / "general.yaml"
if general.is_file():
try:
data = yaml.safe_load(general.read_text()) or {}
pin = (data.get("version") or {}).get("workspace_version")
if pin:
yaml_v = str(pin).strip()
except yaml.YAMLError:
pass
txt_v: str | None = None
vtxt = ws / "version.txt"
if vtxt.is_file():
t = vtxt.read_text().strip()
if t:
txt_v = t
return yaml_v, txt_v


def read_workspace_pin(workspace: str, root: Path = PYAUTO_ROOT) -> str | None:
"""Pinned version: config/general.yaml:version.workspace_version, then version.txt."""
yaml_v, txt_v = read_workspace_pin_sources(workspace, root)
return yaml_v if yaml_v is not None else txt_v
if proc.returncode != 0:
return None
return _newest_version(proc.stdout.splitlines())


def _tuple(v: str) -> tuple[int, ...] | None:
Expand All @@ -110,54 +119,34 @@ def _tuple(v: str) -> tuple[int, ...] | None:
return None


def compare(pinned: str | None, installed: str | None) -> str:
"""MATCH / AHEAD (pinned > installed) / BEHIND (pinned < installed) / BAD."""
pt, it = _tuple(pinned or ""), _tuple(installed or "")
if pt is None or it is None:
def compare_floor(floor: str | None, newest: str | None) -> str:
"""UNSATISFIABLE (floor > newest release) / OK (floor <= newest) / BAD."""
ft, nt = _tuple(floor or ""), _tuple(newest or "")
if ft is None or nt is None:
return "BAD"
if pt == it:
return "MATCH"
return "AHEAD" if pt > it else "BEHIND"
return "UNSATISFIABLE" if ft > nt else "OK"


def run(root: Path = PYAUTO_ROOT) -> dict[str, Any]:
workspaces = []
for workspace, (repo, pkg) in workspace_library().items():
yaml_v, txt_v = read_workspace_pin_sources(workspace, root)
if yaml_v is None and txt_v is None:
continue # no pin (e.g. *_test workspaces) → not a skew candidate
installed = read_library_version(repo, pkg, root)

# general.yaml ↔ version.txt disagreement is the same release-blocking
# condition verify_workspace_versions.sh fails on.
if yaml_v is not None and txt_v is not None and yaml_v != txt_v:
workspaces.append(
{
"workspace": workspace,
"library": repo,
"pinned": yaml_v,
"version_txt": txt_v,
"installed": installed,
"status": "MISMATCH",
}
)
continue

pinned = yaml_v if yaml_v is not None else txt_v
# Library not checked out → cannot compare. Caution, not a hard block
# (mirrors the script's "SKIP (cannot import <pkg>)").
status = "UNKNOWN" if installed is None else compare(pinned, installed)
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
newest = newest_release_tag(repo, root)
# Library not checked out / no tags → cannot resolve the newest release.
# Caution, not a hard block.
status = "UNKNOWN" if newest is None else compare_floor(floor, newest)
workspaces.append(
{
"workspace": workspace,
"library": repo,
"pinned": pinned,
"installed": installed,
"floor": floor,
"newest_release": newest,
"status": status,
}
)
result = {"workspaces": workspaces}
return result
return {"workspaces": workspaces}


def main(argv: list[str]) -> int:
Expand All @@ -172,28 +161,25 @@ def main(argv: list[str]) -> int:
from heart.heart_color import c_ok, c_warn, c_fail, c_info, c_meta, glyph_ok, glyph_warn, glyph_fail

workspaces = result["workspaces"]
ahead = [w for w in workspaces if w["status"] == "AHEAD"]
behind = [w for w in workspaces if w["status"] == "BEHIND"]
mismatch = [w for w in workspaces if w["status"] == "MISMATCH"]
unsatisfiable = [w for w in workspaces if w["status"] == "UNSATISFIABLE"]
bad = [w for w in workspaces if w["status"] == "BAD"]
blocking = ahead + mismatch + bad # release-blocking statuses
unknown = [w for w in workspaces if w["status"] == "UNKNOWN"]
blocking = unsatisfiable + bad # release-blocking statuses
if blocking:
glyph = glyph_fail()
parts = []
if ahead:
parts.append(c_fail(f"{len(ahead)} ahead"))
if mismatch:
parts.append(c_fail(f"{len(mismatch)} mismatch"))
if unsatisfiable:
parts.append(c_fail(f"{len(unsatisfiable)} unsatisfiable"))
if bad:
parts.append(c_warn(f"{len(bad)} bad"))
label = " ".join(parts)
elif behind:
elif unknown:
glyph = glyph_warn()
label = c_warn(f"{len(behind)} behind")
label = c_warn(f"{len(unknown)} unknown")
else:
glyph = glyph_ok()
label = c_ok(f"{len(workspaces)} in sync")
print(f"{glyph} {c_info('version_skew')} {label} {c_meta(f'({len(workspaces)} pinned)')}")
label = c_ok(f"{len(workspaces)} floors satisfiable")
print(f"{glyph} {c_info('version_skew')} {label} {c_meta(f'({len(workspaces)} floors)')}")
return 0


Expand Down
12 changes: 6 additions & 6 deletions heart/dashboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -501,19 +501,19 @@ def build_board(
sections.append(_unobs_section("version_skew", "Version skew"))
else:
skew = (snapshot.get("version_skew") or {}).get("workspaces") or []
off = [w for w in skew if isinstance(w, dict) and w.get("status") not in ("MATCH", None)]
blocking = [w for w in off if str(w.get("status")).upper() in ("AHEAD", "MISMATCH", "BAD")]
off = [w for w in skew if isinstance(w, dict) and w.get("status") not in ("OK", None)]
blocking = [w for w in off if str(w.get("status")).upper() in ("UNSATISFIABLE", "BAD")]
if off:
st = FAIL if blocking else WARN
summary = f"{len(blocking)} blocking" if blocking else f"{len(off)} skewed"
summary = f"{len(blocking)} blocking" if blocking else f"{len(off)} unresolved"
details = [
f"{w.get('status')}: {w.get('workspace')} pinned {w.get('pinned')} "
f"vs installed {w.get('installed')}"
f"{w.get('status')}: {w.get('workspace')} floor {w.get('floor')} "
f"vs newest {w.get('library')} release {w.get('newest_release')}"
for w in off[:8]
]
sections.append(Section("version_skew", "Version skew", st, summary, details))
elif skew:
sections.append(Section("version_skew", "Version skew", OK, "all workspaces in sync", []))
sections.append(Section("version_skew", "Version skew", OK, "all floors satisfiable", []))

# Install verification ---------------------------------------------------
vi = snapshot.get("verify_install") or {}
Expand Down
37 changes: 16 additions & 21 deletions heart/readiness.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,16 @@

- **RED** (a real blocker) if any of the 5 libraries has failing CI, is off
``main``, has uncommitted source changes, or is behind origin; or any workspace
is pinned AHEAD of its installed library, has a ``general.yaml`` ↔
``version.txt`` MISMATCH, or an unparseable (BAD) version; or the deep install
verification last reported ``ready == false``; or the release-validation
report last ingested reports ``release_ready == false`` (a stage failed).
records a compatibility floor (``version.minimum_library_version``) that
exceeds the newest released version of its library (UNSATISFIABLE — no
installable release can satisfy it), or an unparseable (BAD) floor/tag; or the
deep install verification last reported ``ready == false``; or the
release-validation report last ingested reports ``release_ready == false`` (a
stage failed).
- **YELLOW** (caution) for soft signals: workspace-validation not passing (the
workspace scripts/notebooks carry standing debt, so this is advisory — never a
hard block), script-timing regressions, stale open PRs, stale parked scripts, a
workspace pinned BEHIND, a stale or never-run install verification, **no fresh
stale or never-run install verification, **no fresh
release-validation report for the current source** (absent, stale by age, or
whose ``commit_shas`` no longer match the current ``main`` HEADs, or which ran
under a profile other than ``release``), and — crucially — any *unknown*
Expand Down Expand Up @@ -109,16 +111,14 @@
"lib_dirty": (15, 30),
"lib_behind": (20, 40),
"test_failing": (15, 15),
"skew_ahead": (25, 50),
"skew_unsatisfiable": (25, 50),
"lib_unknown": (10, 30),
"test_unknown": (10, 10),
"timing_red": (15, 15),
"timing_yellow": (8, 8),
"profiling_drift": (10, 30),
"open_pr": (5, 15),
"parked": (5, 15),
"skew_behind": (8, 24),
"skew_mismatch": (25, 50),
"skew_bad": (25, 50),
"skew_unknown": (10, 30),
"install_not_ready": (40, 40),
Expand Down Expand Up @@ -344,33 +344,28 @@ def scope_local(msg: str, key: str) -> None:
else:
scope_local("test run status unknown (no report.json)", "test_unknown")

# --- version skew (RED ahead / YELLOW behind) ---
# --- version skew (RED floor unsatisfiable / STALE unknown) ---
skew = snapshot.get("version_skew")
if isinstance(skew, dict):
for w in skew.get("workspaces") or []:
if not isinstance(w, dict):
continue
status = str(w.get("status", "")).upper()
if status == "AHEAD":
red.append(f"{w.get('workspace')}: pinned {w.get('pinned')} AHEAD of installed {w.get('installed')}")
hit("skew_ahead")
elif status == "MISMATCH":
if status == "UNSATISFIABLE":
red.append(
f"{w.get('workspace')}: general.yaml {w.get('pinned')} "
f"≠ version.txt {w.get('version_txt')}"
f"{w.get('workspace')}: floor {w.get('floor')} exceeds newest "
f"{w.get('library')} release {w.get('newest_release')} "
f"(no installable version satisfies it)"
)
hit("skew_mismatch")
hit("skew_unsatisfiable")
elif status == "BAD":
red.append(
f"{w.get('workspace')}: unparseable version "
f"(pinned {w.get('pinned')} / installed {w.get('installed')})"
f"(floor {w.get('floor')} / newest {w.get('newest_release')})"
)
hit("skew_bad")
elif status == "BEHIND":
yellow.append(f"{w.get('workspace')}: pinned BEHIND installed {w.get('installed')}")
hit("skew_behind")
elif status == "UNKNOWN":
stale.append(f"{w.get('workspace')}: installed {w.get('library')} version unknown")
stale.append(f"{w.get('workspace')}: newest {w.get('library')} release unknown")
hit("skew_unknown")

# --- manifest drift (YELLOW — identity hygiene vs PyAutoMind/repos.yaml) ---
Expand Down
Loading
Loading