1
0

[claude] Restore self-modification loop (#983) (#1270)

Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
This commit is contained in:
2026-03-24 01:40:16 +00:00
committed by Timmy Time
parent 7dfbf05867
commit 003e3883fb
6 changed files with 801 additions and 0 deletions

View File

@@ -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``
"""

View File

@@ -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()

View File

@@ -0,0 +1 @@
"""Self-modification loop sub-package."""

View File

@@ -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/<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,
)

View File

View File

@@ -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"