from __future__ import annotations import hashlib import json from typing import Any from src.models import Configuration DISPLAY_ONLY_FIELDS = {"configuration_label"} SET_LIKE_FIELDS = {"skills", "tools"} def _normalize_value(value: Any, key: str | None = None) -> Any: if isinstance(value, dict): normalized: dict[str, Any] = {} for child_key in sorted(value): if child_key in DISPLAY_ONLY_FIELDS: continue child = _normalize_value(value[child_key], child_key) if child is not None: normalized[child_key] = child return normalized if isinstance(value, list): items = [_normalize_value(item) for item in value] if key in SET_LIKE_FIELDS: # Skills/tools describe membership, not declaration order. return sorted(dict.fromkeys(items)) return items return value def normalize_configuration(configuration: Configuration | dict[str, Any] | None) -> dict[str, Any]: """Return a canonical structured representation suitable for comparison. Missing values stay missing and display-only labels are excluded. Skills and tools are treated as set-like fields so equivalent ordering produces the same identity. """ if configuration is None: return {} if isinstance(configuration, Configuration): raw = configuration.model_dump(exclude_none=True) else: raw = dict(configuration) return _normalize_value(raw) def configuration_fingerprint(configuration: Configuration | dict[str, Any] | None) -> str | None: """Return a deterministic SHA-256 identity for a non-empty configuration.""" normalized = normalize_configuration(configuration) if not normalized: return None payload = json.dumps(normalized, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def configuration_label(configuration: Configuration | None) -> str: if configuration is None: return "Configuration metadata unavailable" if configuration.configuration_label: return configuration.configuration_label parts: list[str] = [] if configuration.shell_access is True: parts.append("shell") if configuration.planning_enabled is True: parts.append("planning") if configuration.context_compression and configuration.context_compression.enabled is True: method = configuration.context_compression.method parts.append(f"compression:{method}" if method else "compression") if configuration.memory and configuration.memory.enabled is True: method = configuration.memory.method parts.append(f"memory:{method}" if method else "memory") if configuration.reasoning_effort: parts.append(f"reasoning:{configuration.reasoning_effort}") if configuration.max_context_tokens is not None: parts.append(f"context:{configuration.max_context_tokens}") if configuration.skills: parts.append("skills:" + ",".join(sorted(set(configuration.skills)))) if configuration.tools: parts.append("tools:" + ",".join(sorted(set(configuration.tools)))) return " + ".join(parts) if parts else "Structured configuration" def _flatten(value: Any, prefix: str = "") -> dict[str, Any]: if not isinstance(value, dict): return {prefix: value} flattened: dict[str, Any] = {} for key, child in value.items(): path = f"{prefix}.{key}" if prefix else key if isinstance(child, dict): flattened.update(_flatten(child, path)) else: flattened[path] = child return flattened def changed_configuration_fields( baseline: Configuration | dict[str, Any] | None, treatment: Configuration | dict[str, Any] | None, ) -> list[str]: """Return stable dotted paths whose known structured values differ.""" left = _flatten(normalize_configuration(baseline)) right = _flatten(normalize_configuration(treatment)) keys = sorted(set(left) | set(right)) return [key for key in keys if left.get(key) != right.get(key)]