From eaae4357ebf3f63eeb9679d6a27cdbcd84eee73d Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 12:31:13 +0200 Subject: [PATCH] Build repository transaction core Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d --- rein_aharness/repository_transaction.py | 358 ++++++++++++++++++ tests/test_repository_transaction.py | 225 +++++++++++ ...NESS-WP-0003-governed-runtime-integrity.md | 17 + 3 files changed, 600 insertions(+) create mode 100644 rein_aharness/repository_transaction.py create mode 100644 tests/test_repository_transaction.py diff --git a/rein_aharness/repository_transaction.py b/rein_aharness/repository_transaction.py new file mode 100644 index 0000000..bfc77b3 --- /dev/null +++ b/rein_aharness/repository_transaction.py @@ -0,0 +1,358 @@ +"""Process-safe transaction boundary for a local Git checkout. + +This module is intentionally not wired into the claim loop yet. It provides +the repository-local half of HARNESS-WP-0003-T02 while the Activity Core lease +boundary is still under owner review. +""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import subprocess +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from types import TracebackType +from typing import Any, TextIO + + +class RepositoryTransactionError(RuntimeError): + """Base error for repository transaction refusal or setup failure.""" + + +class GitRepositoryError(RepositoryTransactionError): + """The target cannot provide the required Git repository state.""" + + +class RepositoryBusyError(RepositoryTransactionError): + """Another process owns the canonical repository lock.""" + + def __init__(self, repo_id: str, owner: dict[str, Any] | None = None) -> None: + self.repo_id = repo_id + self.owner = owner or {} + owner_tx = self.owner.get("transaction_id") + suffix = f" owner_transaction={owner_tx}" if owner_tx else "" + super().__init__(f"repository is already locked: repo_id={repo_id}{suffix}") + + +class DirtyRepositoryError(RepositoryTransactionError): + """An unattended transaction was refused because its baseline is dirty.""" + + def __init__(self, baseline: "RepositoryBaseline") -> None: + self.baseline = baseline + super().__init__( + "repository baseline is dirty: " + f"repo_id={baseline.repo_id} entries={baseline.dirty_entries}" + ) + + +@dataclass(frozen=True) +class RepositoryBaseline: + """Exact local facts captured while the canonical repository lock is held.""" + + repo_root: Path + git_common_dir: Path + repo_id: str + head: str + branch: str | None + clean: bool + dirty_entries: int + status_digest: str + index_diff_digest: str + worktree_diff_digest: str + upstream_ref: str | None + upstream_oid: str | None + remote_refs: tuple[tuple[str, str], ...] + + @property + def detached(self) -> bool: + return self.branch is None + + def evidence(self) -> dict[str, Any]: + """Return bounded, path-safe baseline evidence for run results.""" + remote_payload = "\n".join(f"{name} {oid}" for name, oid in self.remote_refs) + return { + "repo_id": self.repo_id, + "repo_name": self.repo_root.name, + "head": self.head, + "branch": self.branch, + "detached": self.detached, + "clean": self.clean, + "dirty_entries": self.dirty_entries, + "status_digest": self.status_digest, + "index_diff_digest": self.index_diff_digest, + "worktree_diff_digest": self.worktree_diff_digest, + "upstream_ref": self.upstream_ref, + "upstream_oid": self.upstream_oid, + "remote_ref_count": len(self.remote_refs), + "remote_refs_digest": _digest(remote_payload), + } + + +class RepositoryTransaction: + """Hold one canonical Git repository lock and capture its baseline. + + Lock files live outside the checkout and are keyed by the resolved Git + common directory. This makes aliases and linked worktrees for one local + repository contend on the same lock while unrelated repositories proceed + independently. + """ + + def __init__( + self, + repo: Path, + *, + correlation_id: str = "", + state_dir: Path | None = None, + require_clean: bool = True, + ) -> None: + self.repo = Path(repo).expanduser().resolve() + self.correlation_id = _bounded_text(correlation_id, 200) + self.state_dir = Path(state_dir).expanduser().resolve() if state_dir else _state_dir() + self.require_clean = require_clean + self.transaction_id = uuid.uuid4().hex + self.baseline: RepositoryBaseline | None = None + self.lock_path: Path | None = None + self._lock_file: TextIO | None = None + + @property + def locked(self) -> bool: + return self._lock_file is not None + + def __enter__(self) -> "RepositoryTransaction": + if self.locked: + raise RepositoryTransactionError("repository transaction is already entered") + + repo_root, common_dir, repo_id = _canonical_repository(self.repo) + locks_dir = self.state_dir / "repository-locks" + locks_dir.mkdir(parents=True, exist_ok=True) + self.lock_path = locks_dir / f"{repo_id}.lock" + lock_file = _open_private_lock(self.lock_path) + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + owner = _read_lock_metadata(lock_file) + lock_file.close() + raise RepositoryBusyError(repo_id, owner) from exc + + self._lock_file = lock_file + self._write_metadata( + { + "transaction_id": self.transaction_id, + "repo_id": repo_id, + "repo_name": repo_root.name, + "pid": os.getpid(), + "correlation_id": self.correlation_id, + "acquired_at": _now(), + "state": "capturing-baseline", + } + ) + try: + baseline = _capture_baseline(repo_root, common_dir, repo_id) + self.baseline = baseline + self._write_metadata( + { + "transaction_id": self.transaction_id, + "repo_id": repo_id, + "repo_name": repo_root.name, + "pid": os.getpid(), + "correlation_id": self.correlation_id, + "acquired_at": _now(), + "state": "active", + "baseline": baseline.evidence(), + } + ) + if self.require_clean and not baseline.clean: + raise DirtyRepositoryError(baseline) + return self + except BaseException: + self.release() + raise + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.release() + + def release(self) -> None: + """Release the OS lock without deleting checkout or lock-state data.""" + lock_file = self._lock_file + if lock_file is None: + return + try: + self._write_metadata( + { + "transaction_id": self.transaction_id, + "repo_id": self.baseline.repo_id if self.baseline else None, + "repo_name": self.baseline.repo_root.name if self.baseline else self.repo.name, + "pid": os.getpid(), + "correlation_id": self.correlation_id, + "released_at": _now(), + "state": "released", + } + ) + finally: + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + finally: + lock_file.close() + self._lock_file = None + + def evidence(self) -> dict[str, Any]: + """Return the bounded transaction and baseline result envelope.""" + if self.baseline is None: + raise RepositoryTransactionError("repository baseline is not captured") + return { + "transaction_id": self.transaction_id, + "correlation_id": self.correlation_id, + "baseline": self.baseline.evidence(), + } + + def _write_metadata(self, value: dict[str, Any]) -> None: + if self._lock_file is None: + return + self._lock_file.seek(0) + self._lock_file.truncate() + json.dump(value, self._lock_file, sort_keys=True, separators=(",", ":")) + self._lock_file.write("\n") + self._lock_file.flush() + os.fsync(self._lock_file.fileno()) + + +def _state_dir() -> Path: + explicit = os.environ.get("REIN_AHARNESS_STATE_DIR", "").strip() + if explicit: + return Path(explicit).expanduser().resolve() + xdg = os.environ.get("XDG_STATE_HOME", "").strip() + if xdg: + return (Path(xdg).expanduser() / "rein-aharness").resolve() + return (Path.home() / ".local" / "state" / "rein-aharness").resolve() + + +def _canonical_repository(repo: Path) -> tuple[Path, Path, str]: + root_result = _run_git(repo, "rev-parse", "--show-toplevel") + if root_result.returncode != 0: + raise GitRepositoryError(f"not a Git working tree: {repo}") + repo_root = Path(root_result.stdout.strip()).resolve() + + common_result = _run_git(repo_root, "rev-parse", "--git-common-dir") + if common_result.returncode != 0 or not common_result.stdout.strip(): + raise GitRepositoryError(f"cannot resolve Git common directory: {repo_root}") + raw_common = Path(common_result.stdout.strip()) + common_dir = (repo_root / raw_common).resolve() if not raw_common.is_absolute() else raw_common.resolve() + repo_id = hashlib.sha256(str(common_dir).encode("utf-8")).hexdigest()[:32] + return repo_root, common_dir, repo_id + + +def _capture_baseline(repo_root: Path, common_dir: Path, repo_id: str) -> RepositoryBaseline: + head = _git(repo_root, "rev-parse", "--verify", "HEAD") + branch_result = _run_git(repo_root, "symbolic-ref", "--quiet", "--short", "HEAD") + if branch_result.returncode not in (0, 1): + raise GitRepositoryError( + f"cannot resolve branch state: {branch_result.stderr.strip()[:200]}" + ) + branch = branch_result.stdout.strip() or None + + status = _git(repo_root, "status", "--porcelain=v2", "--untracked-files=all") + status_lines = tuple(line for line in status.splitlines() if line) + index_diff = _git(repo_root, "diff", "--cached", "--binary", "--no-ext-diff") + worktree_diff = _git(repo_root, "diff", "--binary", "--no-ext-diff") + + remote_lines = _git( + repo_root, + "for-each-ref", + "--format=%(refname) %(objectname)", + "refs/remotes", + ) + remote_refs: list[tuple[str, str]] = [] + for line in remote_lines.splitlines(): + name, separator, oid = line.partition(" ") + if separator and name and oid: + remote_refs.append((name, oid)) + + upstream_result = _run_git( + repo_root, + "rev-parse", + "--abbrev-ref", + "--symbolic-full-name", + "@{upstream}", + ) + upstream_ref = upstream_result.stdout.strip() if upstream_result.returncode == 0 else None + upstream_oid = _git(repo_root, "rev-parse", upstream_ref) if upstream_ref else None + + return RepositoryBaseline( + repo_root=repo_root, + git_common_dir=common_dir, + repo_id=repo_id, + head=head, + branch=branch, + clean=not status_lines, + dirty_entries=len(status_lines), + status_digest=_digest(status), + index_diff_digest=_digest(index_diff), + worktree_diff_digest=_digest(worktree_diff), + upstream_ref=upstream_ref, + upstream_oid=upstream_oid, + remote_refs=tuple(remote_refs), + ) + + +def _run_git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]: + try: + return subprocess.run( + ["git", "-C", str(repo), *args], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise GitRepositoryError(f"git {' '.join(args)} failed: {exc}") from exc + + +def _git(repo: Path, *args: str) -> str: + result = _run_git(repo, *args) + if result.returncode != 0: + reason = (result.stderr or result.stdout or "unknown Git error").strip()[:300] + raise GitRepositoryError(f"git {' '.join(args)} failed: {reason}") + return result.stdout.strip() + + +def _read_lock_metadata(lock_file: TextIO) -> dict[str, Any] | None: + try: + lock_file.seek(0) + raw = lock_file.read(4096) + value = json.loads(raw) if raw.strip() else None + except (OSError, ValueError): + return None + return value if isinstance(value, dict) else None + + +def _open_private_lock(path: Path) -> TextIO: + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + try: + os.fchmod(fd, 0o600) + return os.fdopen(fd, "r+", encoding="utf-8") + except BaseException: + os.close(fd) + raise + + +def _digest(value: str) -> str: + return hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest() + + +def _bounded_text(value: str, limit: int) -> str: + cleaned = " ".join(str(value).split()) + return cleaned[:limit] + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() diff --git a/tests/test_repository_transaction.py b/tests/test_repository_transaction.py new file mode 100644 index 0000000..d55e046 --- /dev/null +++ b/tests/test_repository_transaction.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from rein_aharness.repository_transaction import ( + DirtyRepositoryError, + GitRepositoryError, + RepositoryBusyError, + RepositoryTransaction, +) + + +def _make_repo(path: Path, name: str = "repo") -> Path: + repo = path / name + repo.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + (repo / "README.md").write_text("baseline\n", encoding="utf-8") + subprocess.run(["git", "add", "."], cwd=repo, check=True) + subprocess.run( + [ + "git", + "-c", + "user.email=test@example.invalid", + "-c", + "user.name=test", + "commit", + "-qm", + "baseline", + ], + cwd=repo, + check=True, + ) + return repo + + +def test_transaction_captures_clean_branch_and_remote_refs(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + subprocess.run( + ["git", "update-ref", "refs/remotes/origin/main", "HEAD"], + cwd=repo, + check=True, + ) + state_dir = tmp_path / "state" + + with RepositoryTransaction(repo, correlation_id="run-1", state_dir=state_dir) as tx: + assert tx.locked + assert tx.baseline is not None + evidence = tx.baseline.evidence() + assert evidence["clean"] is True + assert evidence["head"] + assert evidence["branch"] in {"main", "master"} + assert evidence["detached"] is False + assert evidence["remote_ref_count"] == 1 + assert len(evidence["repo_id"]) == 32 + assert tx.evidence() == { + "transaction_id": tx.transaction_id, + "correlation_id": "run-1", + "baseline": evidence, + } + assert tx.lock_path is not None + assert not tx.lock_path.is_relative_to(repo) + assert tx.lock_path.stat().st_mode & 0o777 == 0o600 + metadata = json.loads(tx.lock_path.read_text(encoding="utf-8")) + assert metadata["transaction_id"] == tx.transaction_id + assert metadata["correlation_id"] == "run-1" + + assert tx.locked is False + + +def test_dirty_baseline_is_refused_without_changing_user_files(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + changed = repo / "README.md" + changed.write_text("operator change\n", encoding="utf-8") + before = _status(repo) + + with pytest.raises(DirtyRepositoryError) as excinfo: + with RepositoryTransaction(repo, state_dir=tmp_path / "state"): + pytest.fail("dirty transaction must not enter") + + assert excinfo.value.baseline.clean is False + assert excinfo.value.baseline.dirty_entries == 1 + assert changed.read_text(encoding="utf-8") == "operator change\n" + assert _status(repo) == before + + # Refusal released the lock; inspection can explicitly opt into dirty state. + with RepositoryTransaction( + repo, + state_dir=tmp_path / "state", + require_clean=False, + ) as tx: + assert tx.baseline is not None + assert tx.baseline.clean is False + + +def test_detached_head_is_recorded_without_refusal(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + subprocess.run(["git", "checkout", "--detach", "-q"], cwd=repo, check=True) + + with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx: + assert tx.baseline is not None + assert tx.baseline.detached is True + assert tx.baseline.branch is None + + +def test_staged_and_untracked_state_are_captured_as_dirty(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + (repo / "README.md").write_text("staged change\n", encoding="utf-8") + (repo / "UNTRACKED.txt").write_text("operator file\n", encoding="utf-8") + subprocess.run(["git", "add", "README.md"], cwd=repo, check=True) + + with RepositoryTransaction( + repo, + state_dir=tmp_path / "state", + require_clean=False, + ) as tx: + assert tx.baseline is not None + assert tx.baseline.clean is False + assert tx.baseline.dirty_entries == 2 + assert tx.baseline.index_diff_digest != hashlib.sha256(b"").hexdigest() + assert tx.baseline.status_digest + + +def test_resolved_path_alias_contends_on_the_same_lock(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + alias = tmp_path / "repo-alias" + alias.symlink_to(repo, target_is_directory=True) + state_dir = tmp_path / "state" + + with RepositoryTransaction(repo, state_dir=state_dir): + with pytest.raises(RepositoryBusyError): + with RepositoryTransaction(alias, state_dir=state_dir): + pytest.fail("path alias must use the canonical repository lock") + + +def test_same_repo_is_locked_across_processes(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + state_dir = tmp_path / "state" + child = """ +import sys +from pathlib import Path +from rein_aharness.repository_transaction import RepositoryBusyError, RepositoryTransaction +try: + with RepositoryTransaction(Path(sys.argv[1]), state_dir=Path(sys.argv[2])): + pass +except RepositoryBusyError: + raise SystemExit(23) +raise SystemExit(0) +""" + + with RepositoryTransaction(repo, state_dir=state_dir): + result = subprocess.run( + [sys.executable, "-c", child, str(repo), str(state_dir)], + cwd=Path(__file__).resolve().parents[1], + env=os.environ.copy(), + check=False, + ) + assert result.returncode == 23 + + +def test_distinct_repositories_use_distinct_locks(tmp_path: Path) -> None: + first = _make_repo(tmp_path, "first") + second = _make_repo(tmp_path, "second") + state_dir = tmp_path / "state" + + with RepositoryTransaction(first, state_dir=state_dir) as first_tx: + with RepositoryTransaction(second, state_dir=state_dir) as second_tx: + assert first_tx.baseline is not None + assert second_tx.baseline is not None + assert first_tx.baseline.repo_id != second_tx.baseline.repo_id + assert first_tx.lock_path != second_tx.lock_path + + +def test_exception_releases_lock_for_next_transaction(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + state_dir = tmp_path / "state" + + with pytest.raises(RuntimeError, match="adapter failed"): + with RepositoryTransaction(repo, state_dir=state_dir): + raise RuntimeError("adapter failed") + + with RepositoryTransaction(repo, state_dir=state_dir) as tx: + assert tx.locked + + +def test_non_repository_is_refused_before_lock_creation(tmp_path: Path) -> None: + state_dir = tmp_path / "state" + target = tmp_path / "not-a-repo" + target.mkdir() + + with pytest.raises(GitRepositoryError, match="not a Git working tree"): + with RepositoryTransaction(target, state_dir=state_dir): + pytest.fail("non-repository transaction must not enter") + + assert not state_dir.exists() + + +def test_busy_error_exposes_bounded_owner_metadata(tmp_path: Path) -> None: + repo = _make_repo(tmp_path) + state_dir = tmp_path / "state" + long_ref = "run " + ("x" * 500) + + with RepositoryTransaction(repo, correlation_id=long_ref, state_dir=state_dir) as owner: + with pytest.raises(RepositoryBusyError) as excinfo: + with RepositoryTransaction(repo, state_dir=state_dir): + pytest.fail("second transaction must not enter") + + assert excinfo.value.owner["transaction_id"] == owner.transaction_id + assert len(excinfo.value.owner["correlation_id"]) == 200 + + +def _status(repo: Path) -> str: + return subprocess.run( + ["git", "status", "--porcelain=v2", "--untracked-files=all"], + cwd=repo, + check=True, + capture_output=True, + text=True, + ).stdout diff --git a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index cc340d7..e2720b1 100644 --- a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md +++ b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md @@ -146,6 +146,23 @@ signal cleanup, and close failure; a refused run changes neither repository nor remote refs; and the result carries a bounded transaction identifier and baseline evidence. +### Preparation — 2026-08-23 + +Added the production-inert core transaction primitive in +`rein_aharness/repository_transaction.py`. It keys a private external `flock` +by the resolved Git common directory, captures branch/detached `HEAD`, clean or +dirty index/worktree digests, upstream and remote-tracking refs, and exposes a +bounded transaction/baseline evidence envelope. Ten focused tests cover +cross-process same-repo contention, path aliases, distinct repos, dirty/staged/ +untracked baselines, detached `HEAD`, private lock metadata, exception cleanup, +and non-repository refusal. + +The primitive is intentionally not wired into `runner.py`, legacy approaches, +or the claim loop while ADR-002 still awaits Activity Core, sand-boxer, and +llm-connect acknowledgements. Lease-loss cancellation, moved-`HEAD` acceptance, +timeout/signal integration, and result-close reconciliation remain outstanding; +T02 therefore remains `wait`. + ## Verify accepted commits and reconcile metrics/reporting ```task