"""squash/deprecation_watch.py — Track C / C8 — Model Deprecation Watch. Sprint 35 (W265–W266). OpenAI / Anthropic / Google / Meta / Mistral sunset models quarterly. Every sunset breaks a version-tied Annex IV record and invalidates the attestation. Most teams discover deprecations the day inference returns a 404. Squash deprecation-watch cross-references the Asset Registry against a maintained deprecation schedule, fires configurable-lead-time alerts, and generates a per-model migration effort estimate + re-attestation checklist. Real deprecations (from provider announcements, updated in the built-in feed): OpenAI gpt-4-0613 deprecated Sep 2024; text-davinci-003 retired Jan 2024; gpt-3.5-turbo-0613 deprecated Sep 2024; DALL-E 2 deprecating May 2024 Anthropic claude-1.* retired Nov 2024; claude-instant-1.* retired Nov 2024 Google PaLM 2 deprecated Oct 2024; Gemini 1.0 Pro deprecated Jul 2025 Meta Llama 1 EOL (no first-party cloud API support) Mistral mistral-tiny deprecated Jul 2024 Architecture ------------ DeprecationImpact — BREAKING | SOFT | INFORMATIONAL MigrationEffort — LOW | MEDIUM | HIGH | CRITICAL DeprecationEntry — per-model record with sunset_date, aliases, successor DeprecationAlert — asset matched against entry, with days_remaining + migration_effort + re_attestation_checklist DeprecationStore — SQLite cache (default: ~/.squash/deprecation_cache.db) DeprecationWatcher — main engine: load_feeds(), scan(), check_model() W265: provider feeds + SQLite cache + cross-reference engine W266: migration effort estimator + re-attestation checklist + alert routing Usage ----- :: watcher = DeprecationWatcher() # Scan entire asset registry alerts = watcher.scan() for alert in alerts: print(alert.summary()) # Check a specific model alert = watcher.check_model("gpt-4-0613") # List all known deprecations all_entries = watcher.list_entries() """ from __future__ import annotations import datetime import json import sqlite3 import urllib.error import urllib.request from dataclasses import asdict, dataclass, field from enum import Enum from pathlib import Path from typing import Any VERSION = "0.1.0" _DEFAULT_DB = Path.home() / ".squash" / "deprecation_cache.db" # ── Enumerations ────────────────────────────────────────────────────────────── class DeprecationImpact(str, Enum): BREAKING = "BREAKING" # API calls will fail / model removed SOFT = "SOFT" # degraded / redirected to successor INFORMATIONAL = "INFORMATIONAL" # deprecated but still accessible class MigrationEffort(str, Enum): LOW = "LOW" # drop-in replacement; re-attestation only MEDIUM = "MEDIUM" # minor prompt/param changes + re-attestation HIGH = "HIGH" # significant refactor + full re-attestation CRITICAL = "CRITICAL" # architecture change; high-risk prod migration # ── Data model ──────────────────────────────────────────────────────────────── @dataclass class DeprecationEntry: """A single model's deprecation announcement.""" provider: str # openai | anthropic | google | meta | mistral | other model_id: str # canonical ID (e.g. gpt-4-0613) aliases: list[str] # alternative IDs that resolve to same model sunset_date: str # ISO-8601 date (YYYY-MM-DD) or "" if unknown announced_date: str # when the deprecation was announced impact: DeprecationImpact successor_model: str # recommended replacement (may be empty) migration_url: str # provider docs link notes: str still_accessible: bool = True # True until hard-removed @property def days_until_sunset(self) -> int | None: """Days until sunset from today. None if no sunset_date.""" if not self.sunset_date: return None try: sunset = datetime.date.fromisoformat(self.sunset_date) return (sunset - datetime.date.today()).days except ValueError: return None @property def is_sunsetted(self) -> bool: days = self.days_until_sunset if days is None: return not self.still_accessible return days <= 0 def matches(self, model_id: str) -> bool: """Case-insensitive match against model_id or any alias. Matching rules (in priority order): 1. Exact match (case-insensitive). 2. Segment-prefix match: the shorter string must end at a hyphen boundary in the longer string, e.g. "gpt-4" matches "gpt-4-0613" (next char is "-") but NOT "gpt-4o" (next char is "o"). """ needle = model_id.strip().lower() candidates = [self.model_id.lower()] + [a.lower() for a in self.aliases] if needle in candidates: return True for c in candidates: # needle is a prefix of c (registered ID is shorter, entry has more specifics) if c.startswith(needle) and ( len(c) == len(needle) or c[len(needle)] == "-" ): return True # c is a prefix of needle (entry ID is shorter — alias for a family) if needle.startswith(c) and ( len(needle) == len(c) or needle[len(c)] == "-" ): return True return False def to_dict(self) -> dict[str, Any]: return { "provider": self.provider, "model_id": self.model_id, "aliases": self.aliases, "sunset_date": self.sunset_date, "announced_date": self.announced_date, "impact": self.impact.value, "successor_model": self.successor_model, "migration_url": self.migration_url, "notes": self.notes, "still_accessible": self.still_accessible, "days_until_sunset": self.days_until_sunset, "is_sunsetted": self.is_sunsetted, } @dataclass class DeprecationAlert: """Asset × deprecation entry match with contextual analysis.""" asset_model_id: str asset_id: str environment: str risk_tier: str frameworks: list[str] entry: DeprecationEntry days_remaining: int | None # None if no sunset date migration_effort: MigrationEffort migration_effort_rationale: str re_attestation_checklist: list[str] notified_at: str # ISO-8601 UTC def is_urgent(self, lead_time_days: int = 30) -> bool: if self.days_remaining is None: return self.entry.is_sunsetted return self.days_remaining <= lead_time_days def summary(self, lead_time_days: int = 30) -> str: days_str = (f"{self.days_remaining}d" if self.days_remaining is not None else "already sunsetted") urgent = "🔴 URGENT" if self.is_urgent(lead_time_days) else "🟡" return ( f"{urgent} {self.asset_model_id} [{self.environment}] — " f"{self.entry.provider} sunset in {days_str} → " f"migrate to {self.entry.successor_model or 'no successor specified'} " f"(effort: {self.migration_effort.value})" ) def to_dict(self) -> dict[str, Any]: return { "asset_model_id": self.asset_model_id, "asset_id": self.asset_id, "environment": self.environment, "risk_tier": self.risk_tier, "frameworks": self.frameworks, "entry": self.entry.to_dict(), "days_remaining": self.days_remaining, "migration_effort": self.migration_effort.value, "migration_effort_rationale": self.migration_effort_rationale, "re_attestation_checklist": self.re_attestation_checklist, "notified_at": self.notified_at, "is_urgent": self.is_urgent(), "squash_version": VERSION, } # ── Built-in deprecation feed ───────────────────────────────────────────────── # Real announced deprecations from provider release notes and deprecation pages. # Updated as part of squash version releases. Providers covered: 5. _BUILTIN_FEED: list[dict[str, Any]] = [ # ── OpenAI ──────────────────────────────────────────────────────────────── { "provider": "openai", "model_id": "gpt-4-0613", "aliases": ["gpt-4"], "sunset_date": "2025-06-30", "announced_date": "2024-09-05", "impact": "BREAKING", "successor_model": "gpt-4o", "migration_url": "https://platform.openai.com/docs/deprecations", "notes": "gpt-4-0613 deprecated September 2024. Migrate to gpt-4o.", "still_accessible": True, }, { "provider": "openai", "model_id": "gpt-3.5-turbo-0613", "aliases": ["gpt-3.5-turbo-16k-0613"], "sunset_date": "2025-06-30", "announced_date": "2024-09-05", "impact": "BREAKING", "successor_model": "gpt-3.5-turbo", "migration_url": "https://platform.openai.com/docs/deprecations", "notes": "Snapshot models deprecated September 2024.", "still_accessible": True, }, { "provider": "openai", "model_id": "text-davinci-003", "aliases": ["text-davinci-002", "text-curie-001", "text-babbage-001", "text-ada-001", "davinci", "curie", "babbage", "ada"], "sunset_date": "2024-01-04", "announced_date": "2023-07-06", "impact": "BREAKING", "successor_model": "gpt-3.5-turbo-instruct", "migration_url": "https://platform.openai.com/docs/deprecations", "notes": "GPT-3 base models retired January 4, 2024.", "still_accessible": False, }, { "provider": "openai", "model_id": "gpt-4-32k", "aliases": ["gpt-4-32k-0613"], "sunset_date": "2025-06-30", "announced_date": "2024-09-05", "impact": "BREAKING", "successor_model": "gpt-4o", "migration_url": "https://platform.openai.com/docs/deprecations", "notes": "GPT-4-32k deprecated. Use gpt-4o which has 128k context.", "still_accessible": True, }, { "provider": "openai", "model_id": "gpt-4-vision-preview", "aliases": ["gpt-4-turbo-preview"], "sunset_date": "2025-04-30", "announced_date": "2024-11-06", "impact": "SOFT", "successor_model": "gpt-4o", "migration_url": "https://platform.openai.com/docs/deprecations", "notes": "Vision preview and turbo preview redirected to gpt-4o.", "still_accessible": True, }, { "provider": "openai", "model_id": "dall-e-2", "aliases": ["dalle-2"], "sunset_date": "2024-12-31", "announced_date": "2024-04-01", "impact": "BREAKING", "successor_model": "dall-e-3", "migration_url": "https://platform.openai.com/docs/deprecations", "notes": "DALL-E 2 deprecated. Migrate to DALL-E 3.", "still_accessible": True, }, { "provider": "openai", "model_id": "whisper-1", "aliases": [], "sunset_date": "", "announced_date": "", "impact": "INFORMATIONAL", "successor_model": "", "migration_url": "https://platform.openai.com/docs/models", "notes": "No announced deprecation. Monitor provider page.", "still_accessible": True, }, # ── Anthropic ───────────────────────────────────────────────────────────── { "provider": "anthropic", "model_id": "claude-1", "aliases": ["claude-1.0", "claude-1.2", "claude-v1"], "sunset_date": "2024-11-06", "announced_date": "2024-09-04", "impact": "BREAKING", "successor_model": "claude-3-haiku-20240307", "migration_url": "https://docs.anthropic.com/en/docs/resources/model-deprecations", "notes": "Claude 1 series retired November 2024.", "still_accessible": False, }, { "provider": "anthropic", "model_id": "claude-instant-1", "aliases": ["claude-instant-1.1", "claude-instant-1.2", "claude-instant-v1"], "sunset_date": "2024-11-06", "announced_date": "2024-09-04", "impact": "BREAKING", "successor_model": "claude-3-haiku-20240307", "migration_url": "https://docs.anthropic.com/en/docs/resources/model-deprecations", "notes": "Claude Instant series retired November 2024.", "still_accessible": False, }, { "provider": "anthropic", "model_id": "claude-2", "aliases": ["claude-2.0", "claude-2.1"], "sunset_date": "2025-07-01", "announced_date": "2025-01-21", "impact": "BREAKING", "successor_model": "claude-3-5-haiku-20241022", "migration_url": "https://docs.anthropic.com/en/docs/resources/model-deprecations", "notes": "Claude 2 series deprecating mid-2025. Migrate to claude-3.5.", "still_accessible": True, }, { "provider": "anthropic", "model_id": "claude-3-opus-20240229", "aliases": ["claude-3-opus"], "sunset_date": "2025-07-31", "announced_date": "2025-04-01", "impact": "SOFT", "successor_model": "claude-opus-4-5", "migration_url": "https://docs.anthropic.com/en/docs/resources/model-deprecations", "notes": "claude-3-opus-20240229 being superseded by claude-opus-4 series.", "still_accessible": True, }, # ── Google ──────────────────────────────────────────────────────────────── { "provider": "google", "model_id": "chat-bison", "aliases": ["chat-bison-001", "chat-bison@001", "text-bison", "text-bison-001"], "sunset_date": "2024-10-09", "announced_date": "2024-04-09", "impact": "BREAKING", "successor_model": "gemini-1.5-flash", "migration_url": "https://cloud.google.com/vertex-ai/generative-ai/docs/legacy/legacy-models", "notes": "PaLM 2 chat and text models retired October 2024.", "still_accessible": False, }, { "provider": "google", "model_id": "gemini-1.0-pro", "aliases": ["gemini-1.0-pro-001", "gemini-1.0-pro-latest"], "sunset_date": "2025-04-09", "announced_date": "2025-01-06", "impact": "BREAKING", "successor_model": "gemini-1.5-pro", "migration_url": "https://ai.google.dev/gemini-api/docs/models/gemini", "notes": "Gemini 1.0 Pro deprecated April 2025. Migrate to Gemini 1.5.", "still_accessible": True, }, { "provider": "google", "model_id": "embedding-gecko", "aliases": ["textembedding-gecko", "textembedding-gecko@001", "textembedding-gecko@003"], "sunset_date": "2025-07-31", "announced_date": "2025-01-01", "impact": "BREAKING", "successor_model": "text-embedding-004", "migration_url": "https://cloud.google.com/vertex-ai/generative-ai/docs/embeddings", "notes": "Gecko embedding models being retired. Migrate to text-embedding-004.", "still_accessible": True, }, # ── Meta ───────────────────────────────────────────────────────────────── { "provider": "meta", "model_id": "llama-1", "aliases": ["llama-7b", "llama-13b", "llama-30b", "llama-65b", "meta/llama-1"], "sunset_date": "2024-01-01", "announced_date": "2023-07-18", "impact": "SOFT", "successor_model": "llama-3", "migration_url": "https://llama.meta.com", "notes": "Llama 1 weights still downloadable but no active cloud API support. " "Superseded by Llama 2 then Llama 3.", "still_accessible": True, }, { "provider": "meta", "model_id": "llama-2", "aliases": ["llama-2-7b", "llama-2-13b", "llama-2-70b", "llama-2-7b-chat", "llama-2-13b-chat", "llama-2-70b-chat", "meta/llama-2"], "sunset_date": "2026-01-01", "announced_date": "2024-04-18", "impact": "SOFT", "successor_model": "llama-3", "migration_url": "https://llama.meta.com", "notes": "Llama 2 superseded by Llama 3. Community-hosted weights remain " "available but Llama 3 is recommended for new deployments.", "still_accessible": True, }, # ── Mistral ─────────────────────────────────────────────────────────────── { "provider": "mistral", "model_id": "mistral-tiny", "aliases": ["mistral-tiny-2312"], "sunset_date": "2024-09-30", "announced_date": "2024-07-24", "impact": "BREAKING", "successor_model": "mistral-small", "migration_url": "https://docs.mistral.ai/getting-started/models/", "notes": "mistral-tiny deprecated July 2024. Migrate to mistral-small.", "still_accessible": False, }, { "provider": "mistral", "model_id": "mistral-small", "aliases": ["mistral-small-2312"], "sunset_date": "2025-09-30", "announced_date": "2024-07-24", "impact": "SOFT", "successor_model": "mistral-small-latest", "migration_url": "https://docs.mistral.ai/getting-started/models/", "notes": "Dated snapshot mistral-small-2312 superseded by mistral-small-latest.", "still_accessible": True, }, { "provider": "mistral", "model_id": "open-mistral-7b", "aliases": ["mistral-7b-v0.1", "mistral-7b-v0.2"], "sunset_date": "", "announced_date": "", "impact": "INFORMATIONAL", "successor_model": "open-mistral-nemo", "migration_url": "https://docs.mistral.ai/getting-started/models/", "notes": "Still available; open-mistral-nemo recommended for new deployments.", "still_accessible": True, }, ] # ── SQLite persistence ──────────────────────────────────────────────────────── class DeprecationStore: """SQLite cache for deprecation entries and scan history.""" def __init__(self, db_path: Path | None = None) -> None: self._db_path = Path(db_path) if db_path else _DEFAULT_DB self._db_path.parent.mkdir(parents=True, exist_ok=True) self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) self._init_schema() def _init_schema(self) -> None: self._conn.executescript(""" CREATE TABLE IF NOT EXISTS deprecation_entries ( model_id TEXT PRIMARY KEY, provider TEXT NOT NULL, payload TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS scan_history ( scan_id TEXT PRIMARY KEY, scanned_at TEXT NOT NULL, alert_count INTEGER NOT NULL, payload TEXT NOT NULL ); """) self._conn.commit() def upsert_entry(self, entry: DeprecationEntry) -> None: now = _utc_now() self._conn.execute( "INSERT OR REPLACE INTO deprecation_entries (model_id, provider, payload, updated_at) " "VALUES (?, ?, ?, ?)", (entry.model_id, entry.provider, json.dumps(entry.to_dict()), now), ) self._conn.commit() def get_all(self) -> list[DeprecationEntry]: rows = self._conn.execute( "SELECT payload FROM deprecation_entries ORDER BY provider, model_id" ).fetchall() return [_entry_from_dict(json.loads(r[0])) for r in rows] def save_scan(self, scan_id: str, alerts: list[DeprecationAlert]) -> None: payload = json.dumps([a.to_dict() for a in alerts]) self._conn.execute( "INSERT OR REPLACE INTO scan_history (scan_id, scanned_at, alert_count, payload) " "VALUES (?, ?, ?, ?)", (scan_id, _utc_now(), len(alerts), payload), ) self._conn.commit() def close(self) -> None: self._conn.close() def __enter__(self) -> "DeprecationStore": return self def __exit__(self, *_: object) -> None: self.close() # ── Main engine ─────────────────────────────────────────────────────────────── class DeprecationWatcher: """Cross-reference engine for model deprecation awareness. W265: provider feed loading + SQLite cache + cross-reference against asset_registry.AssetRegistry. W266: migration effort estimator + re-attestation checklist + alert routing. """ def __init__( self, store: DeprecationStore | None = None, db_path: Path | None = None, fetch_live: bool = False, ) -> None: self._store = store or DeprecationStore(db_path=db_path) self._fetch_live = fetch_live self._entries: list[DeprecationEntry] = [] self._loaded = False # ── Feed loading (W265) ─────────────────────────────────────────────────── def load_feeds( self, providers: list[str] | None = None, include_informational: bool = True, ) -> list[DeprecationEntry]: """Load and cache deprecation entries from built-in + optional live feed. Parameters ---------- providers : Filter by provider names (default: all 5). include_informational: Include INFORMATIONAL entries (no sunset date / already accessible but recommended to migrate). """ entries = [_entry_from_dict(d) for d in _BUILTIN_FEED] if self._fetch_live: entries = _merge_live(entries) if providers: lower_providers = {p.lower() for p in providers} entries = [e for e in entries if e.provider.lower() in lower_providers] if not include_informational: entries = [e for e in entries if e.impact != DeprecationImpact.INFORMATIONAL] # Persist to SQLite cache for entry in entries: self._store.upsert_entry(entry) self._entries = entries self._loaded = True return entries def list_entries( self, providers: list[str] | None = None, include_informational: bool = True, include_sunsetted: bool = True, ) -> list[DeprecationEntry]: """Return all known deprecation entries, with optional filters.""" if not self._loaded: self.load_feeds(providers=providers, include_informational=include_informational) entries = self._entries if not include_sunsetted: entries = [e for e in entries if not e.is_sunsetted] return entries # ── Cross-reference engine (W265) ───────────────────────────────────────── def scan( self, lead_time_days: int = 30, providers: list[str] | None = None, registry_db: Path | None = None, model_ids: list[str] | None = None, ) -> list[DeprecationAlert]: """Cross-reference asset registry against deprecation entries. Parameters ---------- lead_time_days : Surface alerts when sunset ≤ this many days away. providers : Restrict to these provider feeds. registry_db : Path to asset_registry.db (default: ~/.squash/asset_registry.db). model_ids : Explicit list of model IDs to check (bypasses registry). """ if not self._loaded: self.load_feeds(providers=providers) # Build synthetic asset list from model_ids or pull from AssetRegistry assets: list[dict[str, Any]] = [] if model_ids: for mid in model_ids: assets.append({ "model_id": mid, "asset_id": mid, "environment": "unknown", "risk_tier": "UNCLASSIFIED", "frameworks": [], }) else: assets = _load_registry_assets(registry_db) alerts: list[DeprecationAlert] = [] for asset in assets: for entry in self._entries: if not entry.matches(asset["model_id"]): continue days = entry.days_until_sunset # Include: sunsetted models AND models within lead_time window if days is not None and days > lead_time_days: continue # too far out; skip alert = _build_alert(asset, entry, days) alerts.append(alert) import uuid scan_id = f"scan-{uuid.uuid4().hex[:12]}" self._store.save_scan(scan_id, alerts) return alerts def check_model( self, model_id: str, providers: list[str] | None = None, ) -> DeprecationAlert | None: """Return a DeprecationAlert if the model is deprecated, else None.""" if not self._loaded: self.load_feeds(providers=providers) for entry in self._entries: if entry.matches(model_id): return _build_alert( {"model_id": model_id, "asset_id": model_id, "environment": "unknown", "risk_tier": "UNCLASSIFIED", "frameworks": []}, entry, entry.days_until_sunset, ) return None def close(self) -> None: self._store.close() def __enter__(self) -> "DeprecationWatcher": return self def __exit__(self, *_: object) -> None: self.close() # ── Migration effort estimator (W266) ──────────────────────────────────────── def estimate_migration_effort( entry: DeprecationEntry, environment: str = "unknown", risk_tier: str = "UNCLASSIFIED", frameworks: list[str] | None = None, ) -> tuple[MigrationEffort, str]: """Heuristic migration effort rating. Factors (in priority order): 1. Impact BREAKING + production environment + high-risk tier → CRITICAL 2. Impact BREAKING + production → HIGH 3. Impact BREAKING + successor change architecture → HIGH 4. SOFT impact + production → MEDIUM 5. Any deprecated + no successor → HIGH (unknown migration path) 6. INFORMATIONAL / staging / dev → LOW """ env_lower = environment.lower() is_prod = env_lower in ("production", "prod", "live") is_high_risk = risk_tier.upper() in ("HIGH", "CRITICAL") no_successor = not entry.successor_model if no_successor and entry.impact == DeprecationImpact.BREAKING: return (MigrationEffort.HIGH, f"BREAKING deprecation with no announced successor for {entry.model_id}. " "Manual assessment required.") if entry.impact == DeprecationImpact.BREAKING and is_prod and is_high_risk: return (MigrationEffort.CRITICAL, f"BREAKING deprecation in production with high-risk tier. " f"Migrate {entry.model_id} → {entry.successor_model} immediately.") if entry.impact == DeprecationImpact.BREAKING and is_prod: return (MigrationEffort.HIGH, f"BREAKING deprecation in production environment. " f"Plan migration from {entry.model_id} → {entry.successor_model}.") if entry.impact == DeprecationImpact.BREAKING: return (MigrationEffort.MEDIUM, f"BREAKING deprecation in {environment} environment. " f"Migrate to {entry.successor_model} before production promotion.") if entry.impact == DeprecationImpact.SOFT and is_prod: return (MigrationEffort.MEDIUM, f"Soft deprecation in production. Model redirected to " f"{entry.successor_model}; re-validate outputs and re-attest.") return (MigrationEffort.LOW, f"Informational deprecation or non-production environment. " f"Plan migration to {entry.successor_model} at next cycle.") def build_reAttestation_checklist( entry: DeprecationEntry, frameworks: list[str] | None = None, ) -> list[str]: """Generate a squash-specific re-attestation checklist for a deprecation. W266: each item maps to a squash command or compliance artefact that must be re-run after migrating to the successor model. """ fws = [f.lower() for f in (frameworks or [])] successor = entry.successor_model or "" items: list[str] = [ f"[ ] Update model_id from `{entry.model_id}` to `{successor}` in deployment config", f"[ ] Run `squash attest ./{successor} --policy eu-ai-act` on the successor model", "[ ] Verify output distribution against baseline: check for behavioural drift", f"[ ] Publish new attestation: `squash publish` → update att:// URI in all CI gates", ] if "eu-ai-act" in fws or not fws: items.append( "[ ] Regenerate Annex IV documentation: " f"`squash annex-iv generate --root ./{successor}` — model_id and architecture sections change" ) if "iso-42001" in fws or "iso42001" in fws: items.append( "[ ] Re-run ISO 42001 gap analysis: `squash iso42001 gap-analysis` — " "verify control mapping holds for successor" ) items += [ "[ ] Re-run bias audit on successor model outputs (demographic parity may differ)", f"[ ] Update model_card: `squash model-card --model-id {successor}`", f"[ ] Close open approval request or re-request approval for new version: " "`squash request-approval --attestation att://...-new`", f"[ ] Update VEX feed subscriptions to track CVEs for {successor}", ] if entry.migration_url: items.append(f"[ ] Review provider migration guide: {entry.migration_url}") return items # ── Alert routing (W266) ────────────────────────────────────────────────────── def route_alerts( alerts: list[DeprecationAlert], channel: str = "stdout", lead_time_days: int = 30, ) -> None: """Route alerts to configured channel. Channels: stdout | slack | json Slack routing delegates to squash/notifications.py (requires SQUASH_SLACK_WEBHOOK_URL). """ if not alerts: return if channel == "json": import sys print(json.dumps([a.to_dict() for a in alerts], indent=2), file=__import__("sys").stdout) return if channel == "slack": try: from squash.notifications import notify for alert in alerts: notify( "model.deprecation_warning", model_id=alert.asset_model_id, details={ "provider": alert.entry.provider, "days_remaining": alert.days_remaining, "successor": alert.entry.successor_model, "effort": alert.migration_effort.value, }, ) except Exception: # noqa: BLE001 — notifications best-effort pass return # Default: stdout for alert in alerts: print(alert.summary(lead_time_days)) # ── Internal helpers ────────────────────────────────────────────────────────── def _entry_from_dict(d: dict[str, Any]) -> DeprecationEntry: return DeprecationEntry( provider=d["provider"], model_id=d["model_id"], aliases=d.get("aliases", []), sunset_date=d.get("sunset_date", ""), announced_date=d.get("announced_date", ""), impact=DeprecationImpact(d.get("impact", "INFORMATIONAL")), successor_model=d.get("successor_model", ""), migration_url=d.get("migration_url", ""), notes=d.get("notes", ""), still_accessible=d.get("still_accessible", True), ) def _build_alert( asset: dict[str, Any], entry: DeprecationEntry, days: int | None, ) -> DeprecationAlert: effort, rationale = estimate_migration_effort( entry, environment=asset.get("environment", "unknown"), risk_tier=asset.get("risk_tier", "UNCLASSIFIED"), frameworks=asset.get("frameworks", []), ) checklist = build_reAttestation_checklist( entry, frameworks=asset.get("frameworks", []) ) return DeprecationAlert( asset_model_id=asset["model_id"], asset_id=asset.get("asset_id", asset["model_id"]), environment=asset.get("environment", "unknown"), risk_tier=asset.get("risk_tier", "UNCLASSIFIED"), frameworks=asset.get("frameworks", []), entry=entry, days_remaining=days, migration_effort=effort, migration_effort_rationale=rationale, re_attestation_checklist=checklist, notified_at=_utc_now(), ) def _load_registry_assets(registry_db: Path | None = None) -> list[dict[str, Any]]: """Pull assets from AssetRegistry, fail gracefully if unavailable.""" try: from squash.asset_registry import AssetRegistry registry = AssetRegistry(db_path=registry_db) assets = [] for record in registry.list_assets(): assets.append({ "model_id": record.model_id, "asset_id": record.asset_id, "environment": record.environment.value, "risk_tier": record.risk_tier.value, "frameworks": record.frameworks, }) registry.close() return assets except Exception: # noqa: BLE001 — registry may not be initialised return [] def _merge_live(entries: list[DeprecationEntry]) -> list[DeprecationEntry]: """Attempt to fetch a live deprecation JSON from a squash-maintained URL. Fails silently and returns the built-in list on any network error. The live feed URL is a placeholder — in production this would point to a squash-maintained CDN endpoint refreshed from provider changelogs. """ LIVE_URL = "https://static.squash.works/deprecation-feed/v1.json" try: with urllib.request.urlopen(LIVE_URL, timeout=3) as resp: raw = json.loads(resp.read().decode("utf-8")) live = [_entry_from_dict(d) for d in raw.get("entries", [])] # Merge: live entries override built-in on model_id existing = {e.model_id: e for e in entries} for le in live: existing[le.model_id] = le return list(existing.values()) except (urllib.error.URLError, OSError, json.JSONDecodeError, KeyError): return entries def _utc_now() -> str: return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")