| 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: |
| |
| 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)] |
|
|
