orchestrata_ultimate.session.controller._changelog

Reference › orchestrata_ultimate › session › controller › _changelog

ChangelogMixin updates CHANGELOG.md with the achievements of a closing session, for the SessionController. _find_changelog locates the file; if there is none, nothing happens. _categorize_achievements sorts each achievement into a section by its leading word: add, implement or create as Added (rendered as a Feat heading), fix or resolve as Fixed, remove or delete as Removed, deprecate as Deprecated, security or harden as Security, and everything else as Changed. _splice_changelog inserts the entries into the existing sections of the file for that session number (with _splice_category_lines and _append_remaining_categories as helpers), and _update_changelog writes the file only if the content changed and then commits it with _commit_changelog. It returns whether the file was updated.

At a glance

Class diagram

Dependency diagram

API

Classes

ChangelogMixin

Line 16

CHANGELOG.md update, categorization, and splicing for sessions.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/controller/_changelog.py

# mypy: ignore-errors
"""ChangelogMixin — CHANGELOG.md categorization and splicing for SessionController."""

import logging
import os
import subprocess
from collections.abc import Mapping
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.models import SessionRecord

logger = logging.getLogger(__name__)


class ChangelogMixin:
    """CHANGELOG.md update, categorization, and splicing for sessions."""

    # Provided by SessionController.__init__ (composed via multiple inheritance).
    root: Path

    _CATEGORY_PREFIXES: dict[str, tuple[str, ...]] = {
        "Added": ("add", "implement", "create", "introduce", "new"),
        "Fixed": ("fix", "resolve", "repair", "correct"),
        "Removed": ("remove", "delete", "drop"),
        "Deprecated": ("deprecate",),
        "Security": ("security", "harden", "vulnerability"),
    }
    _DEFAULT_CATEGORY = "Changed"

    # Map canonical categories to the section headers used by this project's
    # CHANGELOG (e.g. Added -> Feat, Fixed -> Fix).
    _HEADER_FOR_CATEGORY: dict[str, str] = {
        "Added": "Feat",
        "Fixed": "Fix",
        "Removed": "Removed",
        "Deprecated": "Deprecated",
        "Security": "Security",
        "Changed": "Changed",
    }

    # Map section headers found in the wild back to canonical categories.
    _CATEGORY_FOR_HEADER: dict[str, str] = {
        "Feat": "Added",
        "Feature": "Added",
        "Features": "Added",
        "Added": "Added",
        "Fix": "Fixed",
        "Fixed": "Fixed",
        "Refactor": "Changed",
        "Perf": "Changed",
        "Performance": "Changed",
    }

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _update_changelog(self, session: SessionRecord) -> bool:
        """Update CHANGELOG.md with session achievements.

        Categorizes entries by keyword prefix:
        - Add/Implement/Create -> Added (rendered as ### Feat)
        - Fix/Resolve -> Fixed (rendered as ### Fix)
        - Remove/Delete -> Removed
        - Deprecate -> Deprecated
        - Security/Harden -> Security
        - Everything else -> Changed

        Args:
            session: The completed session.

        Returns:
            True if CHANGELOG was updated, False otherwise.
        """
        changelog = self._find_changelog()
        if changelog is None:
            return False

        entries = self._categorize_achievements(session)
        if not entries:
            return False

        content = changelog.read_text(encoding="utf-8")
        updated = self._splice_changelog(content, entries, session.seq)
        if updated == content:
            return False

        changelog.write_text(updated, encoding="utf-8")
        self._commit_changelog(changelog, session.seq)
        return True

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _commit_changelog(self, changelog: Path, seq: int) -> None:
        """Stage and commit the updated CHANGELOG.md.

        Args:
            changelog: Path to CHANGELOG.md.
            seq: Session sequence number for commit message.
        """
        subprocess.run(
            ["git", "add", str(changelog)],
            cwd=self.root,
            capture_output=True,
        )
        subprocess.run(
            ["git", "commit", "-m", f"Update CHANGELOG.md for session {seq:03d}"],
            cwd=self.root,
            capture_output=True,
        )

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _find_changelog(self, env: Mapping[str, str] | None = None) -> Path | None:
        """Find CHANGELOG.md in repo root.

        Args:
            env: Optional environment mapping for the spawned subprocess.
                Defaults to ``os.environ``.
        """
        runtime_env = os.environ if env is None else dict(env)
        try:
            result = subprocess.run(
                ["git", "rev-parse", "--show-toplevel"],
                cwd=self.root,
                capture_output=True,
                text=True,
                env=runtime_env,
            )
            if result.returncode == 0:
                repo_root = Path(result.stdout.strip())
                changelog = repo_root / "CHANGELOG.md"
                if changelog.exists():
                    return changelog
        except FileNotFoundError as exc:
            logger.debug("git not found for changelog: %s", exc)
        return None

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _categorize_achievements(
        self,
        session: SessionRecord,
    ) -> dict[str, list[str]]:
        """Categorize achievements into changelog sections."""
        categories: dict[str, list[str]] = {}

        for item in session.achievements:
            category = self._DEFAULT_CATEGORY
            lower = item.lower().lstrip("- ")
            for cat, prefixes in self._CATEGORY_PREFIXES.items():
                if any(lower.startswith(p) for p in prefixes):
                    category = cat
                    break
            categories.setdefault(category, []).append(item)

        return categories

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _header_for_category(self, category: str) -> str:
        return self._HEADER_FOR_CATEGORY.get(category, category)

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _category_for_header(self, header: str) -> str:
        return self._CATEGORY_FOR_HEADER.get(header, header)

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _splice_changelog(
        self,
        content: str,
        entries: dict[str, list[str]],
        seq: int,
    ) -> str:
        """Splice categorized entries into changelog content."""
        lines = content.split("\n")
        result: list[str] = []
        i = 0
        in_unreleased = False
        remaining = dict(entries)

        while i < len(lines):
            line = lines[i]

            if line.strip() in ("## [Unreleased]", "## Unreleased"):
                in_unreleased = True
                result.append(line)
                i += 1
                continue

            if (
                in_unreleased
                and line.startswith("## ")
                and line.strip() not in ("## [Unreleased]", "## Unreleased")
            ):
                self._append_remaining_categories(result, remaining, seq)
                remaining.clear()
                if result and result[-1].strip():
                    result.append("")
                in_unreleased = False

            if in_unreleased and line.startswith("### "):
                header = line[4:].strip()
                category = self._category_for_header(header)
                result.append(line)
                i += 1
                if category in remaining:
                    i = self._splice_category_lines(
                        lines, i, result, remaining[category], seq
                    )
                    del remaining[category]
                continue

            result.append(line)
            i += 1

        if in_unreleased and remaining:
            self._append_remaining_categories(result, remaining, seq)

        return "\n".join(result)

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _append_remaining_categories(
        self,
        result: list[str],
        remaining: dict[str, list[str]],
        seq: int,
    ) -> None:
        """Append any categories that did not already exist in the Unreleased block."""
        for category, items in remaining.items():
            if result and result[-1].strip():
                result.append("")
            result.append(f"### {self._header_for_category(category)}")
            result.append("")
            for item in items:
                result.append(f"- {item} (Session {seq:03d})")

    @traces(spec="020", story="US-001", reqs=["REQ-020-020"])
    def _splice_category_lines(
        self,
        lines: list[str],
        i: int,
        result: list[str],
        new_items: list[str],
        seq: int,
    ) -> int:
        """Merge new_items into an existing changelog category block.

        Args:
            lines: All changelog lines.
            i: Current index (first line after the ### header).
            result: Accumulated output lines (mutated in place).
            new_items: New achievement strings to insert.
            seq: Session sequence number for bullet suffix.

        Returns:
            Updated line index after consuming the category block.
        """
        # Skip the blank line(s) after the header and any "- Nothing yet" placeholder.
        while i < len(lines) and lines[i].strip() == "":
            i += 1
        if i < len(lines) and lines[i].strip() == "- Nothing yet":
            i += 1
            while i < len(lines) and lines[i].strip() == "":
                i += 1

        # Ensure one blank line between the header and the bullet list.
        if result and result[-1].strip():
            result.append("")

        existing: set[str] = set()
        while i < len(lines) and lines[i].startswith("- "):
            existing.add(lines[i])
            result.append(lines[i])
            i += 1

        for item in new_items:
            bullet = f"- {item} (Session {seq:03d})"
            if bullet not in existing:
                result.append(bullet)

        # Keep a blank line between this category and the next header/block.
        if (
            i < len(lines)
            and result
            and result[-1].strip()
            and lines[i].strip()
            and (lines[i].startswith("### ") or lines[i].startswith("## "))
        ):
            result.append("")

        return i