Fix categorized ATVM watcher host result recovery

atvm/watcher-service/atvm_run_watcher.py

@@ -607,6 +607,78 @@ def collect_latest_host_reporter_artifact(
     return latest
 
 
+def parse_host_reporter_json(artifact_path: Path, host: str, kernels: Dict[str, str]) -> Optional[HostResult]:
+    try:
+        payload = json.loads(artifact_path.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+
+    stats = payload.get("stats")
+    metadata = payload.get("metadata")
+    if not isinstance(stats, dict):
+        return None
+
+    tests = int(stats.get("tests", 0) or 0)
+    failures = int(stats.get("failures", 0) or 0)
+    pending = int(stats.get("pending", 0) or 0)
+    duration_ms = stats.get("duration", 0) or 0
+    timestamp = None
+    if isinstance(metadata, dict):
+        timestamp = parse_reporter_metadata_timestamp(metadata.get("timestamp"))
+    if timestamp is None:
+        timestamp = reporter_artifact_run_timestamp(artifact_path)
+
+    detail_parts = [f"{tests} tests", f"{failures} failures"]
+    if pending:
+        detail_parts.append(f"{pending} pending")
+
+    if failures:
+        status = "FAIL"
+    elif pending:
+        status = "RUN"
+    else:
+        status = "PASS"
+
+    return HostResult(
+        host=host,
+        kernel=kernels.get(host, "unknown"),
+        status=status,
+        detail=", ".join(detail_parts),
+        tests=tests,
+        failures=failures,
+        duration_seconds=float(duration_ms) / 1000 if duration_ms else None,
+        timestamp=timestamp,
+    )
+
+
+def parse_host_reporter_artifact(artifact_path: Path, host: str, kernels: Dict[str, str]) -> Optional[HostResult]:
+    if artifact_path.suffix == ".json":
+        return parse_host_reporter_json(artifact_path, host, kernels)
+
+    artifact_ts = reporter_artifact_run_timestamp(artifact_path)
+    if artifact_ts is None:
+        artifact_ts = datetime.fromtimestamp(artifact_path.stat().st_mtime, tz=timezone.utc)
+
+    try:
+        text = artifact_path.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        text = ""
+
+    failures = 1 if re.search(r"\bTimed out!\b|\bAssertionError\b|\bFAIL\b|\berror\b", text, re.I) else 0
+    status = "FAIL" if failures else "PASS"
+    detail = "1 failures" if failures else "completed"
+    return HostResult(
+        host=host,
+        kernel=kernels.get(host, "unknown"),
+        status=status,
+        detail=detail,
+        failures=failures,
+        timestamp=artifact_ts,
+    )
+
+
 def collect_group_host_reporter_artifacts(
     reporter_root: Path,
     group_label: Optional[str],
@@ -632,7 +704,7 @@ def collect_group_host_reporter_artifacts(
             continue
 
         latest_artifact_mtime: Optional[datetime] = None
-        latest_run_ts: Optional[datetime] = None
+        latest_result: Optional[HostResult] = None
         for artifact_path in sorted(host_dir.iterdir()):
             if artifact_path.suffix not in {".txt", ".json"}:
                 continue
@@ -641,20 +713,24 @@ def collect_group_host_reporter_artifacts(
                 continue
             if run_ended_at and artifact_mtime >= run_ended_at:
                 continue
-            if latest_artifact_mtime is None or artifact_mtime >= latest_artifact_mtime:
-                latest_artifact_mtime = artifact_mtime
-                latest_run_ts = reporter_artifact_run_timestamp(artifact_path) or artifact_mtime
 
-        if latest_artifact_mtime is None:
+            parsed_result = parse_host_reporter_artifact(artifact_path, host, kernels)
+            if parsed_result is None:
+                continue
+
+            # Prefer the newest artifact, and prefer JSON over TXT when timestamps tie.
+            if latest_artifact_mtime is None or artifact_mtime > latest_artifact_mtime or (
+                artifact_mtime == latest_artifact_mtime and artifact_path.suffix == ".json"
+            ):
+                latest_artifact_mtime = artifact_mtime
+                latest_result = parsed_result
+
+        if latest_artifact_mtime is None or latest_result is None:
             continue
 
-        results[host] = HostResult(
-            host=host,
-            kernel=kernels.get(host, "unknown"),
-            status="PASS",
-            detail="completed",
-            timestamp=latest_run_ts or latest_artifact_mtime,
-        )
+        if latest_result.timestamp is None:
+            latest_result.timestamp = latest_artifact_mtime
+        results[host] = latest_result
     return results