Add manifest_drift tick check: continuous body-map identity drift

health_agent/capabilities.yaml

@@ -77,6 +77,11 @@ continuous_checks:
     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"
+  - 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"
+    consumes: "PyAutoMind/scripts/repos_sync.py --check (Mind owns the manifest and the drift logic; Heart only runs and reports it)"
+    gate_role: "YELLOW per drifted surface (identity hygiene, never RED); skipped when no PyAutoMind checkout"
   - id: noise
     impl: heart/noise.py
     measures: "classify git porcelain into real source drift vs regenerated-artifact noise"

heart/checks/manifest_drift.py

@@ -0,0 +1,117 @@
+"""heart/checks/manifest_drift.py — body-map identity drift.
+
+``PyAutoMind/repos.yaml`` is the single source of repo *identity* (GitHub
+home, category, role). ``PyAutoMind/scripts/repos_sync.py --check`` verifies
+the other repo lists against it — Heart's ``config/repos.yaml``, PyAutoBuild's
+``pre_build.sh``, admin_jammy's ``ensure_workspace_labels.sh``, and the actual
+``origin`` remote of every local checkout. That check only fires when someone
+remembers to run it; this module makes it continuous by running it every tick
+and parsing its report.
+
+The drift logic itself stays in ``repos_sync.py`` (never duplicated here —
+Heart observes, Mind owns the manifest). Cheap: local file parses plus one
+``git remote get-url`` per checkout, well inside the tick budget.
+
+Classification: drift is **YELLOW** — identity hygiene that will eventually
+break a `gh` call or a label sweep, not an immediate release blocker (same
+monitoring-not-gating stance as URL hygiene, but visible in readiness
+cautions). A missing PyAutoMind checkout or manifest skips the check
+(``available: false``) — normal for web/CI environments.
+
+The result lands at ``$HEART_STATE_DIR/manifest_drift.json``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+HEART_HOME = Path(__file__).resolve().parents[2]
+_p3 = Path(__file__).resolve().parents[3]
+PYAUTO_ROOT = Path(
+    os.environ.get("PYAUTO_ROOT")
+    or (_p3 if _p3.name == "PyAutoLabs" else Path.home() / "Code" / "PyAutoLabs")
+)
+HEART_STATE_DIR = Path(
+    os.environ.get("HEART_STATE_DIR")
+    or Path.home() / ".pyauto-heart"
+)
+
+_CHECK_LINE = re.compile(r"^check (?P<label>.+?): (?P<status>OK|\d+ mismatch\(es\))$")
+_PROBLEM_LINE = re.compile(r"^\s+[✗x] (?P<problem>.+)$")
+
+
+def parse_check_output(text: str) -> dict[str, dict[str, Any]]:
+    """Parse repos_sync.py --check report lines into {label: {ok, problems}}."""
+    checks: dict[str, dict[str, Any]] = {}
+    current: dict[str, Any] | None = None
+    for line in text.splitlines():
+        m = _CHECK_LINE.match(line)
+        if m:
+            current = {"ok": m.group("status") == "OK", "problems": []}
+            checks[m.group("label")] = current
+            continue
+        m = _PROBLEM_LINE.match(line)
+        if m and current is not None:
+            current["problems"].append(m.group("problem"))
+    return checks
+
+
+def run() -> dict[str, Any]:
+    script = PYAUTO_ROOT / "PyAutoMind" / "scripts" / "repos_sync.py"
+    result: dict[str, Any]
+    if not script.is_file():
+        result = {"available": False, "reason": f"missing {script}", "checks": {}}
+    else:
+        proc = subprocess.run(
+            [sys.executable, str(script), "--check", "--root", str(PYAUTO_ROOT)],
+            capture_output=True,
+            text=True,
+        )
+        checks = parse_check_output(proc.stdout)
+        if not checks:
+            # The script ran but produced nothing parseable — surface that
+            # loudly rather than reporting a hollow green.
+            result = {
+                "available": False,
+                "reason": f"unparseable output (exit {proc.returncode}): "
+                          f"{(proc.stderr or proc.stdout).strip()[:200]}",
+                "checks": {},
+            }
+        else:
+            result = {
+                "available": True,
+                "checks": checks,
+                "problem_count": sum(len(c["problems"]) for c in checks.values()),
+            }
+    sys.path.insert(0, str(HEART_HOME))
+    from heart import state
+
+    state.atomic_write_json(HEART_STATE_DIR / "manifest_drift.json", result)
+    return result
+
+
+def main(argv: list[str]) -> int:
+    result = run()
+    sys.path.insert(0, str(HEART_HOME))
+    from heart.heart_color import c_ok, c_warn, c_info, c_meta, glyph_ok, glyph_warn
+
+    if not result["available"]:
+        print(f"{glyph_warn()} {c_info('manifest_drift')} {c_meta('skipped: ' + str(result.get('reason', '')))}")
+        return 0
+    n = result["problem_count"]
+    surfaces = len(result["checks"])
+    if n:
+        print(f"{glyph_warn()} {c_info('manifest_drift')} {c_warn(f'{n} mismatch(es)')} {c_meta(f'({surfaces} surfaces vs repos.yaml)')}")
+    else:
+        print(f"{glyph_ok()} {c_info('manifest_drift')} {c_ok('identity in sync')} {c_meta(f'({surfaces} surfaces vs repos.yaml)')}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

heart/readiness.py

@@ -298,6 +298,17 @@ def hit(key: str, n: int = 1) -> None:
                 yellow.append(f"{w.get('workspace')}: installed {w.get('library')} version unknown")
                 hit("skew_unknown")
 
+    # --- manifest drift (YELLOW — identity hygiene vs PyAutoMind/repos.yaml) ---
+    manifest = snapshot.get("manifest_drift")
+    if isinstance(manifest, dict) and manifest.get("available"):
+        for label, chk in (manifest.get("checks") or {}).items():
+            if isinstance(chk, dict) and not chk.get("ok", True):
+                n = len(chk.get("problems") or [])
+                yellow.append(
+                    f"manifest drift: {label} — {n} mismatch(es) vs PyAutoMind/repos.yaml"
+                )
+                hit("manifest_drift")
+
     # --- install verification (deep check: RED on fail, YELLOW if stale/unrun) ---
     vi = snapshot.get("verify_install")
     if isinstance(vi, dict) and "ready" in vi:

heart/state.py

@@ -76,6 +76,7 @@ def aggregate() -> dict[str, Any]:
         "script_timing": _read_json_or_default(HEART_STATE_DIR / "script_timing.json", {}),
         "test_run": _read_json_or_default(HEART_STATE_DIR / "test_run.json", {}),
         "version_skew": _read_json_or_default(HEART_STATE_DIR / "version_skew.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", {}),
         "validation_report": _read_json_or_default(HEART_STATE_DIR / "validation_report.json", {}),

heart/tick.sh

@@ -35,6 +35,14 @@ PYTHONPATH="$HEART_HOME" python3 -m heart.checks.test_run || heart_log WARN "$(c
 # Python: workspace-vs-library version skew (cheap file reads; no heavy imports).
 PYTHONPATH="$HEART_HOME" python3 -m heart.checks.version_skew || heart_log WARN "$(c_warn 'version_skew failed')"
 
+# Python: body-map identity drift — PyAutoMind/repos.yaml vs Heart/Build/labels
+# repo lists and every checkout's origin (delegates to repos_sync.py --check).
+if [[ -f "$PYAUTO_ROOT/PyAutoMind/scripts/repos_sync.py" ]]; then
+  PYTHONPATH="$HEART_HOME" python3 -m heart.checks.manifest_drift || heart_log WARN "$(c_warn 'manifest_drift failed')"
+else
+  heart_log INFO "$(c_meta 'manifest_drift: skipped (no PyAutoMind checkout)')"
+fi
+
 # Aggregate into state.json.
 PYTHONPATH="$HEART_HOME" python3 -c "
 from heart import state

tests/test_manifest_drift.py

@@ -0,0 +1,84 @@
+"""tests/test_manifest_drift.py — body-map identity drift check."""
+
+from __future__ import annotations
+
+import json
+
+from heart.checks import manifest_drift as md
+
+
+OK_OUTPUT = """\
+check PyAutoHeart/config/repos.yaml: OK
+check PyAutoBuild/pre_build.sh: OK
+check ensure_workspace_labels.sh: OK
+check local checkout origins: OK
+"""
+
+DRIFT_OUTPUT = """\
+check PyAutoHeart/config/repos.yaml: OK
+check ensure_workspace_labels.sh: 2 mismatch(es)
+  ✗ ensure_workspace_labels targets 'rhayes777/PyAutoFit', manifest says 'PyAutoLabs/PyAutoFit'
+  ✗ ensure_workspace_labels targets 'rhayes777/PyAutoConf', manifest says 'PyAutoLabs/PyAutoConf'
+check local checkout origins: 1 mismatch(es)
+  ✗ 'PyAutoFit': origin is 'rhayes777/PyAutoFit', manifest says 'PyAutoLabs/PyAutoFit'
+"""
+
+
+def test_parse_all_ok():
+    checks = md.parse_check_output(OK_OUTPUT)
+    assert len(checks) == 4
+    assert all(c["ok"] for c in checks.values())
+    assert all(c["problems"] == [] for c in checks.values())
+
+
+def test_parse_drift_attributes_problems_to_surfaces():
+    checks = md.parse_check_output(DRIFT_OUTPUT)
+    assert checks["PyAutoHeart/config/repos.yaml"]["ok"] is True
+    labels = checks["ensure_workspace_labels.sh"]
+    assert labels["ok"] is False
+    assert len(labels["problems"]) == 2
+    assert "rhayes777/PyAutoFit" in labels["problems"][0]
+    origins = checks["local checkout origins"]
+    assert origins["ok"] is False
+    assert len(origins["problems"]) == 1
+
+
+def test_parse_garbage_yields_nothing():
+    assert md.parse_check_output("Traceback (most recent call last):\n  boom\n") == {}
+
+
+def _run_with_fake_script(tmp_path, monkeypatch, script_body: str | None):
+    """Run md.run() against a fake workspace root; None = no script on disk."""
+    if script_body is not None:
+        scripts = tmp_path / "PyAutoMind" / "scripts"
+        scripts.mkdir(parents=True)
+        (scripts / "repos_sync.py").write_text(script_body)
+    state_dir = tmp_path / "state"
+    state_dir.mkdir()
+    monkeypatch.setattr(md, "PYAUTO_ROOT", tmp_path)
+    monkeypatch.setattr(md, "HEART_STATE_DIR", state_dir)
+    result = md.run()
+    on_disk = json.loads((state_dir / "manifest_drift.json").read_text())
+    assert on_disk == result
+    return result
+
+
+def test_run_missing_script_is_unavailable_not_green(tmp_path, monkeypatch):
+    result = _run_with_fake_script(tmp_path, monkeypatch, None)
+    assert result["available"] is False
+    assert "missing" in result["reason"]
+
+
+def test_run_parses_fake_script_report(tmp_path, monkeypatch):
+    body = "import sys\nsys.stdout.write('''" + DRIFT_OUTPUT + "''')\nsys.exit(1)\n"
+    result = _run_with_fake_script(tmp_path, monkeypatch, body)
+    assert result["available"] is True
+    assert result["problem_count"] == 3
+    assert result["checks"]["ensure_workspace_labels.sh"]["ok"] is False
+
+
+def test_run_unparseable_output_is_unavailable(tmp_path, monkeypatch):
+    body = "import sys\nsys.stderr.write('boom')\nsys.exit(1)\n"
+    result = _run_with_fake_script(tmp_path, monkeypatch, body)
+    assert result["available"] is False
+    assert "unparseable" in result["reason"]

tests/test_readiness.py

@@ -565,3 +565,24 @@ def test_render_block_no_color_is_plain(monkeypatch):
     assert "RELEASE READINESS" in text
     assert "GREEN" in text
     assert "\033[" not in text
+
+
+def test_manifest_drift_is_yellow_not_red():
+    snap = make_snapshot(manifest_drift={
+        "available": True,
+        "problem_count": 2,
+        "checks": {
+            "ensure_workspace_labels.sh": {"ok": False, "problems": ["a", "b"]},
+            "local checkout origins": {"ok": True, "problems": []},
+        },
+    })
+    v = compute(snap)
+    assert v["verdict"] == "yellow"
+    assert v["red_reasons"] == []
+    assert any("manifest drift" in r for r in v["yellow_reasons"])
+
+
+def test_manifest_drift_unavailable_stays_green():
+    snap = make_snapshot(manifest_drift={"available": False, "reason": "missing", "checks": {}})
+    v = compute(snap)
+    assert v["verdict"] == "green"