fix: publish complete local monitoring evidence

docs/internals.md

@@ -163,3 +163,15 @@ Old adverse results remain adverse. Undated results require a collector refresh.
 Published local observations carry their full inventory and per-repository
 checkout evidence. Private records retain their status and an opaque ID with a
 local investigation action; private paths never travel to the public board.
+
+
+### Published monitoring fallback
+
+The dev-box publisher exports every observed monitoring family, including
+manifest/workflow drift, URL checks and PyPI floors, even where the older board
+has no section. Each family carries its complete public monitoring inventory
+and original collector timestamp. Private findings remain unresolved placeholders.
+Cloud observations take precedence for families the cloud actually measured;
+otherwise published findings fill the gap. Missing or malformed inventories
+remain unknown, and neither aggregate time nor publication time refreshes old
+or undated evidence. This transport does not change release readiness.

heart/dashboard.py

@@ -1099,7 +1099,7 @@ def build_board(
             sections.append(Section("url_check", "URL hygiene", OK,
                                     f"{len(uc['repos'])} repos clean (swept {uc.get('ts', '?')})", []))
 
-    sections = _devbox_enrich(sections, devbox, now)
+    sections = _devbox_enrich(sections, devbox, now, snapshot=snapshot)
     for section in sections:
         if section.key == "release_validation" and section.state in (FAIL, WARN):
             section.action = {"label": "inspect release validation", "kind": "prompt",
@@ -1131,6 +1131,18 @@ def build_board(
     )
     from heart import monitoring
     board.monitoring = monitoring.assess(board, snapshot, devbox=devbox, now=now)
+    if isinstance(devbox, dict):
+        for sec in board.sections:
+            family = monitoring.ALIASES.get(sec.key, sec.key)
+            rows = [c for c in board.monitoring["checks"]
+                    if c["family"] == family and c["source"] == f"devbox.sections.{family}"]
+            public_section = (devbox.get("sections") or {}).get(family)
+            if rows and isinstance(public_section, dict) and isinstance(public_section.get("monitoring_checks"), list):
+                worst = min(rows, key=lambda c: monitoring.ORDER[c["status"]])
+                sec.state = {"green": OK, "yellow": WARN, "red": FAIL}.get(worst["status"], UNOBS)
+                if worst["status"] not in {"green", "na"}:
+                    sec.summary = worst["summary"]
+                sec.action = worst.get("action") or sec.action
     board.fix_plan = build_fix_plan(board, snapshot, devbox=devbox)
     return board
 
@@ -1749,7 +1761,8 @@ def _structure_reasons(red: list, yellow: list, stales: list, repos: dict,
 
 
 def _devbox_enrich(
-    sections: list[Section], devbox: dict | None, now: datetime.datetime | None
+    sections: list[Section], devbox: dict | None, now: datetime.datetime | None,
+    *, snapshot: dict | None = None
 ) -> list[Section]:
     """Fill unobserved rows from a published dev-box observation.
 
@@ -1761,32 +1774,53 @@ def _devbox_enrich(
     if not isinstance(devbox, dict):
         return sections
     dsecs = devbox.get("sections") or {}
+    from heart import monitoring
+    if not isinstance(dsecs, dict):
+        return sections
+    if snapshot is not None:
+        existing = {monitoring.ALIASES.get(sec.key, sec.key) for sec in sections}
+        for family in dsecs:
+            if (family in monitoring.FAMILIES and family not in existing
+                    and isinstance(dsecs[family], dict)
+                    and monitoring.published_checks(family, dsecs[family])[1]):
+                sections.append(Section(family, monitoring.FAMILIES[family], UNOBS, "No observation here", []))
+        for sec in sections:
+            family = monitoring.ALIASES.get(sec.key, sec.key)
+            if family in dsecs and not monitoring.family_data(snapshot, family):
+                sec.state = UNOBS
     age = _age_seconds(devbox.get("ts"), now)
     if age is None or not isinstance(dsecs, dict):
         return sections
     ago = format_age(age)
     out: list[Section] = []
     for sec in sections:
-        d = dsecs.get(sec.key)
+        d = dsecs.get(monitoring.ALIASES.get(sec.key, sec.key))
         if sec.state != UNOBS or not isinstance(d, dict):
             out.append(sec)
             continue
-        if age <= DEVBOX_FRESH_SECONDS and d.get("state") in (OK, WARN, FAIL, INFO):
+        observation = d.get("observed_at", devbox.get("ts"))
+        age = _age_seconds(observation, now)
+        ago = format_age(age)
+        if age is not None and 0 <= age <= DEVBOX_FRESH_SECONDS and d.get("state") in (OK, WARN, FAIL, INFO):
             out.append(Section(
                 sec.key, sec.title, str(d["state"]),
                 str(d.get("summary") or ""),
                 [str(x) for x in (d.get("details") or [])][:8],
                 links=sec.links, action=sec.action,
                 observed_ago=f"observed {ago} on the dev box",
                 entries=[{**e, "original_source": e.get("source"), "source": "published dev-box observation",
-                          "observed_at": devbox.get("ts")} for e in d.get("entries", [])
+                          "observed_at": e.get("observed_at", observation)} for e in d.get("entries", [])
                          if isinstance(e, dict)],
             ))
         else:
             out.append(Section(
                 sec.key, sec.title, UNOBS,
                 f"not observed here — dev box last looked {ago}",
                 sec.details, links=sec.links, action=sec.action,
+                entries=[{**e, "original_source": e.get("source"),
+                          "source": "published dev-box observation",
+                          "observed_at": e.get("observed_at", observation)}
+                         for e in d.get("entries", []) if isinstance(e, dict)],
             ))
     return out
 

heart/monitoring.py

@@ -91,6 +91,30 @@ def expected_repos():
     return [r["name"] for rows in config["repos"].values() for r in rows]
 
 
+def family_data(snapshot, family):
+    """Whether this vantage actually carries a family's observation."""
+    data = snapshot.get(family)
+    if not data and family in {"import_time", "unit_test_timing"}:
+        data = snapshot.get("unit_timings")
+    return data
+
+
+def published_checks(family, section):
+    """Validate public rows; malformed inventory never establishes coverage."""
+    stored = section.get("monitoring_checks")
+    if not isinstance(stored, list):
+        return [], False
+    rows = [item for item in stored if isinstance(item, dict)
+            and isinstance(item.get("id"), str) and item["id"]
+            and item.get("family", family) == family
+            and item.get("status") in ORDER
+            and isinstance(item.get("summary"), str)
+            and isinstance(item.get("subject"), str)]
+    complete = (len(rows) == len(stored)
+                and any(item["id"] == f"{family}:coverage" for item in rows))
+    return rows, complete
+
+
 def assess(board, snapshot, *, devbox=None, now=None, repos=None, families=None):
     """Return one uncapped inventory used by score, repair and all consumers.
 
@@ -200,9 +224,20 @@ def walk(family, value, path, inherited_ts):
         if family in {"import_time", "unit_test_timing"} and not ts and not (isinstance(data, dict) and "ts" in data):
             ts = (snapshot.get("unit_timings") if isinstance(snapshot.get("unit_timings"), dict) else {}).get("ts")
         ds = (devbox.get("sections") or {}).get(family)
-        published = sec is not None and sec.observed_ago and isinstance(ds, dict)
-        if published:
-            ts = devbox.get("ts")
+        # The publication envelope timestamp is not an observation timestamp.
+        # Consume the uncapped inventory even when no legacy section exists.
+        if not family_data(snapshot, family) and isinstance(ds, dict):
+            rows, complete = published_checks(family, ds)
+            for item in rows:
+                add(family, item["subject"], item["status"], item["summary"],
+                    item.get("evidence"), observed_at=item.get("observed_at"),
+                    action=item.get("action"), identity=item["id"],
+                    source=f"devbox.sections.{family}", na_reason=item.get("applicability_reason"))
+            if not complete:
+                add(family, FAMILIES.get(family, family), "grey",
+                    "Published summary lacks the complete inventory; republish from the dev box",
+                    identity=f"{family}:coverage", source=f"devbox.sections.{family}")
+            continue
         label = FAMILIES.get(family, family)
         state = status(sec.state) if sec else "grey"
         summary = sec.summary if sec else "No observation available"
@@ -223,11 +258,10 @@ def walk(family, value, path, inherited_ts):
             totals = data.get("totals") or {}
             if totals.get("permanent") and not totals.get("slow") and not totals.get("needs_fix"):
                 na_reason = "Only permanent exclusions by design; no repairable skips"
-        if not published and family in {"script_timing", "unit_test_timing", "workspace_testmode_timing"} and isinstance(data, dict):
+        if family in {"script_timing", "unit_test_timing", "workspace_testmode_timing"} and isinstance(data, dict):
             if not any(data.get(k) for k in ("green_count", "red_count", "yellow_count")):
                 state, summary = "grey", "No comparisons measured; collect timings and establish baselines"
-        add(family, label, state, summary, source=f"devbox.sections.{family}" if published else None,
-            observed_at=ts or (devbox.get("ts") if published else None),
+        add(family, label, state, summary, observed_at=ts,
             action=sec.action if sec and state not in {"grey", "stale"} else None,
             identity=f"{family}:coverage", na_reason=na_reason)
         if isinstance(data, dict):
@@ -238,16 +272,6 @@ def walk(family, value, path, inherited_ts):
                 add(family, "orphaned baseline details", "grey", "Collector omitted orphaned baseline details; refresh the complete collector", observed_at=ts)
         if data:
             walk(family, data, family, ts)
-        if published:
-            stored = ds.get("monitoring_checks")
-            if isinstance(stored, list):
-                for item in stored:
-                    if isinstance(item, dict):
-                        add(family, item.get("subject", label), item.get("status"), item.get("summary", ""),
-                            item.get("evidence"), observed_at=item.get("observed_at") or ts,
-                            action=item.get("action"), identity=item.get("id"), source=f"devbox.sections.{family}", na_reason=item.get("applicability_reason"))
-            else:
-                add(family, label + " full detail", "grey", "Published summary lacks the complete inventory; republish from the dev box", observed_at=ts)
     # Raw unit evidence carries individual rows omitted from legacy summaries.
     if isinstance(snapshot.get("unit_timings"), dict) and snapshot["unit_timings"]:
         walk("unit_timings", snapshot["unit_timings"], "unit_timings", snapshot["unit_timings"].get("ts"))

heart/publish.py

@@ -34,13 +34,11 @@
 HEART_ROOT = Path(__file__).resolve().parents[1]
 DEVBOX_FILE = HEART_ROOT / "state" / "devbox_board.json"
 
-# Only the families the cloud job cannot observe travel; everything else the
-# cloud measures itself, and merging two vantages of the same family would
-# break the unify invariant. repo_state is excluded: it folds into per-repo
-# rows, not a section of its own.
-PUBLISH_FAMILIES = tuple(
-    f for f in dashboard.LOCAL_ONLY_FAMILIES if f != "repo_state"
-)
+# Publish observed families as fallback evidence. Cloud observations take
+# precedence; absent families must still be visible on the published board.
+from heart import monitoring
+
+PUBLISH_FAMILIES = tuple(monitoring.FAMILIES)
 
 DEVBOX_SCHEMA_VERSION = 1
 
@@ -72,32 +70,32 @@ def _public_monitoring(item: dict) -> dict:
             "applicability_reason": item.get("applicability_reason")}
 
 
-def build_devbox_board(snapshot: dict | None, verdict: dict | None) -> dict[str, Any]:
-    """Distill the LOCAL board's local-only families. Pure; never raises."""
-    board = dashboard.build_board(snapshot, verdict, unobserved=())
+def build_devbox_board(snapshot: dict | None, verdict: dict | None, *, now=None) -> dict[str, Any]:
+    """Export complete public inventories, retaining the collector timestamps."""
+    board = dashboard.build_board(snapshot, verdict, unobserved=(), now=now)
     sections: dict[str, Any] = {}
-    for sec in board.sections:
-        if sec.key not in PUBLISH_FAMILIES:
+    rendered = {monitoring.ALIASES.get(s.key, s.key): s for s in board.sections}
+    for family in PUBLISH_FAMILIES:
+        if not monitoring.family_data(snapshot or {}, family):
+            continue
+        checks = [_public_monitoring(item) for item in board.monitoring["checks"]
+                  if item["family"] == family]
+        if not checks:
             continue
-        if sec.state == dashboard.UNOBS:
-            continue  # nothing observed locally either — publish no claim
-        sections[sec.key] = {
-            "state": sec.state,
-            "summary": sec.summary,
-            "details": _scrub(sec.details)[:8],
+        coverage = next((c for c in checks if c["id"] == f"{family}:coverage"), {})
+        worst = min(checks, key=lambda c: monitoring.ORDER[c["status"]])
+        sec = rendered.get(family)
+        sections[family] = {
+            "state": {"green": "ok", "yellow": "warn", "red": "fail"}.get(worst["status"], "unobserved"),
+            "summary": _scrub([coverage.get("summary", "")]) or ["Inspect published monitoring evidence"],
+            "observed_at": coverage.get("observed_at"),
+            "details": _scrub(sec.details)[:8] if sec else [],
+            "monitoring_checks": checks,
         }
-        if board.monitoring:
-            sections[sec.key]["monitoring_checks"] = [
-                _public_monitoring(item) for item in board.monitoring["checks"]
-                if item["family"] == sec.key
-            ]
-        if sec.entries:
-            # Structured observations obey the same public-path boundary as
-            # plain detail lines, including paths inside prompts/evidence.
-            sections[sec.key]["entries"] = [
-                entry for entry in sec.entries
-                if _scrub([json.dumps(entry, ensure_ascii=False)])
-            ]
+        sections[family]["summary"] = sections[family]["summary"][0]
+        if sec and sec.entries:
+            sections[family]["entries"] = [entry for entry in sec.entries
+                if _scrub([json.dumps(entry, ensure_ascii=False)])]
     return {
         "schema_version": DEVBOX_SCHEMA_VERSION,
         "ts": (snapshot or {}).get("ts") or "",

tests/test_monitoring.py

@@ -173,3 +173,19 @@ def test_source_run_timestamp_survives_reaggregation(tmp_path):
     os.utime(path, (old, old))
     rows = script_timing.scan_latest_results(tmp_path)
     assert rows[0]["observed_at"].startswith("2026-09-01")
+
+
+@pytest.mark.parametrize("stored", [None, [], [None], [{"id": "manifest_drift:coverage", "status": "green"}]])
+def test_malformed_published_inventory_cannot_establish_coverage(stored):
+    m = monitoring.assess(board(), {}, devbox={"ts": TS, "sections": {
+        "manifest_drift": {"state": "ok", "monitoring_checks": stored}}},
+        now=NOW, families=["manifest_drift"], repos=[])
+    assert not m["complete"] and m["score"] < 100
+
+
+def test_published_item_without_timestamp_cannot_inherit_envelope_time():
+    item = {"id": "manifest_drift:coverage", "subject": "manifest", "status": "green", "summary": "clean"}
+    m = monitoring.assess(board(), {}, devbox={"ts": TS, "sections": {
+        "manifest_drift": {"monitoring_checks": [item]}}}, now=NOW, families=["manifest_drift"], repos=[])
+    assert m["findings"][0]["status"] == "grey"
+    assert m["findings"][0]["observed_at"] is None