tagging.py

# /// script
# dependencies = ["PyGithub>=2,<3", "pyjwt<2.12.0", "charset-normalizer<3.4.6"]
# ///

import os
import re
import argparse
from typing import Optional, List, Callable, Dict
from dataclasses import dataclass, replace
import subprocess
import time
import json
from github import Auth, Github, Repository, InputGitTreeElement, InputGitAuthor
from datetime import datetime, timezone

NEXT_CHANGELOG_FILE_NAME = "NEXT_CHANGELOG.md"
CHANGELOG_FILE_NAME = "CHANGELOG.md"
PACKAGE_FILE_NAME = ".package.json"
CODEGEN_FILE_NAME = ".codegen.json"
CREATED_TAGS_FILE_NAME = "created_tags.json"

# Presence of this env var switches the changelog source from a single
# hand-maintained ``NEXT_CHANGELOG.md`` to per-PR ``<dir>/<section>/*.md``
# fragments, and the release version from the ``## Release vX.Y.Z`` header to a
# ``<dir>/version`` file. Its value is the fragment directory name (e.g.
# ``.nextchanges``). Unset for every SDK repo, so their behavior is unchanged;
# a repo opts in by setting it in the tagging workflow (see the databricks/cli
# release workflow). The per-section ``(slug, header)`` mapping is read from
# ``.codegen.json``'s ``nextchanges_sections`` key.
NEXTCHANGES_DIR_ENV = "NEXTCHANGES_DIR"

# File inside the fragment directory tracking the next release's version —
# read at release time and bumped afterward, the role the ``## Release vX.Y.Z``
# header plays in the ``NEXT_CHANGELOG.md`` flow.
NEXTCHANGES_VERSION_FILE = "version"

# ``README.md`` in a section slug is documentation (e.g. "put CLI changelog
# fragments here"), not a changelog fragment. It is excluded from rendering and
# preserved across releases, so teams can keep per-slug guidance in place.
NEXTCHANGES_README_FILE = "README.md"
"""
This script tags the release of the SDKs using a combination of the GitHub API and Git commands.
It reads the local repository to determine necessary changes, updates changelogs, and creates tags.

### How it Works:
- It does **not** modify the local repository directly.
- Instead of committing and pushing changes locally, it uses the **GitHub API** to create commits and tags.
"""


@dataclass(frozen=True)
class Version:
    """
    A semver 2.0.0-compliant version (https://semver.org).

    Mirrors the API of the `semver` PyPI package so this implementation can be
    swapped for that library if it is ever added to the wheelhouse. Supports
    parsing, stringification, and the two bumps we need: minor (for stable
    releases) and prerelease (for release trains).
    """

    # Permissive pattern for locating a semver version string inside larger
    # text (e.g. a changelog header). Callers use it in f-strings; strict
    # validation happens via Version.parse.
    PATTERN = r"\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?"

    # Strict anchored regex per https://semver.org. Rejects leading zeros in
    # numeric identifiers and invalid pre-release/build identifier charsets.
    _PARSE_REGEX = re.compile(
        r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
        r"(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)"
        r"(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?"
        r"(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$"
    )

    major: int
    minor: int
    patch: int
    prerelease: str = ""
    build: str = ""

    @classmethod
    def parse(cls, text: str) -> "Version":
        """Parse a semver string, raising ValueError on malformed input."""
        match = cls._PARSE_REGEX.match(text)
        if not match:
            raise ValueError(f"Invalid semver version: {text!r}")
        major, minor, patch, prerelease, build = match.groups()
        return cls(
            major=int(major),
            minor=int(minor),
            patch=int(patch),
            prerelease=prerelease or "",
            build=build or "",
        )

    def __str__(self) -> str:
        result = f"{self.major}.{self.minor}.{self.patch}"
        if self.prerelease:
            result += f"-{self.prerelease}"
        if self.build:
            result += f"+{self.build}"
        return result

    def bump_minor(self) -> "Version":
        """
        Bump the minor version and reset patch.

        Per semver item 9, a pre-release version has lower precedence than
        the same MAJOR.MINOR.PATCH, so bumping to a new minor drops any
        pre-release and build metadata.
        """
        return Version(major=self.major, minor=self.minor + 1, patch=0)

    def bump_prerelease(self) -> "Version":
        """
        Increment the rightmost numeric identifier in the pre-release.

        Matches the npm `prerelease` bump semantics:
            0.0.0-alpha.1 -> 0.0.0-alpha.2
            0.0.0-alpha   -> 0.0.0-alpha.1
            0.0.0-rc.1.2  -> 0.0.0-rc.1.3

        Raises ValueError if the version has no pre-release to bump.
        Build metadata is dropped since it does not affect precedence.
        """
        if not self.prerelease:
            raise ValueError(f"Cannot bump prerelease of {self}: no pre-release component")
        parts = self.prerelease.split(".")
        for i in range(len(parts) - 1, -1, -1):
            if parts[i].isdigit():
                parts[i] = str(int(parts[i]) + 1)
                return replace(self, prerelease=".".join(parts), build="")
        # No numeric identifier exists; append ".1" to start a counter.
        return replace(self, prerelease=f"{self.prerelease}.1", build="")

    def next_release_version(self) -> "Version":
        """
        Default next version for the changelog after this one is released.

        If on a pre-release track, stay on it by bumping the pre-release
        identifier (npm convention). Otherwise, bump the minor version,
        the script's historical default for stable releases. Teams can
        override the default in the release PR.
        """
        if self.prerelease:
            return self.bump_prerelease()
        return self.bump_minor()


def _read_local_head_sha() -> str:
    """
    Returns the SHA of the local working tree's HEAD via ``git rev-parse``.
    """
    return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()


def _release_branch() -> str:
    """
    Returns the branch this release is being cut from.

    The tagging workflow sets ``DECO_TAGGING_REF`` to the branch it was
    dispatched on (``github.ref_name``) so a release can be cut from a
    branch other than main. It is unset for local runs and the historical
    main-only release path, so we default to ``main`` and every existing
    caller is unaffected.
    """
    return os.environ.get("DECO_TAGGING_REF", "").strip() or "main"


class MainAdvancedError(Exception):
    """
    Raised when the release branch (``origin/main`` by default; see
    ``_release_branch``) has advanced since the workflow's checkout —
    i.e., another commit landed during this run. The local working tree
    is now stale, so any commit produced from it would silently revert
    whatever the concurrent push added.
    """


# GitHub does not support signing commits for GitHub Apps directly.
# This class replaces usages for git commands such as "git add", "git commit", and "git push".
@dataclass
class GitHubRepo:
    def __init__(self, repo: Repository):
        self.repo = repo
        self.changed_files: list[InputGitTreeElement] = []
        # Branch the changelog-bump commit + tag land on. Defaults to
        # ``heads/main``; ``DECO_TAGGING_REF`` overrides it for a branch
        # release. See ``_release_branch``.
        self.ref = f"heads/{_release_branch()}"
        # Anchor ``self.sha`` to the **local checkout** rather than a
        # live API call. ``actions/checkout`` populates the working tree
        # at this SHA, and every subsequent file read in this run is
        # against that tree; the API HEAD is only relevant when we go
        # to push.
        self.sha = _read_local_head_sha()

    # Replaces "git add file"
    def add_file(self, loc: str, content: str):
        local_path = os.path.relpath(loc, os.getcwd())
        print(f"Adding file {local_path}")
        blob = self.repo.create_git_blob(content=content, encoding="utf-8")
        element = InputGitTreeElement(path=local_path, mode="100644", type="blob", sha=blob.sha)
        self.changed_files.append(element)

    # Replaces "git rm file"
    def delete_file(self, loc: str):
        """``git rm`` equivalent for GitHubRepo: stage a tree deletion (sha=None)."""
        local_path = os.path.relpath(loc, os.getcwd())
        print(f"Deleting file {local_path}")
        self.changed_files.append(InputGitTreeElement(path=local_path, mode="100644", type="blob", sha=None))

    # Replaces "git commit && git push"
    def commit_and_push(self, message: str):
        head_ref = self.repo.get_git_ref(self.ref)
        if head_ref.object.sha != self.sha:
            raise MainAdvancedError(
                f"{self.ref} advanced from {self.sha} to {head_ref.object.sha} "
                f"during this run. Local working tree is stale; aborting before "
                f"the commit would silently revert the new content. Re-run the "
                f"workflow."
            )
        base_tree = self.repo.get_git_tree(sha=head_ref.object.sha)
        new_tree = self.repo.create_git_tree(self.changed_files, base_tree)
        parent_commit = self.repo.get_git_commit(head_ref.object.sha)

        new_commit = self.repo.create_git_commit(message=message, tree=new_tree, parents=[parent_commit])
        # Update branch reference.
        head_ref.edit(new_commit.sha)
        self.sha = new_commit.sha

    def reset(self, sha: Optional[str] = None):
        self.changed_files = []
        if sha:
            self.sha = sha
        else:
            self.sha = _read_local_head_sha()

    def tag(self, tag_name: str, tag_message: str):
        # Create a tag pointing to the new commit
        # The email MUST be the GitHub Apps email.
        # Otherwise, the tag will not be verified.
        tagger = InputGitAuthor(
            name="Databricks SDK Release Bot", email="DECO-SDK-Tagging[bot]@users.noreply.github.com"
        )

        tag = self.repo.create_git_tag(tag=tag_name, message=tag_message, object=self.sha, type="commit", tagger=tagger)
        # Create a Git ref (the actual reference for the tag in the repo)
        self.repo.create_git_ref(ref=f"refs/tags/{tag_name}", sha=tag.sha)


gh: Optional[GitHubRepo] = None


@dataclass
class Package:
    """
    Represents a package in the repository.
    :name: The package name.
    :path: The path to the package relative to the repository root.
    """

    name: str
    path: str


@dataclass
class TagInfo:
    """
    Represents all changes on a release.
    :package: package info.
    :version: release version for the package. Format: v<major>.<minor>.<pacth>
    :content: changes for the release, as they appear in the changelog.
              When written to CHANGELOG.md, the current date (YYYY-MM-DD) is automatically added.

    Example (from NEXT_CHANGELOG.md):

    ## Release v0.56.0

    ### New Features and Improvements
    * Feature
    * Some improvement

    ### Bug Fixes
    * Bug fix

    ### Documentation
    * Doc Changes

    ### Internal Changes
    * More Changes

    ### API Changes
    * Add new Service

    Note: When written to CHANGELOG.md, the header becomes: ## Release v0.56.0 (YYYY-MM-DD)

    """

    package: Package
    version: str
    content: str

    def tag_name(self) -> str:
        return f"{self.package.name}/v{self.version}" if self.package.name else f"v{self.version}"


def get_package_name(package_path: str) -> str:
    """
    Returns the package name from the package path.
    The name is found inside the .package.json file:
    {
        "package": "package_name"
    }
    """
    filepath = os.path.join(os.getcwd(), package_path, PACKAGE_FILE_NAME)
    with open(filepath, "r") as file:
        content = json.load(file)
    if "package" in content:
        return content["package"]
    # Legacy SDKs have no packages.
    return ""


def stage_version_updates(tag_infos: List[TagInfo], packages: List[Package]) -> None:
    """
    Stages all version-related edits for the release in a single pass over
    every package the workspace already opts in via ``.package.json``.
    """

    # Load patterns from '.codegen.json' at the top level of the repository.
    package_file_path = os.path.join(os.getcwd(), CODEGEN_FILE_NAME)
    with open(package_file_path, "r") as file:
        codegen = json.load(file)

    version_patterns = codegen.get("version", {})
    dep_patterns = codegen.get("dependency_pattern", {})
    name_template = codegen.get("dependency_name_template", "")

    if not version_patterns and not dep_patterns:
        print("Neither `version` nor `dependency_pattern` found in .codegen.json. Nothing to update.")
        return

    bumped_by_dir: Dict[str, TagInfo] = {info.package.path: info for info in tag_infos}
    new_dep_versions = compute_dependency_rewrites(tag_infos, name_template)

    files = sorted(set(version_patterns.keys()) | set(dep_patterns.keys()))

    for pkg in packages:
        for filename in files:
            loc = os.path.join(os.getcwd(), pkg.path, filename)

            with open(loc, "r") as file:
                content = file.read()
            original = content

            # Own version (only when this package is being released and the
            # file has a version pattern declared).
            info = bumped_by_dir.get(pkg.path)
            if info is not None and filename in version_patterns:
                pattern = version_patterns[filename]
                previous_version = pattern.replace("$VERSION", Version.PATTERN)
                new_version = pattern.replace("$VERSION", info.version)
                content = re.sub(previous_version, new_version, content)

            # Sibling dependency rewrites (only when the file has a
            # dependency pattern and there is at least one bumped sibling).
            if filename in dep_patterns and new_dep_versions:
                content = rewrite_dependencies(content, dep_patterns[filename], new_dep_versions)

            if content != original:
                gh.add_file(loc, content)


def compute_dependency_rewrites(
    tag_infos: List[TagInfo],
    name_template: str,
) -> Dict[str, str]:
    """
    Returns a map of dependency-name to the new semver string for each
    bumped package.
    """
    if not name_template:
        return {}
    rewrites: Dict[str, str] = {}
    for info in tag_infos:
        # Skip legacy releases that don't have a per-package name; their
        # tag_info has an empty package.name and they can't be referenced
        # as a sibling dep anyway.
        if not info.package.name:
            continue
        dep_name = name_template.replace("$PACKAGE", info.package.name)
        rewrites[dep_name] = info.version
    return rewrites


def rewrite_dependencies(content: str, pattern: str, new_versions: Dict[str, str]) -> str:
    """
    Apply ``pattern`` (with ``$DEPENDENCY`` and ``$VERSION`` placeholders) to
    rewrite every entry in ``content`` whose dependency name appears in
    ``new_versions``.
    """
    # Sentinel strings used to protect the placeholders through re.escape:
    # we substitute them in, escape the whole template, then swap them out
    # for the dep-name literal and Version.PATTERN. Control characters so
    # they can't collide with anything in real .codegen.json patterns.
    dep_sentinel = "\x01DEPENDENCY\x01"
    ver_sentinel = "\x01VERSION\x01"

    for dep_name, new_value in new_versions.items():
        regex = pattern.replace("$DEPENDENCY", dep_sentinel).replace("$VERSION", ver_sentinel)
        regex = re.escape(regex)
        regex = regex.replace(re.escape(dep_sentinel), re.escape(dep_name))
        regex = regex.replace(re.escape(ver_sentinel), Version.PATTERN)

        # Build the literal replacement text by substituting the same
        # placeholders directly. A lambda is used instead of a string to
        # avoid re.sub interpreting \1, \g<...>, etc. inside the value.
        replacement_text = pattern.replace("$DEPENDENCY", dep_name).replace("$VERSION", new_value)
        content = re.sub(regex, lambda _m, text=replacement_text: text, content)
    return content


def clean_next_changelog(package_path: str) -> None:
    """
    Cleans the "NEXT_CHANGELOG.md" file. It performs 2 operations:
    * Increase the version to the next minor version.
    * Remove release notes. Sections names are kept to
      keep consistency in the section names between releases.
    """

    file_path = os.path.join(os.getcwd(), package_path, NEXT_CHANGELOG_FILE_NAME)
    with open(file_path, "r") as file:
        content = file.read()

    # Remove content between ### sections.
    cleaned_content = re.sub(r"(### [^\n]+\n)(?:.*?\n?)*?(?=###|$)", r"\1", content)
    # Ensure there is exactly one empty line before each section.
    cleaned_content = re.sub(r"(\n*)(###[^\n]+)", r"\n\n\2", cleaned_content)
    # Find the version number and compute the default next release version.
    # Teams can adjust the version in the PR if the default is not desired.
    # For stable versions, bump minor (historical default since minor releases
    # are more common than patch or major). For pre-release versions, stay on
    # the same track by bumping the pre-release identifier (npm convention).
    version_match = re.search(rf"Release v({Version.PATTERN})", cleaned_content)
    if not version_match:
        raise Exception("Version not found in the changelog")
    current = Version.parse(version_match.group(1))
    new_header = f"Release v{current.next_release_version()}"
    cleaned_content = cleaned_content.replace(version_match.group(0), new_header)

    # Update file with cleaned content
    gh.add_file(file_path, cleaned_content)


def get_previous_tag_info(package: Package) -> Optional[TagInfo]:
    """
    Extracts the previous tag info from the "CHANGELOG.md" file.
    Used for failure recovery purposes.
    """
    changelog_path = os.path.join(os.getcwd(), package.path, CHANGELOG_FILE_NAME)

    with open(changelog_path, "r") as f:
        changelog = f.read()

    # Extract the latest release section using regex.
    match = re.search(
        rf"## (\[Release\] )?Release v{Version.PATTERN}.*?(?=\n## (\[Release\] )?Release v|\Z)",
        changelog,
        re.S,
    )

    # E.g., for new packages.
    if not match:
        return None

    latest_release = match.group(0)
    version_match = re.search(rf"## (\[Release\] )?Release v({Version.PATTERN})", latest_release)

    if not version_match:
        raise Exception("Version not found in the changelog")

    # Validate the extracted string is spec-compliant; fail loudly otherwise.
    version = str(Version.parse(version_match.group(2)))
    return TagInfo(package=package, version=version, content=latest_release)


def _load_codegen_config(package_path: str = "") -> Dict:
    """
    Loads ``.codegen.json`` for a package: prefers ``<package_path>/.codegen.json``
    and falls back to the repo-root file, returning an empty dict when neither
    exists. ``package_path=""`` (the default) reads the repo root, matching the
    single-package / root-config layout.

    Package-local lookup keeps the section taxonomy and other codegen options in
    lockstep with the package-relative directories the nextchanges helpers
    resolve, so a multi-package repo can give each package its own config
    instead of every package sharing the root file.
    """
    candidates = [os.path.join(os.getcwd(), package_path, CODEGEN_FILE_NAME)]
    root_config = os.path.join(os.getcwd(), CODEGEN_FILE_NAME)
    if root_config not in candidates:
        candidates.append(root_config)
    for candidate in candidates:
        if os.path.exists(candidate):
            with open(candidate, "r") as file:
                return json.load(file)
    return {}


def _nextchanges_dir() -> Optional[str]:
    """
    Returns the fragment directory name when nextchanges mode is enabled
    (``NEXTCHANGES_DIR`` set to a non-empty value), else ``None``. In the
    ``None`` case the historical ``NEXT_CHANGELOG.md`` flow is used, so every
    repo that doesn't set the env var is unaffected.
    """
    return os.environ.get(NEXTCHANGES_DIR_ENV, "").strip() or None


def _nextchanges_sections(package_path: str = "") -> List[tuple]:
    """
    Returns the ordered ``(slug, header)`` section list from the package's
    ``.codegen.json`` ``nextchanges_sections`` — the mapping of ``<dir>/<slug>/``
    subdirectories to the ``### <header>`` blocks they render into, in changelog
    order. Read via ``_load_codegen_config(package_path)`` so it matches the
    package-relative fragment directories (a multi-package repo can scope the
    taxonomy per package).

    Declared as a JSON object ``{"<slug>": "<header>", ...}`` so the per-repo
    section taxonomy stays out of this shared script; insertion order in the
    object is the changelog order (JSON objects preserve order in Python 3.7+).
    Raises when nextchanges mode is on but the key is absent/empty, or is not an
    object, since there would be nothing sensible to render.
    """
    sections = _load_codegen_config(package_path).get("nextchanges_sections", {})
    if not sections:
        raise Exception(
            f"nextchanges mode is enabled ({NEXTCHANGES_DIR_ENV} set) but "
            f"`nextchanges_sections` is missing or empty in {CODEGEN_FILE_NAME}."
        )
    if not isinstance(sections, dict):
        raise Exception(
            f"`nextchanges_sections` in {CODEGEN_FILE_NAME} must be a JSON object "
            f'mapping section slug to header (e.g. {{"cli": "CLI"}}), got '
            f"{type(sections).__name__}."
        )
    return list(sections.items())


def _render_fragment(text: str) -> str:
    """
    Render one fragment body into changelog bullets. Each line that starts with
    a ``* ``/``- `` marker (ignoring leading whitespace) becomes its own `` * ``
    bullet; a line without a marker is a continuation of the preceding bullet
    and is kept as authored. A markerless first line is itself a single bullet.

    So a fragment with multiple ``* ``/``- `` lines renders as multiple bullets,
    while a bullet followed by plain lines stays one bullet spanning those
    lines. Every bullet gets the leading-space ``*`` that matches CHANGELOG.md.
    """
    lines = []
    for line in text.split("\n"):
        marker = line.lstrip()
        if marker.startswith(("* ", "- ")):
            lines.append(f" * {marker[2:]}")
        elif lines:
            lines.append(line)
        else:
            lines.append(f" * {line}")
    return "\n".join(lines)


def render_nextchanges(package_path: str) -> Optional[str]:
    """
    Render ``<package_path>/<dir>/<section>/*.md`` fragments into the changelog
    body: one ``### <Section>`` block per non-empty section in
    ``nextchanges_sections`` order, fragments sorted by filename. Returns
    ``None`` when there are no fragments.

    Every ``.md`` file under a section directory is a fragment, except
    ``README.md`` which is treated as per-slug documentation and skipped (see
    ``NEXTCHANGES_README_FILE``). Empty/whitespace-only files contribute
    nothing. Files at the ``<dir>`` root or under a slug not listed in
    ``nextchanges_sections`` are ignored. Each fragment renders per
    ``_render_fragment``. Link expansion (e.g. ``(#1234)`` → markdown link) is
    assumed to have happened before release, so none here.
    """
    base = os.path.join(os.getcwd(), package_path, _nextchanges_dir())
    if not os.path.isdir(base):
        return None

    blocks = []
    for slug, header in _nextchanges_sections(package_path):
        section_dir = os.path.join(base, slug)
        if not os.path.isdir(section_dir):
            continue
        entries = []
        for name in sorted(os.listdir(section_dir)):
            if not name.endswith(".md") or name == NEXTCHANGES_README_FILE:
                continue
            with open(os.path.join(section_dir, name)) as f:
                text = f.read().strip()
            if not text:
                continue
            entries.append(_render_fragment(text))
        if entries:
            # Blank line after the heading, matching CHANGELOG.md.
            blocks.append(f"### {header}\n\n" + "\n".join(entries))

    if not blocks:
        return None
    return "\n\n".join(blocks)


def _nextchanges_version_path(package_path: str) -> str:
    return os.path.join(os.getcwd(), package_path, _nextchanges_dir(), NEXTCHANGES_VERSION_FILE)


def read_nextchanges_version(package: Package) -> str:
    """
    Release version for this run, read from the package's own ``<dir>/version``
    (resolved under ``package.path``, so each package in a multi-package repo
    keeps its own version source). In nextchanges mode this file — not the
    ``## Release v…`` changelog header — is the source of truth. To cut a patch
    or major release, edit it in the PR; otherwise its default (bumped to the
    next minor after the previous release by ``clean_nextchanges``) applies.

    Raises with an actionable message when the file is absent, so a package that
    opts into nextchanges mode without a version file fails loudly instead of
    with a bare ``FileNotFoundError``.
    """
    version_path = _nextchanges_version_path(package.path)
    if not os.path.exists(version_path):
        raise Exception(
            f"nextchanges mode is enabled ({NEXTCHANGES_DIR_ENV} set) but the version "
            f"file {os.path.relpath(version_path, os.getcwd())} is missing; each package "
            f"in nextchanges mode must provide its own <dir>/version file."
        )
    with open(version_path) as f:
        return str(Version.parse(f.read().strip().lstrip("v")))


def get_next_tag_info_from_nextchanges(package: Package) -> Optional[TagInfo]:
    """
    nextchanges-mode counterpart of ``get_next_tag_info``: build the release
    TagInfo from ``<dir>/`` fragments. Returns ``None`` when there are no
    entries (nothing to release), unless ``allow_empty_changelog`` is set in
    ``.codegen.json`` — matching the ``NEXT_CHANGELOG.md`` skip behavior.
    """
    body = render_nextchanges(package.path)
    if body is None and not _load_codegen_config(package.path).get("allow_empty_changelog", False):
        print(f"No {_nextchanges_dir()}/ entries. No changes will be made to the changelog.")
        return None

    version = read_nextchanges_version(package)
    # write_changelog() keys off the "## Release v…" header, so include it.
    content = f"## Release v{version}\n" + (f"\n{body}\n" if body else "")
    return TagInfo(package=package, version=version, content=content)


def clean_nextchanges(package_path: str) -> None:
    """
    nextchanges-mode counterpart of ``clean_next_changelog``: stage deletion of
    the ``<dir>/`` fragments consumed by this release and bump ``<dir>/version``
    to the next minor (its post-release default; teams can still override it in
    a PR). Deletes every ``.md`` under each section directory — the same set
    ``render_nextchanges`` consumed, so ``README.md`` is preserved — leaving the
    section directories in place.
    """
    base = os.path.join(os.getcwd(), package_path, _nextchanges_dir())
    for slug, _ in _nextchanges_sections(package_path):
        section_dir = os.path.join(base, slug)
        if not os.path.isdir(section_dir):
            continue
        # Deletion order is irrelevant, so listdir as-is (no sort needed).
        for name in os.listdir(section_dir):
            if name.endswith(".md") and name != NEXTCHANGES_README_FILE:
                gh.delete_file(os.path.join(section_dir, name))

    version_path = _nextchanges_version_path(package_path)
    with open(version_path) as f:
        released = Version.parse(f.read().strip().lstrip("v"))
    gh.add_file(version_path, f"{released.next_release_version()}\n")


def get_next_tag_info(package: Package) -> Optional[TagInfo]:
    """
    Extracts the changes for the next release. In nextchanges mode (see
    ``_nextchanges_dir``) it reads ``<dir>/`` fragments; otherwise it reads the
    package's ``NEXT_CHANGELOG.md``. The result is already processed.
    """
    if _nextchanges_dir() is not None:
        return get_next_tag_info_from_nextchanges(package)

    next_changelog_path = os.path.join(os.getcwd(), package.path, NEXT_CHANGELOG_FILE_NAME)
    # Read NEXT_CHANGELOG.md
    with open(next_changelog_path, "r") as f:
        next_changelog = f.read()

    # Remove "# NEXT CHANGELOG" line
    next_changelog = re.sub(r"^# NEXT CHANGELOG(\n+)", "", next_changelog, flags=re.MULTILINE)

    # Remove empty sections
    next_changelog = re.sub(r"###[^\n]+\n+(?=##|\Z)", "", next_changelog)
    # Ensure there is exactly one empty line before each section
    next_changelog = re.sub(r"(\n*)(###[^\n]+)", r"\n\n\2", next_changelog)

    # By default, packages whose NEXT_CHANGELOG.md has no populated
    # sections are skipped — there's nothing meaningful to release.
    # Repos like sdk-js which are still in development can opt in
    # by setting ``allow_empty_changelog: true`` in .codegen.json.
    if not re.search(r"###", next_changelog) and not _load_codegen_config(package.path).get(
        "allow_empty_changelog", False
    ):
        print("All sections are empty. No changes will be made to the changelog.")
        return None

    version_match = re.search(rf"## Release v({Version.PATTERN})", next_changelog)

    if not version_match:
        raise Exception("Version not found in the changelog")

    # Validate the extracted string is spec-compliant; fail loudly otherwise.
    version = str(Version.parse(version_match.group(1)))
    return TagInfo(package=package, version=version, content=next_changelog)


def write_changelog(tag_info: TagInfo) -> None:
    """
    Updates the changelog with a new tag info.
    """
    changelog_path = os.path.join(os.getcwd(), tag_info.package.path, CHANGELOG_FILE_NAME)
    with open(changelog_path, "r") as f:
        changelog = f.read()

    # Add current date to the release header.
    current_date = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")
    content_with_date = re.sub(
        rf"## Release v({Version.PATTERN})",
        rf"## Release v\1 ({current_date})",
        tag_info.content.strip(),
    )

    updated_changelog = re.sub(r"(# Version changelog\n\n)", f"\\1{content_with_date}\n\n\n", changelog)
    gh.add_file(changelog_path, updated_changelog)


def process_package(package: Package) -> TagInfo:
    """
    Processes a package's changelog scaffolding for the release.
    """
    print(f"Processing package {package.name}")
    tag_info = get_next_tag_info(package)

    # If there are no updates, skip.
    if tag_info is None:
        return

    write_changelog(tag_info)
    if _nextchanges_dir() is not None:
        clean_nextchanges(package.path)
    else:
        clean_next_changelog(package.path)
    return tag_info


def find_packages() -> List[Package]:
    """
    Returns all directories which contains a ".package.json" file.
    """
    paths = _find_directories_with_file(PACKAGE_FILE_NAME)
    return [Package(name=get_package_name(path), path=path) for path in paths]


def _find_directories_with_file(target_file: str) -> List[str]:
    root_path = os.getcwd()
    matching_directories = []

    for dirpath, _, filenames in os.walk(root_path):
        if target_file in filenames:
            path = os.path.relpath(dirpath, root_path)
            # If the path is the root directory (e.g., SDK V0), set it to an empty string.
            if path == ".":
                path = ""
            matching_directories.append(path)

    return matching_directories


def is_tag_applied(tag: TagInfo) -> bool:
    """
    Returns whether a tag is already applied in the repository.

    :param tag: The tag to check.
    :return: True if the tag is applied, False otherwise.
    :raises Exception: If the git command fails.
    """
    try:
        # Check if the specific tag exists
        result = subprocess.check_output(["git", "tag", "--list", tag.tag_name()], stderr=subprocess.PIPE, text=True)
        return result.strip() == tag.tag_name()
    except subprocess.CalledProcessError as e:
        # Raise a exception for git command errors
        raise Exception(f"Git command failed: {e.stderr.strip() or e}") from e


def find_last_release_tag(package: Package) -> Optional[str]:
    """
    Returns the most recent ``<package>/v*`` tag in the repository, or
    ``None`` if no such tag exists. Tags are sorted by semver ordering
    (``--sort=-v:refname``) so pre-releases sort below their stable
    counterparts.

    :raises Exception: If the git command fails.
    """
    pattern = f"{package.name}/v*" if package.name else "v*"
    try:
        output = subprocess.check_output(
            ["git", "tag", "--list", pattern, "--sort=-v:refname"],
            stderr=subprocess.PIPE,
            text=True,
        ).strip()
    except subprocess.CalledProcessError as e:
        raise Exception(f"Git command failed: {e.stderr.strip() or e}") from e
    if not output:
        return None
    return output.split("\n")[0].strip()


def has_commits_since_tag(tag: str, path: str) -> bool:
    """
    Returns True iff at least one commit reachable from HEAD but not from
    ``tag`` touches ``path``. Used to detect that a sibling dependency has
    unreleased changes that would ship stale if we tagged a dependent
    without re-tagging the dependency.

    :raises Exception: If the git command fails.
    """
    args = ["git", "log", "--oneline", f"{tag}..HEAD", "--", path or "."]
    try:
        output = subprocess.check_output(args, stderr=subprocess.PIPE, text=True).strip()
    except subprocess.CalledProcessError as e:
        raise Exception(f"Git command failed: {e.stderr.strip() or e}") from e
    return bool(output)


def check_dependency_freshness(tag_infos: List[TagInfo], all_packages: List[Package]) -> None:
    """
    Hard-fails when a package being released depends on a sibling package
    that has unreleased commits since its last tag.

    Why: dependency rewrites (``stage_version_updates``) only fire for
    siblings that are *also* being released. Without this check, releasing
    package_a alone — when package_b has commits since its last tag —
    publishes ``package_a@new`` pinning the *old* package_b artifact, which
    won't have the changes package_a's source depends on. The check is
    commit-based (not changelog-based) so a missing ``NEXT_CHANGELOG.md``
    entry on package_b is still caught.

    No-op when ``.codegen.json`` declares no dependency pattern (legacy
    SDKs without per-package wiring).
    """
    if not tag_infos:
        return

    package_file_path = os.path.join(os.getcwd(), CODEGEN_FILE_NAME)
    with open(package_file_path, "r") as file:
        codegen = json.load(file)

    name_template = codegen.get("dependency_name_template", "")
    dep_patterns = codegen.get("dependency_pattern", {})
    if not name_template or not dep_patterns:
        return

    releasing_paths = {info.package.path for info in tag_infos}
    by_dep_name: Dict[str, Package] = {}
    for pkg in all_packages:
        if not pkg.name:
            continue
        by_dep_name[name_template.replace("$PACKAGE", pkg.name)] = pkg

    issues: List[str] = []
    for info in tag_infos:
        for filename, pattern in dep_patterns.items():
            loc = os.path.join(os.getcwd(), info.package.path, filename)
            if not os.path.exists(loc):
                continue
            with open(loc, "r") as f:
                content = f.read()

            for dep_name, dep_pkg in by_dep_name.items():
                if dep_pkg.path == info.package.path:
                    continue
                if dep_pkg.path in releasing_paths:
                    continue

                # Same regex construction used by ``rewrite_dependencies``,
                # so "is this dep referenced?" matches "would the rewrite
                # touch it?". Keeps the two in lockstep.
                regex = (
                    re.escape(pattern)
                    .replace(re.escape("$DEPENDENCY"), re.escape(dep_name))
                    .replace(re.escape("$VERSION"), Version.PATTERN)
                )
                if not re.search(regex, content):
                    continue

                last_tag = find_last_release_tag(dep_pkg)
                if last_tag is None:
                    # No prior tag means the dep was never released; we
                    # can't reason about staleness. Surface it anyway so
                    # the human resolves it explicitly.
                    issues.append(
                        f"{info.package.name} depends on {dep_pkg.name}, "
                        f"which has never been released. Release "
                        f"{dep_pkg.name} first or include it in this run."
                    )
                    continue
                if has_commits_since_tag(last_tag, dep_pkg.path):
                    issues.append(
                        f"{info.package.name} depends on {dep_pkg.name}, "
                        f"which has commits since {last_tag} but is not "
                        f"being released. Either release {dep_pkg.name} "
                        f"as well, or hold this release until its changes "
                        f"are reverted."
                    )

    if issues:
        raise Exception("Dependency freshness check failed:\n  - " + "\n  - ".join(issues))


def find_last_tags() -> List[TagInfo]:
    """
    Finds the last tags for each package.

    Returns a list of TagInfo objects for each package with a non-None changelog.
    """
    packages = find_packages()

    return [info for info in (get_previous_tag_info(package) for package in packages) if info is not None]


def find_pending_tags() -> List[TagInfo]:
    """
    Finds all tags that are pending to be applied.
    """
    tag_infos = find_last_tags()
    return [tag for tag in tag_infos if not is_tag_applied(tag)]


def generate_commit_message(tag_infos: List[TagInfo]) -> str:
    """
    Generates a commit message for the release.
    """
    if not tag_infos:
        raise Exception("No tag infos provided to generate commit message")

    info = tag_infos[0]
    # Legacy mode for SDKs without per service packaging
    if not info.package.name:
        if len(tag_infos) > 1:
            raise Exception("Multiple packages found in legacy mode")
        return f"[Release] Release v{info.version}\n\n{info.content}"

    # Sort tag_infos by package name for consistency.
    tag_infos.sort(key=lambda info: info.package.name)
    titles = ", ".join(f"{info.package.name}/v{info.version}" for info in tag_infos)
    body = "\n\n".join(f"## {info.package.name}/v{info.version}\n\n{info.content}" for info in tag_infos)
    return f"[Release] {titles}\n\n{body}"


def push_changes(tag_infos: List[TagInfo]) -> None:
    """Pushes changes to the remote repository after handling possible merge conflicts."""

    commit_message = generate_commit_message(tag_infos)

    # Create the release metadata file
    file_name = os.path.join(os.getcwd(), ".release_metadata.json")
    metadata = {"timestamp": datetime.now(tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S%z")}
    content = json.dumps(metadata, indent=4)
    gh.add_file(file_name, content)

    gh.commit_and_push(commit_message)


def reset_repository(hash: Optional[str] = None) -> None:
    """
    Reset git to the specified commit. Defaults to HEAD.

    :param hash: The commit hash to reset to. If None, it resets to HEAD.
    """
    # Fetch the latest changes from the remote repository.
    subprocess.run(["git", "fetch"])

    # Determine the commit hash (default to the release branch's remote
    # head if none is provided). ``_release_branch`` is ``main`` unless
    # ``DECO_TAGGING_REF`` selects a branch release.
    commit_hash = hash or f"origin/{_release_branch()}"

    # ``git reset --hard`` must land before ``gh.reset(None)``, since
    # ``gh.reset(None)`` reads ``git rev-parse HEAD`` to anchor
    # ``self.sha`` to the local working tree.
    subprocess.run(["git", "reset", "--hard", commit_hash], check=True)
    gh.reset(hash)


def retry_function(
    func: Callable[[], List[TagInfo]], cleanup: Callable[[], None], max_attempts: int = 5, delay: int = 5
) -> List[TagInfo]:
    """
    Calls a function call up to `max_attempts` times if an exception occurs.

    :param func: The function to call.
    :param cleanup: Cleanup function in between retries
    :param max_attempts: The maximum number of retries.
    :param delay: The delay between retries in seconds.
    :return: The return value of the function, or None if all retries fail.
    """
    attempts = 0
    while attempts <= max_attempts:
        try:
            return func()  # Call the function
        except MainAdvancedError:
            # Permanent failure: another commit landed on main during
            # this run, so the local tree is stale. Retrying with the
            # same stale tree would just hit the same mismatch — only
            # a fresh workflow run against the new main can recover.
            raise
        except Exception as e:
            attempts += 1
            print(f"Attempt {attempts} failed: {e}")
            if attempts < max_attempts:
                time.sleep(delay)  # Wait before retrying
                cleanup()
            else:
                print("All retry attempts failed.")
                raise e  # Re-raise the exception after max retries


def update_changelogs(selected_packages: List[Package], all_packages: List[Package]) -> List[TagInfo]:
    """
    Updates changelogs and pushes the commits.

    ``selected_packages`` are the packages whose ``NEXT_CHANGELOG.md`` is
    consulted to decide what gets released this run. ``all_packages`` is
    the full repo inventory used for cross-package dep rewrites.

    The freshness check is deliberately *not* called here. ``process``
    runs it before entering the retry loop so a freshness violation
    fails fast — the check is deterministic against the same git state,
    so wrapping it in retry would just delay the same failure five
    times.
    """
    tag_infos = [info for info in (process_package(package) for package in selected_packages) if info is not None]
    # If any package was changed, stage version updates and push.
    if tag_infos:
        stage_version_updates(tag_infos, all_packages)
        push_changes(tag_infos)
    return tag_infos


def preview_tag_infos(packages: List[Package]) -> List[TagInfo]:
    """
    Read-only sibling of ``process_package``: returns the TagInfos that
    would be released for ``packages`` without writing any changelog
    edits. ``process`` calls this before the retry loop so the freshness
    check has a snapshot to validate against. ``process_package`` will
    re-derive the same TagInfos when ``update_changelogs`` runs; the
    duplication is just a couple of NEXT_CHANGELOG.md reads.
    """
    return [info for info in (get_next_tag_info(package) for package in packages) if info is not None]


def order_tag_infos_by_dependency(tag_infos: List[TagInfo]) -> List[TagInfo]:
    """
    Returns ``tag_infos`` in topological order: every package appears
    after every sibling it depends on.
    """
    if not tag_infos:
        return list(tag_infos)

    if any(not info.package.name for info in tag_infos) and len(tag_infos) > 1:
        raise Exception("Multiple packages found in legacy mode")

    package_file_path = os.path.join(os.getcwd(), CODEGEN_FILE_NAME)
    with open(package_file_path, "r") as file:
        codegen = json.load(file)

    name_template = codegen.get("dependency_name_template", "")
    dep_patterns = codegen.get("dependency_pattern", {})
    if not name_template or not dep_patterns:
        return list(tag_infos)

    by_dep_name: Dict[str, TagInfo] = {
        name_template.replace("$PACKAGE", info.package.name): info for info in tag_infos if info.package.name
    }

    # Adjacency: path -> set of paths it depends on (within tag_infos).
    deps: Dict[str, set] = {info.package.path: set() for info in tag_infos}
    for info in tag_infos:
        for filename, pattern in dep_patterns.items():
            loc = os.path.join(os.getcwd(), info.package.path, filename)
            if not os.path.exists(loc):
                continue
            with open(loc, "r") as f:
                content = f.read()
            for dep_name, dep_info in by_dep_name.items():
                if dep_info.package.path == info.package.path:
                    continue
                regex = (
                    re.escape(pattern)
                    .replace(re.escape("$DEPENDENCY"), re.escape(dep_name))
                    .replace(re.escape("$VERSION"), Version.PATTERN)
                )
                if re.search(regex, content):
                    deps[info.package.path].add(dep_info.package.path)

    # Stable topological sort: at each step, emit every node whose deps
    # are already emitted, alphabetically by package name. Ties broken
    # alphabetically so the manifest is reproducible across runs.
    emitted: set = set()
    ordered: List[TagInfo] = []
    while len(ordered) < len(tag_infos):
        ready = sorted(
            (
                info
                for info in tag_infos
                if info.package.path not in emitted and deps[info.package.path].issubset(emitted)
            ),
            key=lambda info: info.package.name,
        )
        if not ready:
            remaining = [info.package.name for info in tag_infos if info.package.path not in emitted]
            raise Exception(f"Cyclic dependency detected among packages: {remaining}")
        for info in ready:
            ordered.append(info)
            emitted.add(info.package.path)
    return ordered


def push_tags(tag_infos: List[TagInfo]) -> None:
    """
    Creates and pushes tags to the repository.

    Tags are emitted in topological order — dependencies before
    dependents — so downstream publishing pipelines reading
    ``created_tags.json`` can walk it sequentially without re-deriving
    the dependency graph. See ``order_tag_infos_by_dependency``.

    As a side effect, writes the names of successfully created tags to
    ``./created_tags.json`` so that workflows triggering this script can
    discover what was produced (the GitHub Actions workflow uploads this
    file as the ``created-tags`` artifact).

    Schema:
        {"tags": ["service-a/v1.2.3", "service-b/v0.4.0"]}

    The manifest is written even if tag creation fails partway through:
    tags that succeeded before the failure are flushed before the
    exception is re-raised, so recovery-mode runs still surface their
    output.
    """
    tag_infos = order_tag_infos_by_dependency(tag_infos)
    created: List[str] = []
    try:
        for tag_info in tag_infos:
            gh.tag(tag_info.tag_name(), tag_info.content)
            created.append(tag_info.tag_name())
    finally:
        manifest_path = os.path.join(os.getcwd(), CREATED_TAGS_FILE_NAME)
        with open(manifest_path, "w") as f:
            json.dump({"tags": created}, f)


def run_command(command: List[str]) -> str:
    """
    Runs a command and returns the output
    """
    output = subprocess.check_output(command)
    print(f'Running command: {" ".join(command)}')
    return output.decode()


def pull_last_release_commit() -> None:
    """
    Reset the repository to the last release.
    Uses commit for last change to .release_metadata.json, since it's only updated on releases.
    """
    commit_hash = subprocess.check_output(
        ["git", "log", "-n", "1", "--format=%H", "--", ".release_metadata.json"], text=True
    ).strip()

    # If no commit is found, raise an exception
    if not commit_hash:
        raise ValueError("No commit found for .release_metadata.json")

    # Reset the repository to the commit
    reset_repository(commit_hash)


def _build_arg_parser() -> argparse.ArgumentParser:
    """
    Builds the CLI parser. Shared by ``get_packages_from_args`` and
    ``args_request_preview`` so both accept the same flags — parsing either one
    with an unknown flag present would otherwise error.
    """
    parser = argparse.ArgumentParser(description="Update changelogs and tag the release.")
    parser.add_argument(
        "--package",
        "-p",
        type=str,
        default="",
        help="Comma-separated list of packages to tag. Leave empty to tag all packages with pending releases.",
    )
    parser.add_argument(
        "--preview",
        action="store_true",
        help=(
            "Print the changelog section the next release would add, then exit "
            "(no writes, no network). Only supported in nextchanges mode."
        ),
    )
    return parser


def get_packages_from_args() -> List[str]:
    """
    Retrieves the list of packages to tag.

        python3 ./tagging.py --package <name>              # single package
        python3 ./tagging.py --package <name1>,<name2>     # multiple packages

    Returns an empty list when --package is omitted, which means all packages
    with pending releases will be tagged.
    """
    args = _build_arg_parser().parse_args()
    return [name.strip() for name in args.package.split(",") if name.strip()]


def _select_packages(all_packages: List[Package]) -> List[Package]:
    """
    Narrow ``all_packages`` to the ``--package`` selection, or return all of
    them when ``--package`` is omitted. Shared by ``process`` and ``preview`` so
    the read-only preview covers exactly the release scope the same CLI
    arguments would execute — ``tagging.py --package foo --preview`` previews
    only ``foo``, not every package in the repo.
    """
    package_names = get_packages_from_args()
    if not package_names:
        return all_packages
    return [package for package in all_packages if package.name in package_names]


def args_request_preview() -> bool:
    """Returns whether ``--preview`` was passed."""
    return _build_arg_parser().parse_args().preview


def preview() -> None:
    """
    Print the ``## Release vX.Y.Z`` section(s) the next release would prepend to
    CHANGELOG.md, rendered from the current ``<dir>/`` fragments — without
    touching git, GitHub, or any file. Mirrors ``write_changelog``'s date stamp
    so the output matches what would land. Read-only: safe to run anytime, no
    credentials. Only meaningful in nextchanges mode.
    """
    if _nextchanges_dir() is None:
        raise Exception(f"--preview requires nextchanges mode ({NEXTCHANGES_DIR_ENV} must be set).")

    current_date = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")
    printed = False
    for package in _select_packages(find_packages()):
        tag_info = get_next_tag_info(package)
        if tag_info is None:
            continue
        dated = re.sub(
            rf"## Release v({Version.PATTERN})",
            rf"## Release v\1 ({current_date})",
            tag_info.content.strip(),
        )
        print(dated)
        printed = True
    if not printed:
        print(f"No {_nextchanges_dir()}/ entries — the next release would add no changelog section.")


def init_github():
    token = os.environ["GITHUB_TOKEN"]
    repo_name = os.environ["GITHUB_REPOSITORY"]
    g = Github(auth=Auth.Token(token))
    repo = g.get_repo(repo_name)
    global gh
    gh = GitHubRepo(repo)


def process():
    """
    Main entry point for tagging process.

    Tagging process consist of multiple steps:
    * For each package, update the corresponding CHANGELOG.md file based on the contents of NEXT_CHANGELOG.md file
    * If any package has been updated, commit and push the changes.
    * Apply and push the new tags matching the version.

    If a specific pagkage is provided as a parameter, only that package will be tagged.

    If any tag are pending from an early process, it will skip updating the CHANGELOG.md files and only apply the tags.
    """

    package_names = get_packages_from_args()
    pending_tags = find_pending_tags()

    # pending_tags is non-empty only when the tagging process previously failed or interrupted.
    # We must complete the interrupted tagging process before starting a new one to avoid inconsistent states and missing changelog entries.
    # Therefore, we don't support specifying packages until the previously started process has been successfully completed.
    if pending_tags and package_names:
        pending_packages = [tag.package.name for tag in pending_tags]
        raise Exception(f"Cannot release packages {package_names}. Pending release for {pending_packages}")

    if pending_tags:
        print("Found pending tags from previous executions, entering recovery mode.")
        pull_last_release_commit()
        push_tags(pending_tags)
        return

    all_packages = find_packages()
    # If packages are specified as an argument, only release those — but
    # dep rewrites and the freshness check still operate over the full
    # set. Shared with preview() so both scope identically.
    selected_packages = _select_packages(all_packages)

    # Run the freshness check against a read-only preview before the
    # retry loop, since the check is deterministic. A freshness
    # violation fails the run immediately, with no commits, no tags, no
    # retry storm.
    check_dependency_freshness(preview_tag_infos(selected_packages), all_packages)

    pending_tags = retry_function(
        func=lambda: update_changelogs(selected_packages, all_packages),
        cleanup=reset_repository,
    )
    push_tags(pending_tags)


def validate_git_root():
    """
    Validate that the script is run from the root of the repository.
    """
    repo_root = subprocess.check_output(["git", "rev-parse", "--show-toplevel"]).strip().decode("utf-8")
    current_dir = subprocess.check_output(["pwd"]).strip().decode("utf-8")
    if repo_root != current_dir:
        raise Exception("Please run this script from the root of the repository.")


def main():
    validate_git_root()
    # Preview is read-only: no GitHub credentials, no network, no commits.
    if args_request_preview():
        preview()
        return
    init_github()
    process()


if __name__ == "__main__":
    main()