test_run: caller-decided fetch_cloud + local/cloud agreement (#83 fix 1)

The server-first cloud verdict was unreachable from every real entrypoint:
run() inferred fetch_cloud from `results_dir is None` and main() — the tick
and the mobile path — always passed results_dir, leaving the leg on the local
test_results/latest (stale since 2026-07-09) since the design merged. main()
now passes fetch_cloud=True explicitly; bare run() defaults to no network.
When both surfaces carry a verdict, green requires both (ready = local AND
cloud), cloud_ready is always recorded, and a mismatch writes a disagreement
field — closing the audit's local-green/cloud-red hole and its mirror.
Verified against the real tick invocation: the sidecar now carries the actual
latest workspace-validation run.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: 2 people

heart/checks/test_run.py

@@ -190,10 +190,14 @@ def _from_per_job(results_dir: Path) -> dict[str, Any]:
 
 
 def run(results_dir: Path | None = None, fetch_cloud: bool | None = None) -> dict[str, Any]:
-    default_path = results_dir is None
+    # fetch_cloud is decided by the CALLER, never inferred: the old
+    # `results_dir is None` inference meant main() — the tick and the mobile
+    # path — always disabled the cloud fetch, leaving the leg on stale local
+    # evidence while claiming to be server-first (PyAutoHeart#83 finding A).
+    # Library/test callers default to no network.
     results_dir = results_dir or TEST_RESULTS_LATEST
     if fetch_cloud is None:
-        fetch_cloud = default_path  # only hit the network on the real tick path
+        fetch_cloud = False
 
     summary: dict[str, Any]
     report = _read_json(results_dir / "report.json")
@@ -208,18 +212,32 @@ def run(results_dir: Path | None = None, fetch_cloud: bool | None = None) -> dic
             report_path.stat().st_mtime, datetime.timezone.utc
         ).isoformat()
 
-    # The server workspace-validation run is the authoritative continuous verdict
-    # (server-first: MCP-supplied file, else `gh`); it sets ready/ts so a missing
-    # local report.json no longer forces "unknown" when the cloud run is green.
-    # The local report, when present, still supplies the count detail.
+    # The server workspace-validation run (MCP-supplied file, else `gh`) sets
+    # ready/ts so a missing local report.json no longer forces "unknown" when
+    # the cloud run is green. The local report, when present, still supplies
+    # the count detail. When BOTH surfaces carry a verdict they must agree to
+    # be green — a disagreement is surfaced, never silently resolved in either
+    # direction (PyAutoHeart#83 §4-§5: a fresh local pass must not green the
+    # leg while the server surface fails, and vice versa).
     cloud = _server_verdict() if fetch_cloud else None
     if cloud is not None:
+        local_ready = summary.get("ready") if summary else None
         if not summary:
             summary = {
                 "passed": 0, "failed": 0, "skipped": 0, "timeout": 0,
                 "per_project": {}, "parked_stale_count": 0, "parked_stale": [],
             }
-        summary["ready"] = cloud["ready"]
+        summary["cloud_ready"] = cloud["ready"]
+        if local_ready is None:
+            summary["ready"] = cloud["ready"]
+        elif cloud["ready"] is None:
+            summary["ready"] = local_ready  # cloud run in progress: keep local
+        else:
+            summary["ready"] = bool(local_ready) and bool(cloud["ready"])
+            if bool(local_ready) != bool(cloud["ready"]):
+                summary["disagreement"] = (
+                    f"local ready={local_ready} vs cloud ready={cloud['ready']}"
+                )
         summary["ts"] = cloud["ts"]
         summary["run_label"] = summary.get("run_label") or f"cloud#{cloud['run_id']}"
         summary["cloud_url"] = cloud["url"]
@@ -230,7 +248,9 @@ def run(results_dir: Path | None = None, fetch_cloud: bool | None = None) -> dic
 
 def main(argv: list[str]) -> int:
     results_dir = Path(argv[1]) if len(argv) > 1 else TEST_RESULTS_LATEST
-    summary = run(results_dir)
+    # The tick/CLI is the real path: always consult the server verdict here,
+    # explicitly — this is the entrypoint the old inference silently disabled.
+    summary = run(results_dir, fetch_cloud=True)
 
     sys.path.insert(0, str(HEART_HOME))
     from heart import state

tests/test_test_run.py

@@ -168,12 +168,63 @@ def test_run_writes_nothing_to_state_dir(tmp_path):
     assert after == before
 
 
-def test_main_persists_summary_to_state_dir(tmp_path):
+def test_main_persists_summary_to_state_dir(monkeypatch, tmp_path):
     """The tick path (python -m heart.checks.test_run) must still persist."""
     import os
     from pathlib import Path
+    monkeypatch.setattr(tr, "_server_verdict", lambda: None)  # hermetic
     (tmp_path / "report.json").write_text(json.dumps({
         "ready": True, "run_label": "R1", "summary": {"passed": 1}}))
     assert tr.main(["test_run", str(tmp_path)]) == 0
     written = json.loads((Path(os.environ["HEART_STATE_DIR"]) / "test_run.json").read_text())
     assert written["ready"] is True and written["passed"] == 1
+
+
+# --- entrypoint wiring + disagreement (PyAutoHeart#83 finding A) ----------------
+
+def test_main_consults_the_server_verdict(monkeypatch, tmp_path):
+    """The tick entrypoint must fetch the cloud verdict — the old inference
+    (fetch_cloud from `results_dir is None`) silently disabled it forever."""
+    import os
+    from pathlib import Path
+    monkeypatch.setattr(tr, "_server_verdict", lambda: {
+        "ready": False, "ts": "2026-07-16T00:00:00Z", "run_id": 3, "url": "U"})
+    (tmp_path / "report.json").write_text(json.dumps({
+        "ready": True, "run_label": "local", "summary": {"passed": 5}}))
+    assert tr.main(["test_run", str(tmp_path)]) == 0
+    written = json.loads((Path(os.environ["HEART_STATE_DIR"]) / "test_run.json").read_text())
+    assert written["source"] == "cloud"
+    assert written["cloud_ready"] is False
+
+
+def test_local_and_cloud_must_agree_to_be_green(monkeypatch, tmp_path):
+    # local False + cloud True → NOT green, disagreement surfaced (the mirror
+    # of the local-green/cloud-red hole; neither side wins silently).
+    (tmp_path / "report.json").write_text(json.dumps({
+        "ready": False, "run_label": "local", "summary": {"passed": 1, "failed": 2}}))
+    monkeypatch.setattr(tr, "_server_verdict", lambda: {
+        "ready": True, "ts": "t", "run_id": 4, "url": "u"})
+    out = tr.run(results_dir=tmp_path, fetch_cloud=True)
+    assert out["ready"] is False
+    assert "disagreement" in out
+    assert out["cloud_ready"] is True
+
+
+def test_cloud_in_progress_keeps_local_verdict(monkeypatch, tmp_path):
+    (tmp_path / "report.json").write_text(json.dumps({
+        "ready": True, "run_label": "local", "summary": {"passed": 1}}))
+    monkeypatch.setattr(tr, "_server_verdict", lambda: {
+        "ready": None, "ts": "t", "run_id": 5, "url": "u"})
+    out = tr.run(results_dir=tmp_path, fetch_cloud=True)
+    assert out["ready"] is True
+    assert out["cloud_ready"] is None
+    assert "disagreement" not in out
+
+
+def test_run_without_explicit_fetch_never_touches_network(monkeypatch, tmp_path):
+    calls = []
+    monkeypatch.setattr(tr, "_server_verdict", lambda: calls.append(1))
+    (tmp_path / "report.json").write_text(json.dumps({
+        "ready": True, "summary": {"passed": 1}}))
+    tr.run(results_dir=tmp_path)
+    assert calls == []