feat: serialize autonomous release runs (#335)

README.md

@@ -310,6 +310,14 @@ without advancing the release. The equivalent environment variables are
 Durable state defaults to `.release-engine/state.json`. Successful checkpoints
 record the repository, issue branch, and exact commit SHA. A restart resumes at
 tests, push, or PR creation only after validating that identity and a clean
-worktree; a pushed checkpoint also verifies the remote branch SHA. Full
+worktree; a pushed checkpoint also verifies the remote branch SHA. Invocations
-behavior and safety gates are documented in
+are serialized by an OS-backed, non-blocking lease at
+`.release-engine/run.lock` under the canonical repository path. If an hourly
+run overlaps a manual or slow prior run, the contender exits with `release run
+already active` and owner diagnostics before Gitea discovery or any Git/agent
+work. The kernel releases ownership when the process exits, so never delete a
+lock file to recover from a crash; stale file metadata cannot retain the lease.
+Checkpoint replacement uses writer-unique, flushed temporary files and an
+atomic rename, preventing concurrent writers from colliding or exposing partial
+JSON. Full behavior and safety gates are documented in
 [`docs/release-engine-spec.md`](docs/release-engine-spec.md).

docs/release-engine-spec.md

@@ -44,6 +44,19 @@ repeating or advancing delivery.
 4. Otherwise sort by priority label (`P0`, `P1`, `P2`, unlabeled) then issue number.
 5. Select exactly one issue per invocation.
 
+## Single-active-run lease
+
+- Before issue discovery, every invocation acquires a non-blocking advisory lock
+  at `.release-engine/run.lock` under the canonical repository path.
+- The lease is held through the final checkpoint and pull-request request. A
+  contended invocation fails immediately with `release run already active`
+  before contacting Gitea or running Git, agent, or test commands.
+- The lock file records bounded owner PID and acquisition-time diagnostics.
+  These are informational only: kernel lock ownership is authoritative, so a
+  killed or crashed process releases the lease automatically and stale metadata
+  cannot block the next run.
+- Different canonical repository paths use independent leases.
+
 ## Claiming
 
 - PATCH the issue with the configured assignee.
@@ -78,6 +91,10 @@ repeating or advancing delivery.
 
 - `--dry-run` performs discovery and planning only: no claim, git mutation, agent command, push, or PR.
 - State is written atomically after each successful transition.
+- Each checkpoint write uses a writer-unique temporary file, flushes file data,
+  atomically replaces the checkpoint, and flushes its parent directory. A
+  failed or concurrent writer therefore cannot collide on a shared temp path or
+  expose partial JSON.
 - Resumable state carries repository and commit identity; successful stages are
   skipped only after local (and, after push, remote) Git verification.
 - One invocation handles at most one issue.
@@ -101,3 +118,5 @@ repeating or advancing delivery.
 13. Dirty or identity-mismatched checkpoints stop before push or PR creation.
 14. Parent credentials are absent from coding-agent, test, and Git subprocesses while explicit agent variables and issue context reach only the coding stage.
 15. Reserved or malformed agent environment allowlist entries abort before claim.
+16. An overlapping invocation fails before Gitea discovery or local command execution and reports bounded owner diagnostics.
+17. Concurrent checkpoint-save stress completes without temp-file collisions or partial state.

src/release_engine.py

@@ -2,12 +2,16 @@
 from __future__ import annotations
 
 import argparse
+import errno
+import fcntl
 import json
 import os
 import re
 import signal
 import shlex
 import subprocess
+import tempfile
+import time
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Any, Iterable, Protocol
@@ -97,9 +101,71 @@ class StateStore:
 
     def save(self, state: RunState) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        tmp = self.path.with_name(self.path.name + ".tmp")
+        temporary_path = None
-        tmp.write_text(json.dumps(asdict(state), indent=2) + "\n")
+        try:
-        tmp.replace(self.path)
+            with tempfile.NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                dir=self.path.parent,
+                prefix=self.path.name + ".",
+                suffix=".tmp",
+                delete=False,
+            ) as temporary:
+                temporary_path = Path(temporary.name)
+                temporary.write(json.dumps(asdict(state), indent=2) + "\n")
+                temporary.flush()
+                os.fsync(temporary.fileno())
+            os.replace(temporary_path, self.path)
+            directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
+        finally:
+            if temporary_path is not None:
+                temporary_path.unlink(missing_ok=True)
+
+
+class RepositoryLease:
+    """Non-blocking, process-backed single-writer lease for one worktree."""
+
+    def __init__(self, repo_path: Path):
+        self.path = Path(repo_path).resolve() / ".release-engine" / "run.lock"
+        self._file = None
+
+    def __enter__(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        lock_file = self.path.open("a+", encoding="utf-8")
+        try:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError as exc:
+            if exc.errno in (errno.EACCES, errno.EAGAIN):
+                detail = ""
+                try:
+                    lock_file.seek(0)
+                    metadata = json.loads(lock_file.read(512))
+                    owner_pid = int(metadata["pid"])
+                    acquired_at = float(metadata["acquired_at"])
+                    detail = f" (owner pid={owner_pid}, acquired_at={acquired_at:.3f})"
+                except (KeyError, TypeError, ValueError, json.JSONDecodeError):
+                    pass
+                lock_file.close()
+                raise RuntimeError(f"release run already active{detail}") from exc
+            lock_file.close()
+            raise
+        self._file = lock_file
+        lock_file.seek(0)
+        lock_file.truncate()
+        lock_file.write(json.dumps({"pid": os.getpid(), "acquired_at": time.time()}) + "\n")
+        lock_file.flush()
+        os.fsync(lock_file.fileno())
+        return self
+
+    def __exit__(self, exc_type, exc, traceback):
+        if self._file is not None:
+            fcntl.flock(self._file.fileno(), fcntl.LOCK_UN)
+            self._file.close()
+            self._file = None
 
 
 def _label_names(labels: Iterable[Any]) -> list[str]:
@@ -267,6 +333,10 @@ class ReleaseEngine:
         self.store = StateStore(config.state_path)
 
     def execute(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]:
+        with RepositoryLease(self.config.repo_path):
+            return self._execute_locked(issue_number=issue_number, dry_run=dry_run)
+
+    def _execute_locked(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]:
         deadlines = {
             "agent timeout seconds": self.config.agent_timeout_seconds,
             "test timeout seconds": self.config.test_timeout_seconds,

tests/test_release_engine.py

@@ -461,6 +461,51 @@ def test_dry_run_does_not_claim_or_execute(tmp_path):
     assert runner.commands == []
 
 
+def test_overlapping_run_fails_before_gitea_or_git_side_effects(tmp_path):
+    entered = threading.Event()
+    release_owner = threading.Event()
+
+    class BlockingClient(FakeClient):
+        def __init__(self):
+            super().__init__([])
+            self.discovery_calls = 0
+
+        def list_open_issues(self, repo):
+            self.discovery_calls += 1
+            if self.discovery_calls > 1:
+                raise AssertionError("contended run reached Gitea discovery")
+            entered.set()
+            release_owner.wait(timeout=2)
+            raise RuntimeError("stop lease owner")
+
+    client = BlockingClient()
+    owner_runner = FakeRunner([])
+    contender_runner = FakeRunner([])
+    owner = ReleaseEngine(config(tmp_path), client, owner_runner)
+    contender = ReleaseEngine(config(tmp_path), client, contender_runner)
+
+    def run_owner():
+        with pytest.raises(RuntimeError, match="stop lease owner"):
+            owner.execute(dry_run=True)
+
+    thread = threading.Thread(target=run_owner)
+    thread.start()
+    assert entered.wait(timeout=1)
+    try:
+        with pytest.raises(RuntimeError, match="release run already active") as error:
+            contender.execute(dry_run=True)
+        assert f"owner pid={os.getpid()}" in str(error.value)
+        assert "acquired_at=" in str(error.value)
+    finally:
+        release_owner.set()
+        thread.join(timeout=2)
+
+    assert not thread.is_alive()
+    assert client.discovery_calls == 1
+    assert owner_runner.commands == []
+    assert contender_runner.commands == []
+
+
 def test_existing_pr_is_reused_without_agent_execution(tmp_path):
     issue = Issue(14, "Engine MVP", "body", "open", [], "timmy")
     client = FakeClient([issue], existing_pr={"number": 9, "html_url": "https://forge/pr/9"})
@@ -482,6 +527,35 @@ def test_state_store_round_trip_is_atomic(tmp_path):
     assert not (tmp_path / "state.json.tmp").exists()
 
 
+def test_concurrent_checkpoint_writers_never_collide_or_leave_partial_state(tmp_path):
+    store = StateStore(tmp_path / "state.json")
+    states = [
+        RunState(issue=index, branch=f"timmy/{index}-engine", status="tests_passed")
+        for index in range(8)
+    ]
+    barrier = threading.Barrier(len(states))
+    errors = []
+
+    def save_repeatedly(state):
+        barrier.wait()
+        for _ in range(100):
+            try:
+                store.save(state)
+            except Exception as exc:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=save_repeatedly, args=(state,)) for state in states]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=5)
+
+    assert not any(thread.is_alive() for thread in threads)
+    assert errors == []
+    assert store.load() in states
+    assert list(tmp_path.glob("state.json.*.tmp")) == []
+
+
 def test_restart_from_agent_complete_skips_claim_and_coding(tmp_path):
     issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
     client = FakeClient([issue])