"""Audit public API compatibility against a previous release tag. This is a release gate, not a replacement for unit tests. It compares the runtime public surface in this checkout against a baseline git ref (by default the latest reachable stable release tag; pre-releases are skipped) and reports unapproved removals, call-signature changes, or return-annotation changes. Usage: uv run python scripts/audit_public_api_compat.py uv run python scripts/audit_public_api_compat.py --baseline-ref v0.4.1 uv run python scripts/audit_public_api_compat.py --json uv run python scripts/audit_public_api_compat.py --check-stale ``--check-stale`` additionally fails when an ``allowed_breaks`` entry matches no current break against the baseline (it is already in the baseline). This keeps the allowlist from silently accumulating cruft — prune such entries at each release boundary (see ``docs/releasing.md`` → prune-allowlist-at-release). Exit codes: 0 No unapproved compatibility breaks (and, with --check-stale, none stale). 1 Unapproved public API breakage detected, or stale allowlist entries under --check-stale. 2 Script/setup error, bad baseline ref, or import/introspection failure. """ from __future__ import annotations import argparse import json import os import re import subprocess import sys import tarfile import tempfile import textwrap from dataclasses import asdict, dataclass from fnmatch import fnmatchcase from pathlib import Path from typing import Any PUBLIC_PACKAGE = "notebooklm" DEFAULT_ALLOWLIST = "scripts/api-compat-allowlist.json" EXCLUDED_TOP_LEVEL_MODULES = {"__main__", "notebooklm_cli"} EXTRA_PUBLIC_PACKAGES = ("rpc",) CLIENT_NAMESPACE_ATTRIBUTES = ( "artifacts", "chat", "labels", "mind_maps", "notes", "notebooks", "research", "settings", "sharing", "sources", ) @dataclass(frozen=True) class ApiBreak: """A backward-incompatible public surface change.""" code: str object: str detail: str @property def key(self) -> str: return f"{self.code}:{self.object}" @dataclass(frozen=True) class Allowance: """A reviewed compatibility break that is allowed for this release.""" code: str object: str reason: str def matches(self, breakage: ApiBreak) -> bool: # These are fnmatch globs, so "*" can cross dots. Keep release # allowlists exact unless a broad match is intentional. return fnmatchcase(breakage.code, self.code) and fnmatchcase( breakage.object, self.object, ) def _run_git(args: list[str], cwd: Path, *, capture: bool = True) -> subprocess.CompletedProcess: return subprocess.run( ["git", *args], cwd=cwd, capture_output=capture, text=False, check=False, ) def latest_release_tag(repo_root: Path) -> str: """Return the latest reachable stable release tag. Restricts ``git describe`` to release-shaped tags (``--match``) and drops pre-release suffixes (``--exclude`` of aN/bN/rcN), so a pushed ``v0.8.0a1`` does not become the compat baseline. Keeping the baseline on the last stable release means the audit checks the real ``vPREV -> vNEXT`` upgrade path for the whole pre-release cycle, and the allowlist prunes once at the final tag. """ result = _run_git( [ "describe", "--tags", "--abbrev=0", "--match", "v[0-9]*.[0-9]*.[0-9]*", # release-shaped tags only (skips recovery/*, docs-*, …) "--exclude", "*a[0-9]*", # drop aN pre-releases "--exclude", "*b[0-9]*", # drop bN pre-releases "--exclude", "*rc[0-9]*", # drop rcN pre-releases ], repo_root, ) if result.returncode != 0: stderr = result.stderr.decode("utf-8", errors="replace") raise RuntimeError( "could not resolve latest stable release tag: " f"{stderr.strip()}. Fetch tags/history or pass --baseline-ref explicitly." ) return result.stdout.decode("utf-8").strip() def export_git_ref(repo_root: Path, ref: str, destination: Path) -> Path: """Extract ``ref`` into ``destination`` and return the extracted repo path.""" result = _run_git(["archive", "--format=tar", ref], repo_root) if result.returncode != 0: stderr = result.stderr.decode("utf-8", errors="replace") raise RuntimeError(f"could not archive baseline ref {ref!r}: {stderr.strip()}") destination.mkdir(parents=True, exist_ok=True) archive_path = destination / "baseline.tar" archive_path.write_bytes(result.stdout) source_root = destination / "baseline" source_root.mkdir() with tarfile.open(archive_path) as archive: archive.extractall(source_root, filter="data") return source_root _COLLECTOR = r""" from __future__ import annotations import dataclasses import enum import importlib import inspect import json import pathlib import sys import typing import warnings ROOT = pathlib.Path(sys.argv[1]).resolve() EXTRA_PUBLIC_NAMES = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {} CLIENT_NAMESPACE_ATTRIBUTES = set(json.loads(sys.argv[3])) if len(sys.argv) > 3 else set() PKG = sys.argv[4] EXCLUDED = set(json.loads(sys.argv[5])) EXTRA_PACKAGES = tuple(json.loads(sys.argv[6])) # Enforce the "every public module declares __all__" rule only for the CURRENT # checkout. Historical baselines (e.g. v0.4.1) predate the rule and legitimately # lack __all__ on some public modules; raising there would abort the baseline # collection before any diff runs (issue #1493 review). ENFORCE_ALL = (sys.argv[7] == "1") if len(sys.argv) > 7 else True def discover_modules() -> list[str]: package_dir = ROOT / "src" / PKG modules = {PKG} if package_dir.is_dir(): for path in package_dir.glob("*.py"): stem = path.stem if stem.startswith("_") or stem in EXCLUDED: continue modules.add(f"{PKG}.{stem}") for name in EXTRA_PACKAGES: if (package_dir / name / "__init__.py").is_file(): modules.add(f"{PKG}.{name}") return sorted(modules) class _ReturnProbe: # Tiny carrier so typing.get_type_hints can resolve a lone return string. def __init__(self, annotation): self.__annotations__ = {"return": annotation} def annotation_repr(annotation, obj=None): if annotation is inspect.Signature.empty: return None # `from __future__ import annotations` (PEP 563) yields string annotations # already; non-postponed modules yield live objects. Resolve string # annotations against the owning module's globals so the captured form is # canonical regardless of a module's PEP 563 status (a transition would # otherwise flip e.g. 'MindMap' <-> 'notebooklm.types.MindMap' and surface a # spurious changed-return). Fall back to the raw string when resolution # fails (e.g. TYPE_CHECKING-only names). if isinstance(annotation, str): module_name = getattr(obj, "__module__", None) module = sys.modules.get(module_name) if module_name else None if module is None: return annotation try: annotation = typing.get_type_hints( _ReturnProbe(annotation), globalns=vars(module) )["return"] except Exception: return annotation return inspect.formatannotation(annotation) def signature_payload(obj): try: sig = inspect.signature(obj) except (TypeError, ValueError): return None params = [] for param in sig.parameters.values(): params.append( { "name": param.name, "kind": param.kind.name, "has_default": param.default is not inspect.Parameter.empty, "default_repr": None if param.default is inspect.Parameter.empty else repr(param.default), } ) return { "text": str(sig), "parameters": params, "return_annotation": annotation_repr(sig.return_annotation, obj), } def kind_of(obj) -> str: if inspect.isclass(obj): if issubclass(obj, enum.Enum): return "enum" return "class" if inspect.isfunction(obj) or inspect.ismethod(obj) or inspect.iscoroutinefunction(obj): return "function" if inspect.ismodule(obj): return "module" return type(obj).__name__ def unwrap_member(obj): if isinstance(obj, (staticmethod, classmethod)): return obj.__func__ if isinstance(obj, property): return obj.fget return obj def member_kind(obj) -> str: if isinstance(obj, property): return "property" unwrapped = unwrap_member(obj) if inspect.isfunction(unwrapped) or inspect.ismethod(unwrapped): return "method" if inspect.isclass(unwrapped): return "class" return type(obj).__name__ def collect_class(cls) -> dict: payload = { "kind": kind_of(cls), "signature": signature_payload(cls), "members": {}, "enum_members": {}, } enum_member_names = set(cls.__members__) if issubclass(cls, enum.Enum) else set() if issubclass(cls, enum.Enum): payload["enum_members"] = {name: member.value for name, member in cls.__members__.items()} if dataclasses.is_dataclass(cls): for field in dataclasses.fields(cls): payload["members"][field.name] = { "kind": "dataclass-field", "signature": None, } for base in reversed(cls.__mro__): if base is object: continue if not getattr(base, "__module__", "").startswith(PKG): continue for name, raw in vars(base).items(): if name.startswith("_"): continue if name in enum_member_names: continue if name in payload["members"]: continue target = unwrap_member(raw) payload["members"][name] = { "kind": member_kind(raw), "signature": signature_payload(target), } if cls.__module__ == f"{PKG}.client" and cls.__name__ == "NotebookLMClient": from notebooklm.auth import AuthTokens instance = cls( AuthTokens( cookies={"SID": "compat-audit"}, csrf_token="compat-audit", session_id="compat-audit", ) ) for name in vars(instance): if not name.startswith("_"): payload["members"].setdefault( name, { "kind": "instance-attribute", "signature": None, }, ) if name not in CLIENT_NAMESPACE_ATTRIBUTES: continue subclient = getattr(instance, name) for base in reversed(type(subclient).__mro__): if base is object: continue if not getattr(base, "__module__", "").startswith(PKG): continue for child_name, raw in vars(base).items(): if child_name.startswith("_"): continue target = unwrap_member(raw) payload["members"][f"{name}.{child_name}"] = { "kind": member_kind(raw), "signature": signature_payload(target), } return payload def collect_module(module_name: str) -> dict: module = importlib.import_module(module_name) has_all = hasattr(module, "__all__") if not has_all and ENFORCE_ALL: # Every discovered public top-level module MUST declare ``__all__`` so a # brand-new public module cannot ship un-baselined (its surface would # otherwise be invisible to this audit). The presence flag was captured # historically but never enforced; enforce it now (issue #1493) — but # only for the current checkout (ENFORCE_ALL), never for an older # baseline that predates the rule. raise RuntimeError( f"public module {module_name!r} must declare __all__ " "(every public top-level module defines its exported surface so " "the compat audit can baseline it)" ) all_names = list(getattr(module, "__all__", [])) extra_names = list(EXTRA_PUBLIC_NAMES.get(module_name, [])) names = [] for name in [*all_names, *extra_names]: if name not in names: names.append(name) payload = {"exports": {}, "has_all": has_all} for name in names: try: value = getattr(module, name) except AttributeError: if name in extra_names and name not in all_names: continue raise entry = { "kind": kind_of(value), "signature": signature_payload(value) if ( inspect.isfunction(value) or inspect.ismethod(value) or inspect.iscoroutinefunction(value) ) else None, } if inspect.isclass(value): entry.update(collect_class(value)) payload["exports"][name] = entry return payload def main() -> None: sys.path.insert(0, str(ROOT / "src")) warnings.simplefilter("ignore", DeprecationWarning) modules = discover_modules() manifest = {"modules": {}} errors = [] for module_name in modules: try: manifest["modules"][module_name] = collect_module(module_name) except Exception as exc: errors.append(f"{module_name}: {type(exc).__name__}: {exc}") if errors: print(json.dumps({"errors": errors}, sort_keys=True)) raise SystemExit(2) print(json.dumps(manifest, sort_keys=True)) main() """ _OBJECT_SENTINEL_REPR_RE = re.compile(r"") def normalize_default_repr(default_repr: str | None) -> str | None: """Collapse a bare object() sentinel default repr to an address-free form. A bare object() sentinel default (e.g. the wait_for_completion initial_interval sentinel) reprs as ; the hex address differs between the baseline collector process and the current one, so identical code would otherwise read as a changed default. Only the bare object() sentinel (matching the whole repr) is normalized, so two same-identity sentinels compare equal while every other default — including an address-bearing instance or function repr — is left intact and a genuine change is still caught. """ if default_repr is None: return None if _OBJECT_SENTINEL_REPR_RE.fullmatch(default_repr): return "" return default_repr def collect_manifest( source_root: Path, extra_public_names: dict[str, list[str]] | None = None, *, enforce_all: bool = True, ) -> dict[str, Any]: """Run the collector in a clean Python process for ``source_root``. ``enforce_all`` gates the "every public module declares ``__all__``" rule (issue #1493): pass ``True`` for the current checkout and ``False`` for a historical baseline that predates the rule, so baseline collection never aborts before the diff. """ env = os.environ.copy() existing_pythonpath = env.get("PYTHONPATH") pythonpath = str(source_root / "src") if existing_pythonpath: pythonpath = pythonpath + os.pathsep + existing_pythonpath env["PYTHONPATH"] = pythonpath result = subprocess.run( [ sys.executable, "-c", _COLLECTOR, str(source_root), json.dumps(extra_public_names or {}, sort_keys=True), json.dumps(CLIENT_NAMESPACE_ATTRIBUTES), PUBLIC_PACKAGE, json.dumps(sorted(EXCLUDED_TOP_LEVEL_MODULES)), json.dumps(EXTRA_PUBLIC_PACKAGES), "1" if enforce_all else "0", ], cwd=source_root, env=env, capture_output=True, text=True, check=False, ) if result.returncode != 0: message = result.stderr.strip() or result.stdout.strip() if result.stdout.strip(): try: payload = json.loads(result.stdout) except json.JSONDecodeError: pass else: errors = payload.get("errors") if isinstance(payload, dict) else None if isinstance(errors, list): message = "; ".join(str(error) for error in errors) raise RuntimeError(f"public API collection failed for {source_root}: {message}") try: return json.loads(result.stdout) except json.JSONDecodeError as exc: raise RuntimeError( f"public API collection returned invalid JSON for {source_root}: {exc}" ) from exc def _has_kind(params: list[dict[str, Any]], kind: str) -> bool: return any(param["kind"] == kind for param in params) def _accepts_keyword(param: dict[str, Any]) -> bool: return param["kind"] in {"POSITIONAL_OR_KEYWORD", "KEYWORD_ONLY", "VAR_KEYWORD"} def _accepts_positional(param: dict[str, Any]) -> bool: return param["kind"] in { "POSITIONAL_ONLY", "POSITIONAL_OR_KEYWORD", "VAR_POSITIONAL", } def _signature_breakage(old: dict[str, Any] | None, new: dict[str, Any] | None) -> str | None: """Return a short incompatibility reason, or ``None`` when old calls still fit.""" if old is None or new is None: if old != new: return f"signature changed from {old!r} to {new!r}" return None old_params = old["parameters"] new_params = new["parameters"] old_by_name = {param["name"]: param for param in old_params} new_by_name = {param["name"]: param for param in new_params} new_has_var_keyword = _has_kind(new_params, "VAR_KEYWORD") new_has_var_positional = _has_kind(new_params, "VAR_POSITIONAL") for old_param in old_params: kind = old_param["kind"] name = old_param["name"] if kind == "VAR_POSITIONAL" and not new_has_var_positional: return f"old signature accepted *{name}, new signature does not" if kind == "VAR_KEYWORD" and not new_has_var_keyword: return f"old signature accepted **{name}, new signature does not" if _accepts_keyword(old_param): new_param = new_by_name.get(name) if new_param is None: if not new_has_var_keyword: return f"keyword parameter {name!r} was removed" continue if not _accepts_keyword(new_param): return f"parameter {name!r} no longer accepts keyword calls" if old_param["has_default"] and not new_param["has_default"]: return f"optional parameter {name!r} became required" old_default = normalize_default_repr(old_param.get("default_repr")) new_default = normalize_default_repr(new_param.get("default_repr")) if old_param["has_default"] and new_param["has_default"] and old_default != new_default: return f"default for parameter {name!r} changed from {old_default} to {new_default}" old_positional = [param for param in old_params if _accepts_positional(param)] new_positional = [param for param in new_params if _accepts_positional(param)] if not new_has_var_positional and len(new_positional) < len(old_positional): return ( f"new signature accepts only {len(new_positional)} positional argument(s); " f"old accepted {len(old_positional)}" ) old_fixed_positional = [param for param in old_positional if param["kind"] != "VAR_POSITIONAL"] new_fixed_positional = [param for param in new_positional if param["kind"] != "VAR_POSITIONAL"] new_fixed_names = [param["name"] for param in new_fixed_positional] for index, old_param in enumerate(old_fixed_positional): if index >= len(new_fixed_positional): break old_name = old_param["name"] new_name = new_fixed_positional[index]["name"] if old_name == new_name: continue if old_name in new_fixed_names: new_index = new_fixed_names.index(old_name) return ( f"positional parameter {old_name!r} moved from position " f"{index + 1} to {new_index + 1}" ) return ( f"positional parameter {old_name!r} was replaced at position " f"{index + 1} by {new_name!r}" ) for new_param in new_params: if new_param["kind"] in {"VAR_POSITIONAL", "VAR_KEYWORD"}: continue if new_param["has_default"]: continue if new_param["name"] not in old_by_name: return f"new required parameter {new_param['name']!r} was added" return None def _return_breakage(old: dict[str, Any] | None, new: dict[str, Any] | None) -> str | None: """Return a reason when the return annotation changed, else ``None``. Older baselines predate return-annotation capture, so a missing ``return_annotation`` key is treated as "unknown" and never reported — only an observed value-to-value change counts as a break. An annotation appearing where there was none before is additive and also ignored. The mirror case — an annotation disappearing (annotated -> unannotated) — *is* reported; acknowledge it via the allowlist if intentional. """ if old is None or new is None: return None if "return_annotation" not in old or "return_annotation" not in new: return None old_return = old["return_annotation"] new_return = new["return_annotation"] if old_return is None or old_return == new_return: return None return f"return annotation changed from {old_return!r} to {new_return!r}" def _compare_export( module_name: str, export_name: str, old: dict[str, Any], new: dict[str, Any], ) -> list[ApiBreak]: path = f"{module_name}.{export_name}" breaks: list[ApiBreak] = [] if old["kind"] != new["kind"]: breaks.append( ApiBreak( "changed-kind", path, f"kind changed from {old['kind']!r} to {new['kind']!r}", ) ) return breaks if old["kind"] in {"function", "class", "enum"}: reason = _signature_breakage(old.get("signature"), new.get("signature")) if reason: breaks.append(ApiBreak("changed-signature", path, reason)) return_reason = _return_breakage(old.get("signature"), new.get("signature")) if return_reason: breaks.append(ApiBreak("changed-return", path, return_reason)) for member_name, old_member in old.get("members", {}).items(): new_member = new.get("members", {}).get(member_name) member_path = f"{path}.{member_name}" if new_member is None: breaks.append( ApiBreak( "removed-member", member_path, f"{member_path} existed in the baseline and is missing now", ) ) continue if old_member["kind"] != new_member["kind"]: breaks.append( ApiBreak( "changed-member-kind", member_path, f"kind changed from {old_member['kind']!r} to {new_member['kind']!r}", ) ) continue reason = _signature_breakage(old_member.get("signature"), new_member.get("signature")) if reason: breaks.append(ApiBreak("changed-signature", member_path, reason)) return_reason = _return_breakage(old_member.get("signature"), new_member.get("signature")) if return_reason: breaks.append(ApiBreak("changed-return", member_path, return_reason)) for enum_name, old_value in old.get("enum_members", {}).items(): enum_members = new.get("enum_members", {}) enum_path = f"{path}.{enum_name}" if enum_name not in enum_members: breaks.append( ApiBreak( "removed-enum-member", enum_path, f"{enum_path} existed in the baseline and is missing now", ) ) elif enum_members[enum_name] != old_value: breaks.append( ApiBreak( "changed-enum-value", enum_path, f"value changed from {old_value!r} to {enum_members[enum_name]!r}", ) ) return breaks def compare_manifests(baseline: dict[str, Any], current: dict[str, Any]) -> list[ApiBreak]: """Return public API breaks from ``baseline`` to ``current``.""" breaks: list[ApiBreak] = [] current_modules = current.get("modules", {}) for module_name, old_module in sorted(baseline.get("modules", {}).items()): new_module = current_modules.get(module_name) if new_module is None: breaks.append( ApiBreak( "removed-module", module_name, f"public module {module_name} existed in the baseline and is missing now", ) ) continue for export_name, old_export in sorted(old_module.get("exports", {}).items()): new_export = new_module.get("exports", {}).get(export_name) export_path = f"{module_name}.{export_name}" if new_export is None: breaks.append( ApiBreak( "removed-export", export_path, f"{export_path} existed in the baseline public surface and is missing now", ) ) continue breaks.extend(_compare_export(module_name, export_name, old_export, new_export)) return sorted(breaks, key=lambda item: (item.code, item.object, item.detail)) def load_policy(path: Path | None) -> tuple[list[Allowance], dict[str, list[str]]]: if path is None or str(path) == "": return [], {} if not path.exists(): raise RuntimeError(f"allowlist file not found: {path}") try: payload = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise RuntimeError(f"invalid JSON in {path}: {exc}") from exc if not isinstance(payload, dict): raise RuntimeError(f"{path} must contain a JSON object") schema_version = payload.get("schema_version", 1) if schema_version != 1: raise RuntimeError(f"{path}: unsupported schema_version {schema_version!r} (expected 1)") entries = payload.get("allowed_breaks", []) if not isinstance(entries, list): raise RuntimeError(f"{path} must contain an 'allowed_breaks' list") extra_public_names = payload.get("extra_public_names", {}) if not isinstance(extra_public_names, dict): raise RuntimeError(f"{path} extra_public_names must be an object") normalized_extra_names: dict[str, list[str]] = {} for module_name, names in extra_public_names.items(): if not isinstance(names, list) or not all(isinstance(name, str) for name in names): raise RuntimeError( f"{path} extra_public_names[{module_name!r}] must be a list of strings" ) normalized_extra_names[str(module_name)] = sorted(set(names), key=str.lower) allowances: list[Allowance] = [] for index, entry in enumerate(entries): if not isinstance(entry, dict): raise RuntimeError(f"{path}: allowed_breaks[{index}] must be an object") try: code = str(entry["code"]) obj = str(entry["object"]) reason = str(entry["reason"]) except KeyError as exc: raise RuntimeError( f"{path}: allowed_breaks[{index}] is missing required key {exc.args[0]!r}" ) from exc allowances.append(Allowance(code=code, object=obj, reason=reason)) return allowances, normalized_extra_names def partition_allowed( breakages: list[ApiBreak], allowances: list[Allowance], ) -> tuple[list[ApiBreak], list[tuple[ApiBreak, Allowance]]]: unapproved: list[ApiBreak] = [] approved: list[tuple[ApiBreak, Allowance]] = [] for breakage in breakages: allowance = next((item for item in allowances if item.matches(breakage)), None) if allowance is None: unapproved.append(breakage) else: approved.append((breakage, allowance)) return unapproved, approved def _sibling_object(obj: str) -> str | None: """Return the other path-view of an exported object, or ``None``. The audit records the same client-namespace callable under two dotted paths: the re-export view ``notebooklm.X`` and the defining-module view ``notebooklm.client.X``. An allowance is written against one view; this maps between them so the staleness check can treat the pair as a single unit. A glob object (containing ``*``) is left for the caller to handle — the returned sibling is a literal string and is only consulted via an exact lookup, so a glob's sibling never spuriously matches. """ client_prefix = f"{PUBLIC_PACKAGE}.client." bare_prefix = f"{PUBLIC_PACKAGE}." if obj.startswith(client_prefix): return bare_prefix + obj[len(client_prefix) :] if obj.startswith(bare_prefix): return client_prefix + obj[len(bare_prefix) :] return None def stale_allowances( breakages: list[ApiBreak], allowances: list[Allowance], ) -> list[Allowance]: """Return allowances that match no current break against the baseline. An allowance is *stale* when it describes a break already baked into the baseline (so it no longer surfaces as a break against it). Such entries are dead weight: harmless to the gate, but the set only ever grows. Pruning them at each release boundary keeps the allowlist scoped to the breaks pending the *next* release (see ``docs/releasing.md`` → prune-allowlist-at-release). Pair-aware rule: the two path-views ``notebooklm.X`` and ``notebooklm.client.X`` of the same callable are treated as one unit — a unit is live (kept) if *either* view matches a break. So a non-stale allowance is one that itself matches a break, or whose sibling path-view has *any* matching allowance. Today both views always match together, but a future change that only one view detects must not flag its still load-bearing sibling. """ # Per-allowance self-match, keyed by (code, object) so two allowances on the # same object but different codes never collapse onto one another. self_matched: dict[tuple[str, str], bool] = { (allowance.code, allowance.object): any( allowance.matches(breakage) for breakage in breakages ) for allowance in allowances } # Per-object aggregate for the sibling lookup: an object is "kept" if *any* # of its allowances (any code) matches a break. The pair stays live as long # as the sibling object has a live allowance, regardless of code. object_kept: dict[str, bool] = {} for (_code, obj), is_match in self_matched.items(): object_kept[obj] = object_kept.get(obj, False) or is_match def _is_live(allowance: Allowance) -> bool: if self_matched[(allowance.code, allowance.object)]: return True sibling = _sibling_object(allowance.object) return sibling is not None and object_kept.get(sibling, False) return [allowance for allowance in allowances if not _is_live(allowance)] def _render_stale(stale: list[Allowance], baseline_ref: str, allowlist_path: Path | str) -> str: lines = [ f"Stale allowlist entries — they match no break against the {baseline_ref} " "baseline, so they are already in the baseline:", ] for allowance in stale: lines.append(f" - [{allowance.code}] {allowance.object}") lines.append( f"Prune them from {allowlist_path} (see docs/releasing.md → prune-allowlist-at-release)." ) return "\n".join(lines) def _render_breakages(title: str, breakages: list[ApiBreak]) -> str: if not breakages: return "" lines = [title] for breakage in breakages: lines.append(f" - [{breakage.code}] {breakage.object}: {breakage.detail}") return "\n".join(lines) def _render_approved(approved: list[tuple[ApiBreak, Allowance]]) -> str: if not approved: return "" lines = ["Allowlisted compatibility breaks:"] for breakage, allowance in approved: lines.append(f" - [{breakage.code}] {breakage.object}: {allowance.reason}") return "\n".join(lines) def main(argv: list[str] | None = None) -> int: repo_root = Path(__file__).resolve().parent.parent parser = argparse.ArgumentParser( description="Compare the public Python API against a previous release tag." ) parser.add_argument( "--baseline-ref", default=None, help=( "Git ref to compare against. Defaults to the latest reachable stable " "release tag (pre-release aN/bN/rcN and non-release tags are skipped)." ), ) parser.add_argument( "--allowlist", default=str(repo_root / DEFAULT_ALLOWLIST), help=( "JSON file containing reviewed compatibility breaks. Use an empty string to disable." ), ) parser.add_argument("--json", action="store_true", help="Print machine-readable output.") parser.add_argument( "--check-stale", action="store_true", help=( "Also fail when an allowlist entry matches no break against the " "baseline (it is already in the baseline — prune it)." ), ) args = parser.parse_args(argv) try: baseline_ref = args.baseline_ref or latest_release_tag(repo_root) allowlist_path = Path(args.allowlist) if args.allowlist else None allowances, extra_public_names = load_policy(allowlist_path) with tempfile.TemporaryDirectory(prefix="notebooklm-api-compat-") as tmp: baseline_root = export_git_ref(repo_root, baseline_ref, Path(tmp)) # Enforce the __all__ rule only for the current checkout; an older # baseline may legitimately predate it (issue #1493 review). baseline_manifest = collect_manifest( baseline_root, extra_public_names, enforce_all=False ) current_manifest = collect_manifest(repo_root, extra_public_names, enforce_all=True) breakages = compare_manifests(baseline_manifest, current_manifest) unapproved, approved = partition_allowed(breakages, allowances) stale = stale_allowances(breakages, allowances) except RuntimeError as exc: if args.json: print(json.dumps({"error": str(exc)}, sort_keys=True)) else: print(f"ERROR: {exc}", file=sys.stderr) return 2 allowlist_display = allowlist_path or DEFAULT_ALLOWLIST # ``--check-stale`` promotes stale entries from informational to a gate. stale_blocks = args.check_stale and bool(stale) failed = bool(unapproved) or stale_blocks if args.json: print( json.dumps( { "baseline_ref": baseline_ref, "approved": [ {"break": asdict(breakage), "reason": allowance.reason} for breakage, allowance in approved ], "unapproved": [asdict(item) for item in unapproved], "stale_allowances": [asdict(allowance) for allowance in stale], }, indent=2, sort_keys=True, ) ) return 1 if failed else 0 if unapproved: print( textwrap.dedent( f"""\ Public API compatibility audit failed. Baseline: {baseline_ref} {_render_breakages("Unapproved compatibility breaks:", unapproved)} Add back-compat shims, or document an intentional break in {allowlist_display} with a reviewer-readable reason. """ ).strip(), file=sys.stderr, ) approved_text = _render_approved(approved) if approved_text: print("\n" + approved_text, file=sys.stderr) elif stale_blocks: # Compat surface is clean, but stale allowlist entries fail the gate # under --check-stale. Don't print an "OK:" line that contradicts the # non-zero exit; the stale report below carries the actionable message. print( f"Public API is compatible with {baseline_ref} " f"({len(approved)} reviewed break(s) allowlisted), " "but the allowlist has stale entries.", file=sys.stderr, ) else: print( f"OK: public API is compatible with {baseline_ref} " f"({len(approved)} reviewed break(s) allowlisted)." ) approved_text = _render_approved(approved) if approved_text: print(approved_text) if stale_blocks: print("\n" + _render_stale(stale, baseline_ref, allowlist_display), file=sys.stderr) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())