Timmy-time-dashboard/src/self_coding/self_modify/loop.py

"""Self-modification loop — branch → edit → test → commit/revert.

Timmy's self-coding capability, restored after deletion in
Operation Darling Purge (commit 584eeb679e88).

## Cycle
1. **Branch** — create ``self-modify/<slug>`` from ``main``
2. **Edit**   — apply the proposed change (patch string or callable)
3. **Test**   — run ``pytest tests/ -x -q``; never commit on failure
4. **Commit** — stage and commit on green; revert branch on red
5. **PR**     — open a Gitea pull request (requires no direct push to main)

## Guards
- Never push directly to ``main`` or ``master``
- All changes land via PR (enforced by ``_guard_branch``)
- Test gate is mandatory; ``skip_tests=True`` is for unit-test use only
- Commits only happen when ``pytest tests/ -x -q`` exits 0

## Usage::

    from self_coding.self_modify.loop import SelfModifyLoop

    loop = SelfModifyLoop()
    result = await loop.run(
        slug="add-hello-tool",
        description="Add hello() convenience tool",
        edit_fn=my_edit_function,  # callable(repo_root: str) -> None
    )
    if result.success:
        print(f"PR: {result.pr_url}")
    else:
        print(f"Failed: {result.error}")
"""

from __future__ import annotations

import logging
import subprocess
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path

from config import settings

logger = logging.getLogger(__name__)

# Branches that must never receive direct commits
_PROTECTED_BRANCHES = frozenset({"main", "master", "develop"})

# Test command used as the commit gate
_TEST_COMMAND = ["pytest", "tests/", "-x", "-q", "--tb=short"]

# Max time (seconds) to wait for the test suite
_TEST_TIMEOUT = 300


@dataclass
class LoopResult:
    """Result from one self-modification cycle."""

    success: bool
    branch: str = ""
    commit_sha: str = ""
    pr_url: str = ""
    pr_number: int = 0
    test_output: str = ""
    error: str = ""
    elapsed_ms: float = 0.0
    metadata: dict = field(default_factory=dict)


class SelfModifyLoop:
    """Orchestrate branch → edit → test → commit/revert → PR.

    Args:
        repo_root: Absolute path to the git repository (defaults to
                   ``settings.repo_root``).
        remote:    Git remote name (default ``origin``).
        base_branch: Branch to fork from and target for the PR
                     (default ``main``).
    """

    def __init__(
        self,
        repo_root: str | None = None,
        remote: str = "origin",
        base_branch: str = "main",
    ) -> None:
        self._repo_root = Path(repo_root or settings.repo_root)
        self._remote = remote
        self._base_branch = base_branch

    # ── public ──────────────────────────────────────────────────────────────

    async def run(
        self,
        slug: str,
        description: str,
        edit_fn: Callable[[str], None],
        issue_number: int | None = None,
        skip_tests: bool = False,
    ) -> LoopResult:
        """Execute one full self-modification cycle.

        Args:
            slug:         Short identifier used for the branch name
                          (e.g. ``"add-hello-tool"``).
            description:  Human-readable description for commit message
                          and PR body.
            edit_fn:      Callable that receives the repo root path (str)
                          and applies the desired code changes in-place.
            issue_number: Optional Gitea issue number to reference in PR.
            skip_tests:   If ``True``, skip the test gate (unit-test use
                          only — never use in production).

        Returns:
            :class:`LoopResult` describing the outcome.
        """
        start = time.time()
        branch = f"self-modify/{slug}"

        try:
            self._guard_branch(branch)
            self._checkout_base()
            self._create_branch(branch)

            try:
                edit_fn(str(self._repo_root))
            except Exception as exc:
                self._revert_branch(branch)
                return LoopResult(
                    success=False,
                    branch=branch,
                    error=f"edit_fn raised: {exc}",
                    elapsed_ms=self._elapsed(start),
                )

            if not skip_tests:
                test_output, passed = self._run_tests()
                if not passed:
                    self._revert_branch(branch)
                    return LoopResult(
                        success=False,
                        branch=branch,
                        test_output=test_output,
                        error="Tests failed — branch reverted",
                        elapsed_ms=self._elapsed(start),
                    )
            else:
                test_output = "(tests skipped)"

            sha = self._commit_all(description)
            self._push_branch(branch)

            pr = self._create_pr(
                branch=branch,
                description=description,
                test_output=test_output,
                issue_number=issue_number,
            )

            return LoopResult(
                success=True,
                branch=branch,
                commit_sha=sha,
                pr_url=pr.html_url if pr else "",
                pr_number=pr.number if pr else 0,
                test_output=test_output,
                elapsed_ms=self._elapsed(start),
            )

        except Exception as exc:
            logger.warning("Self-modify loop failed: %s", exc)
            return LoopResult(
                success=False,
                branch=branch,
                error=str(exc),
                elapsed_ms=self._elapsed(start),
            )

    # ── private helpers ──────────────────────────────────────────────────────

    @staticmethod
    def _elapsed(start: float) -> float:
        return (time.time() - start) * 1000

    def _git(self, *args: str, check: bool = True) -> subprocess.CompletedProcess:
        """Run a git command in the repo root."""
        cmd = ["git", *args]
        logger.debug("git %s", " ".join(args))
        return subprocess.run(
            cmd,
            cwd=str(self._repo_root),
            capture_output=True,
            text=True,
            check=check,
        )

    def _guard_branch(self, branch: str) -> None:
        """Raise if the target branch is a protected branch name."""
        if branch in _PROTECTED_BRANCHES:
            raise ValueError(
                f"Refusing to operate on protected branch '{branch}'. "
                "All self-modifications must go via PR."
            )

    def _checkout_base(self) -> None:
        """Checkout the base branch and pull latest."""
        self._git("checkout", self._base_branch)
        # Best-effort pull; ignore failures (e.g. no remote configured)
        self._git("pull", self._remote, self._base_branch, check=False)

    def _create_branch(self, branch: str) -> None:
        """Create and checkout a new branch, deleting an old one if needed."""
        # Delete local branch if it already exists (stale prior attempt)
        self._git("branch", "-D", branch, check=False)
        self._git("checkout", "-b", branch)
        logger.info("Created branch: %s", branch)

    def _revert_branch(self, branch: str) -> None:
        """Checkout base and delete the failed branch."""
        try:
            self._git("checkout", self._base_branch, check=False)
            self._git("branch", "-D", branch, check=False)
            logger.info("Reverted and deleted branch: %s", branch)
        except Exception as exc:
            logger.warning("Failed to revert branch %s: %s", branch, exc)

    def _run_tests(self) -> tuple[str, bool]:
        """Run the test suite. Returns (output, passed)."""
        logger.info("Running test suite: %s", " ".join(_TEST_COMMAND))
        try:
            result = subprocess.run(
                _TEST_COMMAND,
                cwd=str(self._repo_root),
                capture_output=True,
                text=True,
                timeout=_TEST_TIMEOUT,
            )
            output = (result.stdout + "\n" + result.stderr).strip()
            passed = result.returncode == 0
            logger.info(
                "Test suite %s (exit %d)", "PASSED" if passed else "FAILED", result.returncode
            )
            return output, passed
        except subprocess.TimeoutExpired:
            msg = f"Test suite timed out after {_TEST_TIMEOUT}s"
            logger.warning(msg)
            return msg, False
        except FileNotFoundError:
            msg = "pytest not found on PATH"
            logger.warning(msg)
            return msg, False

    def _commit_all(self, message: str) -> str:
        """Stage all changes and create a commit. Returns the new SHA."""
        self._git("add", "-A")
        self._git("commit", "-m", message)
        result = self._git("rev-parse", "HEAD")
        sha = result.stdout.strip()
        logger.info("Committed: %s  sha=%s", message[:60], sha[:12])
        return sha

    def _push_branch(self, branch: str) -> None:
        """Push the branch to the remote."""
        self._git("push", "-u", self._remote, branch)
        logger.info("Pushed branch: %s -> %s", branch, self._remote)

    def _create_pr(
        self,
        branch: str,
        description: str,
        test_output: str,
        issue_number: int | None,
    ):
        """Open a Gitea PR. Returns PullRequest or None on failure."""
        from self_coding.gitea_client import GiteaClient

        client = GiteaClient()

        issue_ref = f"\n\nFixes #{issue_number}" if issue_number else ""
        test_section = (
            f"\n\n## Test results\n```\n{test_output[:2000]}\n```"
            if test_output and test_output != "(tests skipped)"
            else ""
        )

        body = (
            f"## Summary\n{description}"
            f"{issue_ref}"
            f"{test_section}"
            "\n\n🤖 Generated by Timmy's self-modification loop"
        )

        return client.create_pull_request(
            title=f"[self-modify] {description[:60]}",
            body=body,
            head=branch,
            base=self._base_branch,
        )