Improve ATVM watcher status metadata and run workflow

atvm/watcher-service/atvm_run_watcher.py

@@ -196,6 +196,35 @@ def run_ps() -> str:
     return proc.stdout
 
 
+def normalize_logged_command(raw: str, command_name: str) -> Optional[str]:
+    patterns = {
+        "cmc-templates.py": r"((?:python3?\s+)?(?:\./)?cmc-templates\.py\b.*)",
+        "run-sorry-cypress.py": r"((?:python3?\s+)?(?:\./)?run-sorry-cypress\.py\b.*)",
+    }
+    pattern = patterns.get(command_name)
+    if not pattern:
+        return None
+    match = re.search(pattern, raw)
+    if not match:
+        return None
+    normalized = " ".join(match.group(1).split())
+    return normalized or None
+
+
+def extract_command_from_ps(build_name: str, command_name: str) -> Optional[str]:
+    output = run_ps()
+    matches: List[str] = []
+    for line in output.splitlines():
+        if command_name not in line:
+            continue
+        if command_name == "run-sorry-cypress.py" and f"--build_name {build_name}" not in line:
+            continue
+        normalized = normalize_logged_command(line, command_name)
+        if normalized:
+            matches.append(normalized)
+    return matches[-1] if matches else None
+
+
 def process_active(build_name: str) -> bool:
     output = run_ps()
     for line in output.splitlines():
@@ -261,6 +290,20 @@ def extract_currents_url(log_text: str) -> Optional[str]:
     return match.group(1) if match else None
 
 
+def extract_command_from_log(log_text: str, command_name: str, build_name: Optional[str] = None) -> Optional[str]:
+    matches: List[str] = []
+    for line in log_text.splitlines():
+        if command_name not in line:
+            continue
+        normalized = normalize_logged_command(line, command_name)
+        if not normalized:
+            continue
+        if command_name == "run-sorry-cypress.py" and build_name and f"--build_name {build_name}" not in normalized:
+            continue
+        matches.append(normalized)
+    return matches[-1] if matches else None
+
+
 def load_state(state_file: Path) -> Dict[str, object]:
     if not state_file.exists():
         return {}
@@ -1108,7 +1151,7 @@ def infer_host_from_subrun_build(
     return remaining_hosts[0] if remaining_hosts else None
 
 
-def infer_metadata() -> Dict[str, object]:
+def infer_metadata(build_name: str, log_text: str) -> Dict[str, object]:
     try:
         extra_options = json.loads(os.environ.get("ATVM_WATCHER_EXTRA_OPTIONS", "[]"))
     except json.JSONDecodeError:
@@ -1116,9 +1159,16 @@ def infer_metadata() -> Dict[str, object]:
     if not isinstance(extra_options, list):
         extra_options = []
     extra_options = [value for value in extra_options if isinstance(value, str) and value]
+    template_command = os.environ.get("ATVM_WATCHER_TEMPLATE_COMMAND", "")
+    if not template_command:
+        template_command = extract_command_from_log(log_text, "cmc-templates.py") or extract_command_from_ps(build_name, "cmc-templates.py") or ""
+    runner_command = os.environ.get("ATVM_WATCHER_RUNNER_COMMAND", "")
+    if not runner_command:
+        runner_command = extract_command_from_log(log_text, "run-sorry-cypress.py", build_name) or extract_command_from_ps(build_name, "run-sorry-cypress.py") or ""
     return {
         "template": os.environ.get("ATVM_WATCHER_TEMPLATE", "unknown"),
-        "template_command": os.environ.get("ATVM_WATCHER_TEMPLATE_COMMAND", ""),
+        "template_command": template_command,
+        "runner_command": runner_command,
         "config_family": os.environ.get("ATVM_WATCHER_CONFIG_FAMILY", "unknown"),
         "config_file": os.environ.get("ATVM_WATCHER_CONFIG_FILE", "unknown"),
         "migration_style": os.environ.get("ATVM_WATCHER_MIGRATION_STYLE", "ATVM automation validation"),
@@ -1309,6 +1359,9 @@ def build_status_markdown(
     template_command = metadata.get("template_command")
     if isinstance(template_command, str) and template_command:
         notes = notes + [f"Template command: `{template_command}`"]
+    runner_command = metadata.get("runner_command")
+    if isinstance(runner_command, str) and runner_command:
+        notes = notes + [f"Run command: `{runner_command}`"]
     template_name = metadata.get("template")
     integration_plugin = metadata.get("integration_plugin")
     if (
@@ -1908,10 +1961,9 @@ if __name__ == "__main__":
     posted_marker = build_dir / "posted.marker"
 
     inventory = load_inventory(inventory_file)
-    metadata = infer_metadata()
-
     state = load_state(state_file)
     log_text_for_start = read_text(run_log)
+    metadata = infer_metadata(build_name, log_text_for_start)
     default_started_at = first_log_timestamp(log_text_for_start) or (datetime.fromtimestamp(run_log.stat().st_mtime, tz=timezone.utc) if run_log.exists() else now_utc())
     started_at = parse_xml_timestamp(state.get("started_at")) or default_started_at
     state.setdefault("build_name", build_name)
@@ -1927,6 +1979,11 @@ if __name__ == "__main__":
         if active:
             process_gone_since = None
         current_log_text = read_text(run_log)
+        refreshed_metadata = infer_metadata(build_name, current_log_text)
+        for key in ("template_command", "runner_command"):
+            value = refreshed_metadata.get(key)
+            if isinstance(value, str) and value and not metadata.get(key):
+                metadata[key] = value
 
         run_state, subrun_states, host_results, start_ts, end_ts, currents_url, notes = determine_state(
             build_name=build_name,