""" AI Agent version freshness snapshot. Loads the latest committed, read-only version freshness snapshot for OpenClaw, Hermes, and NemoTron. This module never enables schedules, queries external registries, installs or upgrades packages, writes lockfiles, builds or pulls images, probes hosts, creates PRs, sends Telegram messages, or changes production routing. """ from __future__ import annotations import json from pathlib import Path from typing import Any from src.services.snapshot_paths import default_evaluations_dir _DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__)) _SNAPSHOT_PATTERN = "ai_agent_version_freshness_snapshot_*.json" _SCHEMA_VERSION = "ai_agent_version_freshness_snapshot_v1" _RUNTIME_AUTHORITY = "repo_only_snapshot_no_external_lookup_or_update" def load_latest_ai_agent_version_freshness_snapshot( evaluations_dir: Path | None = None, ) -> dict[str, Any]: """Load the newest committed AI Agent version freshness snapshot.""" directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR candidates = sorted(directory.glob(_SNAPSHOT_PATTERN)) if not candidates: raise FileNotFoundError( f"no AI Agent version freshness snapshots found in {directory}" ) latest = candidates[-1] with latest.open(encoding="utf-8") as handle: payload = json.load(handle) if not isinstance(payload, dict): raise ValueError(f"{latest}: expected JSON object") _require_schema(payload, _SCHEMA_VERSION, str(latest)) _require_read_only_boundaries(payload, str(latest)) _require_rollup_consistency(payload, str(latest)) _require_source_safety(payload, str(latest)) return payload def _require_schema(payload: dict[str, Any], expected: str, label: str) -> None: actual = payload.get("schema_version") if actual != expected: raise ValueError(f"{label}: expected schema_version={expected}, got {actual!r}") def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None: program_status = payload.get("program_status") or {} if program_status.get("read_only_mode") is not True: raise ValueError(f"{label}: program_status.read_only_mode must be true") if program_status.get("runtime_authority") != _RUNTIME_AUTHORITY: raise ValueError(f"{label}: runtime_authority must stay {_RUNTIME_AUTHORITY}") scan_boundaries = payload.get("scan_boundaries") or {} if scan_boundaries.get("read_only_repo_snapshot_allowed") is not True: raise ValueError(f"{label}: read_only_repo_snapshot_allowed must be true") blocked_scan_flags = { "schedule_activation_allowed", "workflow_write_allowed", "external_registry_lookup_allowed", "external_cve_lookup_allowed", "package_installation_allowed", "package_upgrade_allowed", "lockfile_write_allowed", "docker_build_allowed", "image_pull_allowed", "host_probe_allowed", "auto_pr_allowed", "auto_merge_allowed", "telegram_direct_send_allowed", "paid_external_service_allowed", "production_route_change_allowed", } enabled_scan_flags = sorted( flag for flag in blocked_scan_flags if scan_boundaries.get(flag) is not False ) if enabled_scan_flags: raise ValueError(f"{label}: scan boundaries must remain false: {enabled_scan_flags}") approval_boundaries = payload.get("approval_boundaries") or {} blocked_approval_flags = { "daily_schedule_enabled", "external_source_lookup_allowed", "package_or_model_upgrade_allowed", "host_or_k3s_probe_allowed", "telegram_digest_send_allowed", "gitea_pr_creation_allowed", "production_route_change_allowed", } enabled_approval_flags = sorted( flag for flag in blocked_approval_flags if approval_boundaries.get(flag) is not False ) if enabled_approval_flags: raise ValueError( f"{label}: approval boundaries must remain false: {enabled_approval_flags}" ) def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None: sources = payload.get("freshness_sources") or [] agents = payload.get("agent_roles") or [] steps = payload.get("daily_snapshot_design") or [] rollups = payload.get("rollups") or {} expected_counts = { "source_count": len(sources), "repo_only_source_count": sum( 1 for source in sources if source.get("source_type") == "repo_only" ), "agent_count": len(agents), "daily_step_count": len(steps), } mismatched = { key: {"expected": expected, "actual": rollups.get(key)} for key, expected in expected_counts.items() if rollups.get(key) != expected } if mismatched: raise ValueError(f"{label}: rollup counts must match payload sections: {mismatched}") action_required = sorted( source.get("source_id") for source in sources if source.get("status") == "action_required" or source.get("action_required") is True ) if sorted(rollups.get("action_required_source_ids") or []) != action_required: raise ValueError(f"{label}: rollups.action_required_source_ids mismatch") blocked_external_lookup = sorted( source.get("source_id") for source in sources if source.get("external_lookup_allowed") is False ) if sorted(rollups.get("blocked_external_lookup_source_ids") or []) != blocked_external_lookup: raise ValueError(f"{label}: rollups.blocked_external_lookup_source_ids mismatch") mutation_allowed_count = sum( 1 for source in sources if source.get("mutation_allowed") is True ) if rollups.get("mutation_allowed_source_count") != mutation_allowed_count: raise ValueError(f"{label}: rollups.mutation_allowed_source_count mismatch") if rollups.get("schedule_enabled_count") != 0: raise ValueError(f"{label}: rollups.schedule_enabled_count must remain 0") if rollups.get("telegram_direct_send_count") != 0: raise ValueError(f"{label}: rollups.telegram_direct_send_count must remain 0") def _require_source_safety(payload: dict[str, Any], label: str) -> None: unsafe_sources = [ source.get("source_id") for source in payload.get("freshness_sources") or [] if source.get("source_type") != "repo_only" or source.get("external_lookup_allowed") is not False or source.get("mutation_allowed") is not False or not source.get("evidence_refs") or not source.get("next_action") ] if unsafe_sources: raise ValueError(f"{label}: freshness sources must stay repo-only: {unsafe_sources}") unsafe_steps = [ step.get("step_id") for step in payload.get("daily_snapshot_design") or [] if step.get("allowed_now") is not True ] if unsafe_steps: raise ValueError(f"{label}: daily snapshot design steps must stay read-only: {unsafe_steps}")