.env.example

@@ -25,3 +25,9 @@ NISSE_MATRIX_ROOM_ID=!roomid:example
 # Dedicated room for the live integration test; never the production room.
 NISSE_MATRIX_TEST_ROOM_ID=!testroomid:example
 NISSE_MATRIX_OPERATORS=@you:example
+
+# --- Gitea (B2): service account for branches, PRs, and comment intake ---
+NISSE_GITEA_URL=https://gitea.example
+NISSE_GITEA_TOKEN=your-scoped-service-account-token
+# Sandbox repo for live tests; never the real project repo.
+NISSE_GITEA_TEST_REPO=owner/sandbox-repo

.gitea/workflows/ci.yml

@@ -39,4 +39,7 @@ jobs:
           NISSE_MATRIX_USER_ID: ${{ secrets.NISSE_MATRIX_USER_ID }}
           NISSE_MATRIX_TOKEN: ${{ secrets.NISSE_MATRIX_TOKEN }}
           NISSE_MATRIX_TEST_ROOM_ID: ${{ secrets.NISSE_MATRIX_TEST_ROOM_ID }}
+          NISSE_GITEA_URL: ${{ secrets.NISSE_GITEA_URL }}
+          NISSE_GITEA_TOKEN: ${{ secrets.NISSE_GITEA_TOKEN }}
+          NISSE_GITEA_TEST_REPO: ${{ secrets.NISSE_GITEA_TEST_REPO }}
         run: uv run pytest -m integration -rs

nisse-dev/src/nisse_dev/git.py

@@ -0,0 +1,96 @@
+"""Git operations for the agent's workspace (spec B2 outbound).
+
+Push authentication uses the Gitea service-account token embedded in the
+remote URL. The token is never written to disk in the repo config and is
+scrubbed from anything this module raises or returns, so a failed push cannot
+leak the credential into a transcript, a log line, or a PR comment.
+"""
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+TIMEOUT_S = 120
+
+
+class GitError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class GitRepo:
+    root: Path
+
+    def _run(self, *args: str, secret: str | None = None) -> str:
+        try:
+            completed = subprocess.run(
+                ["git", *args],
+                cwd=self.root,
+                capture_output=True,
+                text=True,
+                timeout=TIMEOUT_S,
+                check=False,
+            )
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            raise GitError(f"git {args[0]} failed: {_scrub(str(exc), secret)}") from exc
+        if completed.returncode != 0:
+            raise GitError(
+                f"git {args[0]} failed: "
+                f"{_scrub(completed.stderr.strip() or completed.stdout.strip(), secret)}"
+            )
+        return completed.stdout.strip()
+
+    # -- inspection -------------------------------------------------------
+
+    def current_branch(self) -> str:
+        # --show-current rather than rev-parse: it answers correctly on a
+        # freshly initialised repo, where HEAD points at an unborn branch.
+        return self._run("branch", "--show-current")
+
+    def head_sha(self) -> str:
+        return self._run("rev-parse", "HEAD")
+
+    def has_changes(self) -> bool:
+        return bool(self._run("status", "--porcelain"))
+
+    # -- mutation ---------------------------------------------------------
+
+    def init(self, *, default_branch: str = "main") -> None:
+        self.root.mkdir(parents=True, exist_ok=True)
+        self._run("init", "-q", "-b", default_branch)
+
+    def configure_identity(self, name: str, email: str) -> None:
+        self._run("config", "user.name", name)
+        self._run("config", "user.email", email)
+
+    def checkout_new_branch(self, name: str) -> None:
+        self._run("checkout", "-q", "-b", name)
+
+    def commit_all(self, message: str) -> str | None:
+        """Returns the new commit sha, or None when there was nothing to
+        commit — an empty milestone is a fact to report, not an error."""
+        if not self.has_changes():
+            return None
+        self._run("add", "-A")
+        self._run("commit", "-q", "-m", message)
+        return self.head_sha()
+
+    def push(self, remote_url: str, branch: str, *, token: str | None = None) -> None:
+        # The URL is passed per-invocation rather than stored as a remote, so
+        # the token never lands in .git/config.
+        self._run("push", "-q", remote_url, f"HEAD:refs/heads/{branch}", secret=token)
+
+    def delete_remote_branch(
+        self, remote_url: str, branch: str, *, token: str | None = None
+    ) -> None:
+        self._run("push", "-q", remote_url, f":refs/heads/{branch}", secret=token)
+
+
+def authenticated_url(base_url: str, repo: str, user: str, token: str) -> str:
+    host = base_url.split("://", 1)[-1].rstrip("/")
+    scheme = base_url.split("://", 1)[0] if "://" in base_url else "https"
+    return f"{scheme}://{user}:{token}@{host}/{repo}.git"
+
+
+def _scrub(text: str, secret: str | None) -> str:
+    return text.replace(secret, "***") if secret else text

nisse-dev/src/nisse_dev/gitea.py

@@ -0,0 +1,187 @@
+"""Gitea client (spec B2).
+
+Outbound: open PRs and post comments through a dedicated service account with
+a scoped token. Inbound is deliberately *polled* rather than webhook-driven,
+so the agent VM stays outbound-only (B3), and intake is structured JSON —
+never routed through a chat bridge.
+
+``find_pull_request`` is the external-truth query that makes opening a PR
+idempotent across a crash: boot reconciliation (A3) asks Gitea whether the
+effect already happened rather than repeating it.
+"""
+
+import os
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from nisse_dev.reconcile import Reconciler
+from nisse_dev.state import Intent
+
+PR_PAGE_SIZE = 50
+MAX_PR_PAGES = 40  # 2000 PRs; beyond that, say so rather than guess
+
+
+class GiteaError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class PullRequest:
+    number: int
+    state: str
+    head: str
+    url: str
+
+
+@dataclass(frozen=True)
+class Comment:
+    id: int
+    author: str
+    body: str
+    created_at: str
+
+
+class GiteaClient:
+    def __init__(
+        self,
+        base_url: str,
+        token: str,
+        *,
+        timeout_s: float = 30.0,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        self._http = httpx.Client(
+            base_url=f"{base_url.rstrip('/')}/api/v1",
+            headers={"Authorization": f"token {token}"},
+            timeout=timeout_s,
+            transport=transport,
+        )
+
+    @classmethod
+    def from_env(cls, env: Mapping[str, str] = os.environ) -> "GiteaClient":
+        base_url = env.get("NISSE_GITEA_URL")
+        token = env.get("NISSE_GITEA_TOKEN")
+        if not base_url or not token:
+            raise GiteaError("set NISSE_GITEA_URL and NISSE_GITEA_TOKEN")
+        return cls(base_url, token)
+
+    def _request(self, method: str, path: str, **kwargs: Any) -> Any:
+        try:
+            response = self._http.request(method, path, **kwargs)
+        except httpx.HTTPError as exc:
+            raise GiteaError(f"gitea request failed: {exc}") from exc
+        if response.status_code >= 400:
+            raise GiteaError(
+                f"gitea {method} {path} returned {response.status_code}: "
+                f"{response.text[:300]}"
+            )
+        return response.json() if response.content else None
+
+    def whoami(self) -> str:
+        return str(self._request("GET", "/user")["login"])
+
+    def open_pull_request(
+        self, repo: str, *, head: str, base: str, title: str, body: str
+    ) -> PullRequest:
+        data = self._request(
+            "POST",
+            f"/repos/{repo}/pulls",
+            json={"head": head, "base": base, "title": title, "body": body},
+        )
+        return _pull_request(data)
+
+    def find_pull_request(self, repo: str, head: str) -> PullRequest | None:
+        """External truth for A3: did the PR for this branch already get
+        opened? Includes closed PRs — a crash after opening must not be
+        answered with "no PR is open, so open another".
+
+        Pages to the end rather than sampling the first page: a "not found"
+        that only means "not in the newest 50" would licence a duplicate PR.
+        If the history is longer than this walk, raise instead of guessing —
+        the reconciler turns that into "unverifiable" and gates to the human.
+        """
+        for page in range(1, MAX_PR_PAGES + 1):
+            data = self._request(
+                "GET",
+                f"/repos/{repo}/pulls",
+                params={"state": "all", "limit": PR_PAGE_SIZE, "page": page},
+            )
+            items = data or []
+            for item in items:
+                if (item.get("head") or {}).get("ref") == head:
+                    return _pull_request(item)
+            if len(items) < PR_PAGE_SIZE:
+                return None  # reached the end: definitively absent
+        raise GiteaError(
+            f"searched {MAX_PR_PAGES * PR_PAGE_SIZE} pull requests in {repo} "
+            f"without reaching the end; cannot confirm whether {head!r} was "
+            "already opened"
+        )
+
+    def list_comments(
+        self, repo: str, index: int, since: str | None = None
+    ) -> list[Comment]:
+        params: dict[str, str] = {}
+        if since is not None:
+            params["since"] = since
+        data = self._request(
+            "GET", f"/repos/{repo}/issues/{index}/comments", params=params
+        )
+        return [
+            Comment(
+                id=item["id"],
+                author=(item.get("user") or {}).get("login", ""),
+                body=item.get("body", ""),
+                created_at=item.get("created_at", ""),
+            )
+            for item in data or []
+        ]
+
+    def post_comment(self, repo: str, index: int, body: str) -> Comment:
+        data = self._request(
+            "POST", f"/repos/{repo}/issues/{index}/comments", json={"body": body}
+        )
+        return Comment(
+            id=data["id"],
+            author=(data.get("user") or {}).get("login", ""),
+            body=data.get("body", ""),
+            created_at=data.get("created_at", ""),
+        )
+
+    def close_pull_request(self, repo: str, index: int) -> None:
+        self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"})
+
+    def close(self) -> None:
+        self._http.close()
+
+
+def _pull_request(data: dict[str, Any]) -> PullRequest:
+    return PullRequest(
+        number=data["number"],
+        state=data.get("state", ""),
+        head=(data.get("head") or {}).get("ref", ""),
+        url=data.get("html_url", ""),
+    )
+
+
+def pull_request_reconciler(client: GiteaClient, repo: str) -> Reconciler:
+    """Reconciler for an ``open_pr`` intent (A3 step 4).
+
+    An unreachable Gitea yields None — unverifiable, so the agent gates to
+    the human rather than risking a duplicate PR. That is the conservative
+    direction: a delayed decision costs nothing, a repeated side effect does.
+    """
+
+    def check(intent: Intent) -> bool | None:
+        branch = intent.payload.get("branch")
+        if not branch:
+            return None
+        try:
+            return client.find_pull_request(repo, branch) is not None
+        except GiteaError:
+            return None
+
+    return check

nisse-dev/tests/test_git.py

@@ -0,0 +1,70 @@
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.git import GitError, GitRepo, _scrub, authenticated_url
+
+
+@pytest.fixture
+def repo(tmp_path: Path) -> GitRepo:
+    r = GitRepo(tmp_path / "work")
+    r.init()
+    r.configure_identity("nisse-dev-testing", "nisse@example.test")
+    return r
+
+
+def test_init_creates_repo_on_default_branch(repo: GitRepo) -> None:
+    assert repo.current_branch() == "main"
+
+
+def test_commit_all_returns_sha_and_branch_switching(repo: GitRepo) -> None:
+    (repo.root / "a.txt").write_text("a")
+    sha = repo.commit_all("first")
+    assert sha is not None
+    assert sha == repo.head_sha()
+
+    repo.checkout_new_branch("milestone-1")
+    assert repo.current_branch() == "milestone-1"
+
+
+def test_commit_with_nothing_staged_returns_none(repo: GitRepo) -> None:
+    (repo.root / "a.txt").write_text("a")
+    repo.commit_all("first")
+    # An empty milestone is a fact to report, not an error to raise.
+    assert repo.commit_all("nothing changed") is None
+
+
+def test_has_changes_tracks_the_worktree(repo: GitRepo) -> None:
+    assert not repo.has_changes()
+    (repo.root / "a.txt").write_text("a")
+    assert repo.has_changes()
+
+
+def test_authenticated_url_shape() -> None:
+    url = authenticated_url("https://gitea.test/", "o/r", "bot", "secret")
+    assert url == "https://bot:secret@gitea.test/o/r.git"
+
+
+def test_push_failure_does_not_leak_the_token(repo: GitRepo) -> None:
+    (repo.root / "a.txt").write_text("a")
+    repo.commit_all("first")
+    url = authenticated_url("https://127.0.0.1:1", "o/r", "bot", "sup3rsecret")
+    with pytest.raises(GitError) as excinfo:
+        repo.push(url, "main", token="sup3rsecret")
+    # git redacts credentials from its own messages, but that is git's
+    # behaviour, not ours; the scrub below is the guarantee we control.
+    assert "sup3rsecret" not in str(excinfo.value)
+
+
+def test_scrub_removes_the_token_when_git_does_echo_it() -> None:
+    assert _scrub("failed to push to https://bot:s3cret@h/o/r.git", "s3cret") == (
+        "failed to push to https://bot:***@h/o/r.git"
+    )
+    assert _scrub("no secret here", None) == "no secret here"
+
+
+def test_git_error_on_bad_command(tmp_path: Path) -> None:
+    repo = GitRepo(tmp_path / "not-a-repo")
+    repo.root.mkdir()
+    with pytest.raises(GitError):
+        repo.head_sha()

nisse-dev/tests/test_gitea.py

@@ -0,0 +1,202 @@
+import json
+
+import httpx
+import pytest
+
+from nisse_dev.gitea import (
+    PR_PAGE_SIZE,
+    Comment,
+    GiteaClient,
+    GiteaError,
+    pull_request_reconciler,
+)
+from nisse_dev.state import Intent
+
+INTENT = Intent(
+    id=1,
+    kind="open_pr",
+    payload={"branch": "milestone-1"},
+    status="pending",
+    created_at="t",
+    resolved_at=None,
+)
+
+
+def client(handler) -> GiteaClient:
+    return GiteaClient(
+        "https://gitea.test/", "tok", transport=httpx.MockTransport(handler)
+    )
+
+
+def pr_json(number: int, head: str, state: str = "open") -> dict:
+    return {
+        "number": number,
+        "state": state,
+        "head": {"ref": head},
+        "html_url": f"https://gitea.test/o/r/pulls/{number}",
+    }
+
+
+def test_open_pull_request_posts_expected_body() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        seen["auth"] = request.headers.get("authorization")
+        seen["body"] = json.loads(request.content)
+        return httpx.Response(201, json=pr_json(7, "milestone-1"))
+
+    pr = client(handler).open_pull_request(
+        "o/r", head="milestone-1", base="main", title="T", body="B"
+    )
+    assert seen["url"] == "https://gitea.test/api/v1/repos/o/r/pulls"
+    assert seen["auth"] == "token tok"
+    assert seen["body"] == {
+        "head": "milestone-1",
+        "base": "main",
+        "title": "T",
+        "body": "B",
+    }
+    assert pr.number == 7
+    assert pr.head == "milestone-1"
+
+
+def test_find_pull_request_matches_head_branch() -> None:
+    handler = lambda request: httpx.Response(200, json=[pr_json(3, "other")])
+    assert client(handler).find_pull_request("o/r", "milestone-1") is None
+
+
+def test_find_pull_request_also_finds_closed() -> None:
+    # A crash after opening must not be answered with "nothing open, open
+    # another" — the effect already happened.
+    handler = lambda request: httpx.Response(
+        200, json=[pr_json(4, "milestone-1", "closed")]
+    )
+    found = client(handler).find_pull_request("o/r", "milestone-1")
+    assert found is not None
+    assert found.number == 4
+
+
+def test_find_pull_request_pages_past_the_first_page() -> None:
+    # The match sits on page 2; sampling only the newest page would report
+    # "absent" and licence a duplicate PR.
+    def handler(request: httpx.Request) -> httpx.Response:
+        page = int(request.url.params.get("page", 1))
+        if page == 1:
+            return httpx.Response(
+                200, json=[pr_json(i, f"other-{i}") for i in range(PR_PAGE_SIZE)]
+            )
+        return httpx.Response(200, json=[pr_json(999, "milestone-1", "closed")])
+
+    found = client(handler).find_pull_request("o/r", "milestone-1")
+    assert found is not None
+    assert found.number == 999
+
+
+def test_short_page_ends_the_search_definitively() -> None:
+    calls: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(int(request.url.params.get("page", 1)))
+        return httpx.Response(200, json=[pr_json(1, "other")])
+
+    assert client(handler).find_pull_request("o/r", "milestone-1") is None
+    assert calls == [1]  # one short page proves absence; no further paging
+
+
+def test_unbounded_history_raises_rather_than_guessing() -> None:
+    # Never answer "absent" from an incomplete search: the reconciler turns
+    # this error into "unverifiable" and gates to the human.
+    handler = lambda request: httpx.Response(
+        200, json=[pr_json(i, f"other-{i}") for i in range(PR_PAGE_SIZE)]
+    )
+    with pytest.raises(GiteaError, match="without reaching the end"):
+        client(handler).find_pull_request("o/r", "milestone-1")
+
+
+def test_error_status_raises_gitea_error() -> None:
+    handler = lambda request: httpx.Response(403, text="forbidden")
+    with pytest.raises(GiteaError, match="403"):
+        client(handler).whoami()
+
+
+def test_list_and_post_comments() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST":
+            return httpx.Response(
+                201,
+                json={
+                    "id": 9,
+                    "user": {"login": "don"},
+                    "body": "hi",
+                    "created_at": "t",
+                },
+            )
+        return httpx.Response(
+            200,
+            json=[
+                {
+                    "id": 8,
+                    "user": {"login": "don"},
+                    "body": "approve rev-1",
+                    "created_at": "t",
+                }
+            ],
+        )
+
+    c = client(handler)
+    assert c.list_comments("o/r", 1) == [
+        Comment(id=8, author="don", body="approve rev-1", created_at="t")
+    ]
+    assert c.post_comment("o/r", 1, "hi").id == 9
+
+
+def test_list_comments_passes_since() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["since"] = request.url.params.get("since")
+        return httpx.Response(200, json=[])
+
+    client(handler).list_comments("o/r", 1, since="2026-09-07T00:00:00Z")
+    assert seen["since"] == "2026-09-07T00:00:00Z"
+
+
+def test_from_env_requires_config() -> None:
+    with pytest.raises(GiteaError, match="NISSE_GITEA_URL"):
+        GiteaClient.from_env({})
+
+
+# --- reconciler (A3 external truth) --------------------------------------
+
+
+def test_reconciler_reports_existing_pr_as_done() -> None:
+    handler = lambda request: httpx.Response(200, json=[pr_json(5, "milestone-1")])
+    assert pull_request_reconciler(client(handler), "o/r")(INTENT) is True
+
+
+def test_reconciler_reports_missing_pr_as_not_done() -> None:
+    handler = lambda request: httpx.Response(200, json=[])
+    assert pull_request_reconciler(client(handler), "o/r")(INTENT) is False
+
+
+def test_unreachable_gitea_is_unverifiable_not_a_guess() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("refused")
+
+    # None gates to the human: a delayed decision is cheap, a duplicate PR
+    # is not.
+    assert pull_request_reconciler(client(handler), "o/r")(INTENT) is None
+
+
+def test_intent_without_a_branch_is_unverifiable() -> None:
+    handler = lambda request: httpx.Response(200, json=[])
+    blank = Intent(
+        id=2,
+        kind="open_pr",
+        payload={},
+        status="pending",
+        created_at="t",
+        resolved_at=None,
+    )
+    assert pull_request_reconciler(client(handler), "o/r")(blank) is None

tests/integration/test_gitea_live.py

@@ -0,0 +1,95 @@
+"""Live Gitea round trip against the sandbox repo: push a branch, open a PR,
+comment, and prove the A3 reconciler reads external truth correctly.
+
+Cleans up after itself (closes the PR, deletes the branch) so repeated runs
+do not silt up the sandbox.
+"""
+
+import os
+import uuid
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.envfile import load_env_file
+from nisse_dev.git import GitRepo, authenticated_url
+from nisse_dev.gitea import GiteaClient, pull_request_reconciler
+from nisse_dev.state import Intent
+
+load_env_file(Path(__file__).resolve().parents[2] / ".env")
+
+SENTINEL = "NISSE_GITEA_URL"
+ALSO_REQUIRED = ("NISSE_GITEA_TOKEN", "NISSE_GITEA_TEST_REPO")
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not os.environ.get(SENTINEL),
+        reason=f"no live Gitea configured ({SENTINEL} unset)",
+    ),
+]
+
+
+def intent_for(branch: str) -> Intent:
+    return Intent(
+        id=1,
+        kind="open_pr",
+        payload={"branch": branch},
+        status="pending",
+        created_at="t",
+        resolved_at=None,
+    )
+
+
+def test_push_open_pr_and_reconcile(tmp_path: Path) -> None:
+    missing = [var for var in ALSO_REQUIRED if not os.environ.get(var)]
+    assert not missing, (
+        f"{SENTINEL} is set, so the live Gitea path is expected to work, "
+        f"but these are missing: {', '.join(missing)}"
+    )
+    base_url = os.environ[SENTINEL]
+    token = os.environ["NISSE_GITEA_TOKEN"]
+    repo_path = os.environ["NISSE_GITEA_TEST_REPO"]
+    client = GiteaClient.from_env()
+
+    account = client.whoami()
+    assert "testing" in account, (
+        f"live tests must run as a testing account, not {account}"
+    )
+
+    branch = f"live-test-{uuid.uuid4().hex[:8]}"
+    reconcile = pull_request_reconciler(client, repo_path)
+
+    # Before the PR exists, external truth says the effect has not happened.
+    assert reconcile(intent_for(branch)) is False
+
+    repo = GitRepo(tmp_path / "work")
+    repo.init()
+    repo.configure_identity(account, f"{account}@example.test")
+    (repo.root / f"{branch}.txt").write_text("live test artifact\n")
+    assert repo.commit_all("Live test commit") is not None
+
+    url = authenticated_url(base_url, repo_path, account, token)
+    pr = None
+    try:
+        repo.push(url, branch, token=token)
+        pr = client.open_pull_request(
+            repo_path,
+            head=branch,
+            base="main",
+            title=f"live test {branch}",
+            body="Opened by the nisse-dev live test suite.",
+        )
+        assert pr.head == branch
+
+        client.post_comment(repo_path, pr.number, "live test comment")
+        bodies = [c.body for c in client.list_comments(repo_path, pr.number)]
+        assert "live test comment" in bodies
+
+        # Now the same reconciler must see the effect and refuse to repeat it.
+        assert reconcile(intent_for(branch)) is True
+    finally:
+        if pr is not None:
+            client.close_pull_request(repo_path, pr.number)
+        repo.delete_remote_branch(url, branch, token=token)
+        client.close()