From 003e3883fb8aff3728b12bb4a166fddc5a308aad Mon Sep 17 00:00:00 2001 From: "Claude (Opus 4.6)" Date: Tue, 24 Mar 2026 01:40:16 +0000 Subject: [PATCH] [claude] Restore self-modification loop (#983) (#1270) Co-authored-by: Claude (Opus 4.6) Co-committed-by: Claude (Opus 4.6) --- src/self_coding/__init__.py | 7 + src/self_coding/gitea_client.py | 129 +++++++++ src/self_coding/self_modify/__init__.py | 1 + src/self_coding/self_modify/loop.py | 301 ++++++++++++++++++++ tests/self_coding/__init__.py | 0 tests/self_coding/test_loop.py | 363 ++++++++++++++++++++++++ 6 files changed, 801 insertions(+) create mode 100644 src/self_coding/__init__.py create mode 100644 src/self_coding/gitea_client.py create mode 100644 src/self_coding/self_modify/__init__.py create mode 100644 src/self_coding/self_modify/loop.py create mode 100644 tests/self_coding/__init__.py create mode 100644 tests/self_coding/test_loop.py diff --git a/src/self_coding/__init__.py b/src/self_coding/__init__.py new file mode 100644 index 00000000..e237a682 --- /dev/null +++ b/src/self_coding/__init__.py @@ -0,0 +1,7 @@ +"""Self-coding package — Timmy's self-modification capability. + +Provides the branch→edit→test→commit/revert loop that allows Timmy +to propose and apply code changes autonomously, gated by the test suite. + +Main entry point: ``self_coding.self_modify.loop`` +""" diff --git a/src/self_coding/gitea_client.py b/src/self_coding/gitea_client.py new file mode 100644 index 00000000..5f199f88 --- /dev/null +++ b/src/self_coding/gitea_client.py @@ -0,0 +1,129 @@ +"""Gitea REST client — thin wrapper for PR creation and issue commenting. + +Uses ``settings.gitea_url``, ``settings.gitea_token``, and +``settings.gitea_repo`` (owner/repo) from config. Degrades gracefully +when the token is absent or the server is unreachable. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + + +@dataclass +class PullRequest: + """Minimal representation of a created pull request.""" + + number: int + title: str + html_url: str + + +class GiteaClient: + """HTTP client for Gitea's REST API v1. + + All methods return structured results and never raise — errors are + logged at WARNING level and indicated via return value. + """ + + def __init__( + self, + base_url: str | None = None, + token: str | None = None, + repo: str | None = None, + ) -> None: + from config import settings + + self._base_url = (base_url or settings.gitea_url).rstrip("/") + self._token = token or settings.gitea_token + self._repo = repo or settings.gitea_repo + + # ── internal ──────────────────────────────────────────────────────────── + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"token {self._token}", + "Content-Type": "application/json", + } + + def _api(self, path: str) -> str: + return f"{self._base_url}/api/v1/{path.lstrip('/')}" + + # ── public API ─────────────────────────────────────────────────────────── + + def create_pull_request( + self, + title: str, + body: str, + head: str, + base: str = "main", + ) -> PullRequest | None: + """Open a pull request. + + Args: + title: PR title (keep under 70 chars). + body: PR body in markdown. + head: Source branch (e.g. ``self-modify/issue-983``). + base: Target branch (default ``main``). + + Returns: + A ``PullRequest`` dataclass on success, ``None`` on failure. + """ + if not self._token: + logger.warning("Gitea token not configured — skipping PR creation") + return None + + try: + import requests as _requests + + resp = _requests.post( + self._api(f"repos/{self._repo}/pulls"), + headers=self._headers(), + json={"title": title, "body": body, "head": head, "base": base}, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + pr = PullRequest( + number=data["number"], + title=data["title"], + html_url=data["html_url"], + ) + logger.info("PR #%d created: %s", pr.number, pr.html_url) + return pr + except Exception as exc: + logger.warning("Failed to create PR: %s", exc) + return None + + def add_issue_comment(self, issue_number: int, body: str) -> bool: + """Post a comment on an issue or PR. + + Returns: + True on success, False on failure. + """ + if not self._token: + logger.warning("Gitea token not configured — skipping issue comment") + return False + + try: + import requests as _requests + + resp = _requests.post( + self._api(f"repos/{self._repo}/issues/{issue_number}/comments"), + headers=self._headers(), + json={"body": body}, + timeout=15, + ) + resp.raise_for_status() + logger.info("Comment posted on issue #%d", issue_number) + return True + except Exception as exc: + logger.warning("Failed to post comment on issue #%d: %s", issue_number, exc) + return False + + +# Module-level singleton +gitea_client = GiteaClient() diff --git a/src/self_coding/self_modify/__init__.py b/src/self_coding/self_modify/__init__.py new file mode 100644 index 00000000..13c93863 --- /dev/null +++ b/src/self_coding/self_modify/__init__.py @@ -0,0 +1 @@ +"""Self-modification loop sub-package.""" diff --git a/src/self_coding/self_modify/loop.py b/src/self_coding/self_modify/loop.py new file mode 100644 index 00000000..57b22566 --- /dev/null +++ b/src/self_coding/self_modify/loop.py @@ -0,0 +1,301 @@ +"""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/`` 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, + ) diff --git a/tests/self_coding/__init__.py b/tests/self_coding/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/self_coding/test_loop.py b/tests/self_coding/test_loop.py new file mode 100644 index 00000000..f5383655 --- /dev/null +++ b/tests/self_coding/test_loop.py @@ -0,0 +1,363 @@ +"""Unit tests for the self-modification loop. + +Covers: +- Protected branch guard +- Successful cycle (mocked git + tests) +- Edit function failure → branch reverted, no commit +- Test failure → branch reverted, no commit +- Gitea PR creation plumbing +- GiteaClient graceful degradation (no token, network error) + +All git and subprocess calls are mocked so these run offline without +a real repo or test suite. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_loop(repo_root="/tmp/fake-repo"): + """Construct a SelfModifyLoop with a fake repo root.""" + from self_coding.self_modify.loop import SelfModifyLoop + + return SelfModifyLoop(repo_root=repo_root, remote="origin", base_branch="main") + + +def _noop_edit(repo_root: str) -> None: + """Edit function that does nothing.""" + + +def _failing_edit(repo_root: str) -> None: + """Edit function that raises.""" + raise RuntimeError("edit exploded") + + +# --------------------------------------------------------------------------- +# Guard tests (sync — no git calls needed) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_guard_blocks_main(): + loop = _make_loop() + with pytest.raises(ValueError, match="protected branch"): + loop._guard_branch("main") + + +@pytest.mark.unit +def test_guard_blocks_master(): + loop = _make_loop() + with pytest.raises(ValueError, match="protected branch"): + loop._guard_branch("master") + + +@pytest.mark.unit +def test_guard_allows_feature_branch(): + loop = _make_loop() + # Should not raise + loop._guard_branch("self-modify/some-feature") + + +@pytest.mark.unit +def test_guard_allows_self_modify_prefix(): + loop = _make_loop() + loop._guard_branch("self-modify/issue-983") + + +# --------------------------------------------------------------------------- +# Full cycle — success path +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_run_success(): + """Happy path: edit succeeds, tests pass, PR created.""" + loop = _make_loop() + + fake_completed = MagicMock() + fake_completed.stdout = "abc1234\n" + fake_completed.returncode = 0 + + fake_test_result = MagicMock() + fake_test_result.stdout = "3 passed" + fake_test_result.stderr = "" + fake_test_result.returncode = 0 + + from self_coding.gitea_client import PullRequest as _PR + + fake_pr = _PR(number=42, title="test PR", html_url="http://gitea/pr/42") + + with ( + patch.object(loop, "_git", return_value=fake_completed), + patch("subprocess.run", return_value=fake_test_result), + patch.object(loop, "_create_pr", return_value=fake_pr), + ): + result = await loop.run( + slug="test-feature", + description="Add test feature", + edit_fn=_noop_edit, + issue_number=983, + ) + + assert result.success is True + assert result.branch == "self-modify/test-feature" + assert result.pr_url == "http://gitea/pr/42" + assert result.pr_number == 42 + assert "3 passed" in result.test_output + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_run_skips_tests_when_flag_set(): + """skip_tests=True should bypass the test gate.""" + loop = _make_loop() + + fake_completed = MagicMock() + fake_completed.stdout = "deadbeef\n" + fake_completed.returncode = 0 + + with ( + patch.object(loop, "_git", return_value=fake_completed), + patch.object(loop, "_create_pr", return_value=None), + patch("subprocess.run") as mock_run, + ): + result = await loop.run( + slug="skip-test-feature", + description="Skip test feature", + edit_fn=_noop_edit, + skip_tests=True, + ) + + # subprocess.run should NOT be called for tests + mock_run.assert_not_called() + assert result.success is True + assert "(tests skipped)" in result.test_output + + +# --------------------------------------------------------------------------- +# Failure paths +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_run_reverts_on_edit_failure(): + """If edit_fn raises, the branch should be reverted and no commit made.""" + loop = _make_loop() + + fake_completed = MagicMock() + fake_completed.stdout = "" + fake_completed.returncode = 0 + + revert_called = [] + + def _fake_revert(branch): + revert_called.append(branch) + + with ( + patch.object(loop, "_git", return_value=fake_completed), + patch.object(loop, "_revert_branch", side_effect=_fake_revert), + patch.object(loop, "_commit_all") as mock_commit, + ): + result = await loop.run( + slug="broken-edit", + description="This will fail", + edit_fn=_failing_edit, + skip_tests=True, + ) + + assert result.success is False + assert "edit exploded" in result.error + assert "self-modify/broken-edit" in revert_called + mock_commit.assert_not_called() + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_run_reverts_on_test_failure(): + """If tests fail, branch should be reverted and no commit made.""" + loop = _make_loop() + + fake_completed = MagicMock() + fake_completed.stdout = "" + fake_completed.returncode = 0 + + fake_test_result = MagicMock() + fake_test_result.stdout = "FAILED test_foo" + fake_test_result.stderr = "1 failed" + fake_test_result.returncode = 1 + + revert_called = [] + + def _fake_revert(branch): + revert_called.append(branch) + + with ( + patch.object(loop, "_git", return_value=fake_completed), + patch("subprocess.run", return_value=fake_test_result), + patch.object(loop, "_revert_branch", side_effect=_fake_revert), + patch.object(loop, "_commit_all") as mock_commit, + ): + result = await loop.run( + slug="tests-will-fail", + description="This will fail tests", + edit_fn=_noop_edit, + ) + + assert result.success is False + assert "Tests failed" in result.error + assert "self-modify/tests-will-fail" in revert_called + mock_commit.assert_not_called() + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_run_slug_with_main_creates_safe_branch(): + """A slug of 'main' produces branch 'self-modify/main', which is not protected.""" + + loop = _make_loop() + + fake_completed = MagicMock() + fake_completed.stdout = "deadbeef\n" + fake_completed.returncode = 0 + + # 'self-modify/main' is NOT in _PROTECTED_BRANCHES so the run should succeed + with ( + patch.object(loop, "_git", return_value=fake_completed), + patch.object(loop, "_create_pr", return_value=None), + ): + result = await loop.run( + slug="main", + description="try to write to self-modify/main", + edit_fn=_noop_edit, + skip_tests=True, + ) + assert result.branch == "self-modify/main" + assert result.success is True + + +# --------------------------------------------------------------------------- +# GiteaClient tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_gitea_client_returns_none_without_token(): + """GiteaClient should return None gracefully when no token is set.""" + from self_coding.gitea_client import GiteaClient + + client = GiteaClient(base_url="http://localhost:3000", token="", repo="owner/repo") + pr = client.create_pull_request( + title="Test PR", + body="body", + head="self-modify/test", + ) + assert pr is None + + +@pytest.mark.unit +def test_gitea_client_comment_returns_false_without_token(): + """add_issue_comment should return False gracefully when no token is set.""" + from self_coding.gitea_client import GiteaClient + + client = GiteaClient(base_url="http://localhost:3000", token="", repo="owner/repo") + result = client.add_issue_comment(123, "hello") + assert result is False + + +@pytest.mark.unit +def test_gitea_client_create_pr_handles_network_error(): + """create_pull_request should return None on network failure.""" + from self_coding.gitea_client import GiteaClient + + client = GiteaClient(base_url="http://localhost:3000", token="fake-token", repo="owner/repo") + + mock_requests = MagicMock() + mock_requests.post.side_effect = Exception("Connection refused") + mock_requests.exceptions.ConnectionError = Exception + + with patch.dict("sys.modules", {"requests": mock_requests}): + pr = client.create_pull_request( + title="Test PR", + body="body", + head="self-modify/test", + ) + assert pr is None + + +@pytest.mark.unit +def test_gitea_client_comment_handles_network_error(): + """add_issue_comment should return False on network failure.""" + from self_coding.gitea_client import GiteaClient + + client = GiteaClient(base_url="http://localhost:3000", token="fake-token", repo="owner/repo") + + mock_requests = MagicMock() + mock_requests.post.side_effect = Exception("Connection refused") + + with patch.dict("sys.modules", {"requests": mock_requests}): + result = client.add_issue_comment(456, "hello") + assert result is False + + +@pytest.mark.unit +def test_gitea_client_create_pr_success(): + """create_pull_request should return a PullRequest on HTTP 201.""" + from self_coding.gitea_client import GiteaClient, PullRequest + + client = GiteaClient(base_url="http://localhost:3000", token="tok", repo="owner/repo") + + fake_resp = MagicMock() + fake_resp.raise_for_status = MagicMock() + fake_resp.json.return_value = { + "number": 77, + "title": "Test PR", + "html_url": "http://localhost:3000/owner/repo/pulls/77", + } + + mock_requests = MagicMock() + mock_requests.post.return_value = fake_resp + + with patch.dict("sys.modules", {"requests": mock_requests}): + pr = client.create_pull_request("Test PR", "body", "self-modify/feat") + + assert isinstance(pr, PullRequest) + assert pr.number == 77 + assert pr.html_url == "http://localhost:3000/owner/repo/pulls/77" + + +# --------------------------------------------------------------------------- +# LoopResult dataclass +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_loop_result_defaults(): + from self_coding.self_modify.loop import LoopResult + + r = LoopResult(success=True) + assert r.branch == "" + assert r.commit_sha == "" + assert r.pr_url == "" + assert r.pr_number == 0 + assert r.test_output == "" + assert r.error == "" + assert r.elapsed_ms == 0.0 + assert r.metadata == {} + + +@pytest.mark.unit +def test_loop_result_failure(): + from self_coding.self_modify.loop import LoopResult + + r = LoopResult(success=False, error="something broke", branch="self-modify/test") + assert r.success is False + assert r.error == "something broke"