atvm: preserve authoritative host results in watcher artifacts

Prefer failing and terminal host results when merging supplemental
subrun artifacts so a later partial artifact cannot overwrite a more
authoritative outcome.

Also point ATVM agent guidance at the ATVM-specific git guide and add
that guide to document draft-only commit handling and the default SSH
command pattern.

atvm/watcher-service/atvm_run_watcher.py

@@ -1668,6 +1668,39 @@ def merge_categorized_state(
         existing["state"] = state
 
 
+def merge_host_results_prefer_authoritative(
+    base_results: Dict[str, HostResult],
+    supplemental_results: Dict[str, HostResult],
+) -> Dict[str, HostResult]:
+    merged = dict(base_results)
+    for host, candidate in supplemental_results.items():
+        existing = merged.get(host)
+        if existing is None:
+            merged[host] = candidate
+            continue
+
+        # Never let a supplemental artifact overwrite a known failure with PASS.
+        if existing.failures and not candidate.failures:
+            continue
+        if candidate.failures and not existing.failures:
+            merged[host] = candidate
+            continue
+
+        existing_terminal = existing.status in {"PASS", "FAIL"}
+        candidate_terminal = candidate.status in {"PASS", "FAIL"}
+        if existing_terminal and not candidate_terminal:
+            continue
+        if candidate_terminal and not existing_terminal:
+            merged[host] = candidate
+            continue
+
+        existing_ts = existing.timestamp or datetime.fromtimestamp(0, tz=timezone.utc)
+        candidate_ts = candidate.timestamp or datetime.fromtimestamp(0, tz=timezone.utc)
+        if candidate_ts >= existing_ts:
+            merged[host] = candidate
+    return merged
+
+
 def extract_segment_build_name(segment_text: str, parent_build_name: str) -> Optional[str]:
     patterns = [
         rf"({re.escape(parent_build_name)}-[A-Za-z0-9_.-]*batch\d+_\d+)",
@@ -1856,9 +1889,10 @@ def discover_categorized_subruns(
                     run_ended_at=check_ts + timedelta(seconds=5),
                 )
                 if group_host_results:
-                    merged_results = dict(host_results)
-                    merged_results.update(group_host_results)
-                    host_results = merged_results
+                    host_results = merge_host_results_prefer_authoritative(
+                        host_results,
+                        group_host_results,
+                    )
                     completed_hosts.extend([host for host in host_results if host not in completed_hosts])
 
             if not host_results and check_ts: