orchestrata_ultimate.session.controller._git

Reference › orchestrata_ultimate › session › controller › _git

GitOpsMixin gives the SessionController its git operations at session close. _commit_and_push_changes stages everything and commits with the message Session NNN: <purpose> when there are tracked changes (untracked files alone do not count); a failed commit is logged as a warning. It pushes to the current branch only when the controller's root was supplied explicitly (RUL-011): a controller whose root was inferred may commit locally but never publishes. _create_session_tag creates an annotated tag named session-NNN-YYYY-MM-DD-HHMM, _push_session_tag pushes it, _current_branch reads the branch and _check_git_clean reports whether the tree is clean. Every method takes an optional environment mapping, returns quietly when git is not installed, and reports success as a boolean where it can.

At a glance

Class diagram

Dependency diagram

API

Classes

GitOpsMixin

Line 32

Git commit, push, tag, and cleanliness operations for sessions.

Functions

Source

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

# mypy: ignore-errors
"""GitOpsMixin — git commit/push/tag operations 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__)


@traces(spec="020", story="US-001", reqs=["REQ-020-021", "REQ-020-022"])
def _git_env(env: Mapping[str, str] | None = None) -> dict[str, str]:
    """Return a copy of ``env`` (or ``os.environ``) with ``GITLAB_TOKEN`` removed.

    ``GITLAB_TOKEN`` overrides glab's stored OAuth token and causes HTTP
    Basic auth failures during ``git push origin``. The token is kept in
    the shell for other operations (e.g. GitLab API calls), but git
    operations invoked by the session CLI must not see it.

    Args:
        env: Optional environment mapping. Defaults to ``os.environ``.
    """
    source = os.environ if env is None else dict(env)
    return {k: v for k, v in source.items() if k != "GITLAB_TOKEN"}


class GitOpsMixin:
    """Git commit, push, tag, and cleanliness operations for sessions."""

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

    @traces(spec="020", story="US-001", reqs=["REQ-020-022"])
    def _commit_and_push_changes(
        self, session: SessionRecord, env: Mapping[str, str] | None = None
    ) -> None:
        """Stage, commit, and push all code changes before tagging.

        Commits with message ``Session NNN: <purpose>``.  Skips if working
        directory is already clean (only untracked files allowed).

        Args:
            session: The session being closed.
            env: Optional environment mapping for spawned subprocesses.
                Defaults to ``os.environ``.
        """
        runtime_env = os.environ if env is None else dict(env)
        try:
            # Check if there are changes to commit (excluding untracked)
            status = subprocess.run(
                ["git", "status", "--porcelain"],
                cwd=self.root,
                capture_output=True,
                text=True,
                env=runtime_env,
            )
            if status.returncode != 0:
                return
            lines = [
                ln
                for ln in status.stdout.strip().split("\n")
                if ln and not ln.startswith("??")
            ]
            if not lines:
                return  # Nothing to commit

            # Stage everything
            subprocess.run(
                ["git", "add", "-A"],
                cwd=self.root,
                capture_output=True,
                env=runtime_env,
            )

            # Commit
            msg = f"Session {session.seq:03d}: {session.purpose}"
            commit = subprocess.run(
                ["git", "commit", "-m", msg],
                cwd=self.root,
                capture_output=True,
                text=True,
                env=runtime_env,
            )
            if commit.returncode != 0:
                logger.warning("Session auto-commit failed: %s", commit.stderr.strip())
                return

            # Push to current branch -- only when the caller vouched for
            # this root (fix/session-root, RUL-011). A controller whose root
            # was inferred (not explicitly supplied) may commit locally but
            # must never publish that commit.
            if not self.root_explicit:
                logger.info(
                    "Session auto-push skipped: controller root was not "
                    "explicitly supplied (fix/session-root, RUL-011)."
                )
                return
            branch = self._current_branch()
            if branch:
                push = subprocess.run(
                    ["git", "push", "origin", branch],
                    cwd=self.root,
                    capture_output=True,
                    text=True,
                    env=_git_env(runtime_env),
                )
                if push.returncode != 0:
                    logger.warning("Session auto-push failed: %s", push.stderr.strip())
        except FileNotFoundError:
            logger.debug("git not found; skipping auto-commit/push")

    @traces(spec="020", story="US-001", reqs=["REQ-020-022"])
    def _current_branch(self) -> str | None:
        """Return the current git branch name, or None on failure."""
        try:
            result = subprocess.run(
                ["git", "rev-parse", "--abbrev-ref", "HEAD"],
                cwd=self.root,
                capture_output=True,
                text=True,
            )
            if result.returncode == 0:
                return result.stdout.strip()
        except FileNotFoundError:
            logger.debug("git not found; cannot determine current branch")
        return None

    @traces(spec="020", story="US-001", reqs=["REQ-020-019"])
    def _create_session_tag(
        self, session: SessionRecord, env: Mapping[str, str] | None = None
    ) -> bool:
        """Create git tag for completed session.

        Args:
            session: The session that was completed.
            env: Optional environment mapping for the spawned subprocess.
                Defaults to ``os.environ``.

        Returns:
            True if tag was created successfully, False otherwise.
        """
        # Format: session-XXX-YYYY-MM-DD-HHMM
        tag_name = f"session-{session.seq:03d}-{session.when.strftime('%Y-%m-%d-%H%M')}"
        tag_message = f"Session {session.seq:03d}: {session.purpose}"
        runtime_env = os.environ if env is None else dict(env)

        try:
            result = subprocess.run(
                ["git", "tag", "-a", tag_name, "-m", tag_message],
                cwd=self.root,
                capture_output=True,
                text=True,
                env=runtime_env,
            )
            return result.returncode == 0
        except FileNotFoundError:
            return False

    @traces(spec="020", story="US-001", reqs=["REQ-020-021"])
    def _push_session_tag(
        self, session: SessionRecord, env: Mapping[str, str] | None = None
    ) -> bool:
        """Push session tag to origin.

        Args:
            session: The completed session.
            env: Optional environment mapping for the spawned subprocess.
                Defaults to ``os.environ``.

        Returns:
            True if push succeeded, False otherwise. Also False (no
            subprocess spawned) when the controller's root was not
            explicitly supplied -- see ``_commit_and_push_changes``.
        """
        if not self.root_explicit:
            logger.info(
                "Session tag push skipped: controller root was not "
                "explicitly supplied (fix/session-root, RUL-011)."
            )
            return False
        tag_name = f"session-{session.seq:03d}-{session.when.strftime('%Y-%m-%d-%H%M')}"
        runtime_env = os.environ if env is None else dict(env)
        try:
            result = subprocess.run(
                ["git", "push", "origin", tag_name],
                cwd=self.root,
                capture_output=True,
                text=True,
                env=_git_env(runtime_env),
            )
            return result.returncode == 0
        except FileNotFoundError:
            return False

    def _check_git_clean(self, env: Mapping[str, str] | None = None) -> bool:
        """Check if git working directory is clean (only untracked allowed).

        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", "status", "--porcelain"],
                cwd=self.root,
                capture_output=True,
                text=True,
                env=runtime_env,
            )
            output = result.stdout.strip()
            if not output:
                return True
            lines = output.split("\n")
            return all(line.startswith("??") for line in lines if line)
        except FileNotFoundError:
            return True