penguin: fixup patch conflict warnings

Author: lacraig2

src/penguin/common.py

@@ -3,7 +3,7 @@
 import re
 import coloredlogs
 import yaml
-from os.path import join, isfile
+from os.path import join, isfile, basename
 from yamlcore import CoreDumper, CoreLoader
 
 
@@ -54,21 +54,44 @@ def hash_yaml(section_to_hash):
     return hash_digest
 
 
-def patch_config(logger, base_config, patch):
+def patch_config(logger, base_config, patch, patch_name="patch", origin_map=None, verbose=False):
+    # Initialize origin map if it wasn't passed in
+    if origin_map is None:
+        origin_map = {}
+
+    # Helper to recursively claim ownership of keys in the origin map
+    def _record_origins(obj, path_prefix, source_name):
+        if hasattr(obj, "model_fields_set"):
+            for k in obj.model_fields_set:
+                _record_origins(getattr(obj, k), f"{path_prefix}.{k}" if path_prefix else k, source_name)
+            if obj.model_extra is not None:
+                for k, val in obj.model_extra.items():
+                    _record_origins(val, f"{path_prefix}.{k}" if path_prefix else k, source_name)
+        elif isinstance(obj, dict):
+            for k, val in obj.items():
+                _record_origins(val, f"{path_prefix}.{k}" if path_prefix else k, source_name)
+        else:
+            # Leaves and lists get recorded directly
+            origin_map[path_prefix] = source_name
+
     if not patch:
         # Empty patch, possibly an empty file or one with all comments
         return base_config
 
+    # If this is the very first run, populate the origin map with the base config
+    if not origin_map:
+        _record_origins(base_config, "", "base_config")
+
     # Merge configs.
     def _recursive_update(base, new, config_option):
         if base is None:
+            _record_origins(new, config_option, patch_name)
             return new
         if new is None:
             return base
 
-        # assert type(base) is type(new)
-
         if hasattr(base, "merge"):
+            origin_map[config_option] = patch_name
             return base.merge(new)
 
         if hasattr(base, "model_fields_set"):
@@ -80,61 +103,87 @@ def _recursive_update(base, new, config_option):
                     result[base_key] = base_value
             for new_key in new.model_fields_set:
                 new_value = getattr(new, new_key)
+                full_path = f"{config_option}.{new_key}" if config_option else new_key
                 if new_key in result:
                     result[new_key] = _recursive_update(
                         result[new_key],
                         new_value,
-                        f"{config_option}.{new_key}" if config_option else new_key,
+                        full_path,
                     )
                 else:
                     result[new_key] = new_value
+                    _record_origins(new_value, full_path, patch_name)
+
             if new.model_extra is not None:
                 for new_key, new_value in new.model_extra.items():
+                    full_path = f"{config_option}.{new_key}" if config_option else new_key
                     if new_key in result:
                         result[new_key] = _recursive_update(
                             result[new_key],
                             new_value,
-                            f"{config_option}.{new_key}" if config_option else new_key,
+                            full_path,
                         )
                     else:
                         result[new_key] = new_value
+                        _record_origins(new_value, full_path, patch_name)
             return type(base)(**result)
 
         if isinstance(base, list):
+            # We treat list appends differently, no "conflict" per se, just an addition
             return base + new
 
         if isinstance(base, dict):
             result = dict()
             for key, base_value in base.items():
+                full_path = f"{config_option}.{key}" if config_option else key
                 if key in new:
                     new_value = new[key]
                     result[key] = _recursive_update(
                         base_value,
                         new_value,
-                        f"{config_option}.{key}" if config_option else key,
+                        full_path,
                     )
                 else:
                     result[key] = base_value
             for new_key, new_value in new.items():
                 if new_key not in base:
+                    full_path = f"{config_option}.{new_key}" if config_option else new_key
                     result[new_key] = new_value
+                    _record_origins(new_value, full_path, patch_name)
             return result
 
         if base == new:
             return base
 
-        base_str = yaml.dump(base).strip().removesuffix("...").strip()
-        new_str = yaml.dump(new).strip().removesuffix("...").strip()
-        change_str = (
-            f"\n```\n{base_str}\n```↓\n```\n{new_str}\n```"
-            if "\n" in base_str + new_str
-            else f"`{base_str}` → `{new_str}`"
-        )
-        logger.warning(f"patch conflict: {config_option}: {change_str}")
+        # --> WE HAVE A CONFLICT <--
+        previous_source = origin_map.get(config_option, "base_config")
+
+        # Clean up long paths to just the filenames
+        prev_file = basename(previous_source)
+        new_file = basename(patch_name)
+
+        # Strip out Pydantic '.root' noise from the config key
+        clean_option = config_option.replace(".root", "")
+
+        if verbose:
+            base_str = yaml.dump(base).strip().removesuffix("...").strip()
+            new_str = yaml.dump(new).strip().removesuffix("...").strip()
+            change_str = (
+                f"\n```\n{base_str}\n```↓\n```\n{new_str}\n```"
+                if "\n" in base_str + new_str
+                else f"`{base_str}` → `{new_str}`"
+            )
+
+            # Use a much tighter logging format
+            logger.info(
+                f"conflict: {clean_option}: {change_str} ({prev_file} -> {new_file})"
+            )
 
+        # Claim ownership of the newly overwritten key
+        origin_map[config_option] = patch_name
         return new
 
-    return _recursive_update(base_config, patch, None)
+    return _recursive_update(base_config, patch, "")
 
 
 class PathHighlightingFormatter(coloredlogs.ColoredFormatter):

src/penguin/manager.py

@@ -20,7 +20,6 @@
 import signal
 import subprocess
 import time
-import logging
 from penguin import getColoredLogger
 from .common import yaml
 
@@ -74,7 +73,7 @@ def calculate_score(result_dir: str, have_console: bool = True) -> dict[str, int
     # --- Load core_config.yaml ---
     try:
         with open(os.path.join(result_dir, "core_config.yaml")) as f:
-            config = yaml.safe_load(f) or {} # Ensure config is a dict even if file is empty
+            config = yaml.safe_load(f) or {}  # Ensure config is a dict even if file is empty
     except FileNotFoundError:
         logger.warning(f"Config file not found in {result_dir}. Cannot determine blocked signals.")
     except yaml.YAMLError as e:
@@ -83,14 +82,13 @@ def calculate_score(result_dir: str, have_console: bool = True) -> dict[str, int
     # --- System Health: execs, sockets, devices ---
     try:
         with open(os.path.join(result_dir, "health_final.yaml")) as f:
-            health_data = yaml.safe_load(f) or {} # Ensure health_data is a dict
+            health_data = yaml.safe_load(f) or {}  # Ensure health_data is a dict
     except FileNotFoundError:
         # Instead of returning {}, just log a warning and continue. Scores will default to 0.
         logger.warning(f"{result_dir}/health_final.yaml not found - health scores will be 0.")
     except yaml.YAMLError as e:
         logger.error(f"Error parsing health_final.yaml in {result_dir}: {e}")
 
-
     # --- Panic or not (nopanic) ---
     console_log_path = os.path.join(result_dir, "console.log")
     if have_console and os.path.isfile(console_log_path):
@@ -134,7 +132,6 @@ def calculate_score(result_dir: str, have_console: bool = True) -> dict[str, int
         except (IOError, csv.Error) as e:
             logger.error(f"Could not read or parse {coverage_csv_path}: {e}")
 
-
     if config:
         blocked_signals = -len(config.get("blocked_signals", []))
 

src/penguin/penguin_config/__init__.py

@@ -191,11 +191,15 @@ def load_unpatched_config(path):
     return config
 
 
-def load_config(proj_dir, path, validate=True, resolved_kernel=None):
+def load_config(proj_dir, path, validate=True, resolved_kernel=None, verbose=False):
     """Load penguin config from path"""
     with open(path, "r") as f:
         config = yaml.load(f, Loader=CoreLoader)
     config = structure.Patch(**config)
+
+    # 1. Initialize the empty map to track our provenance
+    origin_map = {}
+
     # look for files called patch_*.yaml in the same directory as the config file
     if config.core.auto_patching:
         patch_files = list(Path(proj_dir).glob("patch_*.yaml"))
@@ -213,12 +217,20 @@ def load_config(proj_dir, path, validate=True, resolved_kernel=None):
             # patches are loaded relative to the main config file
             patch_relocated = Path(proj_dir, patch)
             if patch_relocated.exists():
-                # TODO: If we're missing a patch we should warn, but this happens 3-4x
-                # and that's too verbose.
                 with open(patch_relocated, "r") as f:
-                    patch = yaml.load(f, Loader=CoreLoader)
-                patch = structure.Patch(**patch)
-                config = patch_config(logger, config, patch)
+                    patch_data = yaml.load(f, Loader=CoreLoader)
+                patch_data = structure.Patch(**patch_data)
+
+                # 2. Pass the origin map and the patch name down into the merger
+                config = patch_config(
+                    logger=logger,
+                    base_config=config,
+                    patch=patch_data,
+                    patch_name=str(patch_relocated),  # Give it a name to log
+                    origin_map=origin_map,           # Pass the state map
+                    verbose=verbose
+                )
+
     config = config.model_dump()
     if config["core"].get("guest_cmd", False) is True:
         config["static_files"]["/igloo/utils/guesthopper"] = dict(
@@ -249,7 +261,8 @@ def load_config(proj_dir, path, validate=True, resolved_kernel=None):
             if Path(proj_dir, "base/fs.tar.gz").exists():
                 config["core"]["fs"] = "./base/fs.tar.gz"
             else:
-                logger.info("No core.fs specified in config - using empty fs - most likely a test")
+                if verbose:
+                    logger.info("No core.fs specified in config - using empty fs - most likely a test")
                 config["core"]["fs"] = "./base/empty_fs.tar.gz"
                 empty_fs_path = os.path.join(proj_dir, "./base/empty_fs.tar.gz")
                 if not os.path.exists(empty_fs_path):

src/penguin/penguin_run.py

@@ -110,9 +110,9 @@ def run_config(
     # Read input config and validate
     if resolved_kernel:
         logger.info(f"Using pre-resolved kernel: {resolved_kernel}")
-        conf = load_config(proj_dir, conf_yaml, resolved_kernel=resolved_kernel)
+        conf = load_config(proj_dir, conf_yaml, resolved_kernel=resolved_kernel, verbose=True)
     else:
-        conf = load_config(proj_dir, conf_yaml)
+        conf = load_config(proj_dir, conf_yaml, verbose=True)
 
     pkversion = get_penguin_kernel_version(conf)