fix: make pull merges durably auditable (Closes #1028)

tests/test_pull_api.py

@@ -4,6 +4,7 @@ import httpx
 import pytest
 
 from src import gitea_proxy, main
+from src.security_event_store import SecurityEventStoreError
 
 
 @pytest.mark.anyio
@@ -312,11 +313,98 @@ async def test_assigned_pull_merge_requires_current_eligible_head(monkeypatch):
     assert calls == [("stackchain/api", 7, "abc123")]
 
 
+@pytest.mark.anyio
+async def test_assigned_pull_merge_fails_before_gitea_when_audit_reservation_fails(
+    monkeypatch,
+):
+    merge_calls = []
+
+    class UnavailableJournal:
+        def reserve(self, *_args, **_kwargs):
+            raise SecurityEventStoreError("unavailable")
+
+    async def assigned(*_args):
+        return True
+
+    async def merge(*args):
+        merge_calls.append(args)
+        return {"number": 7, "merged": True, "state": "closed"}
+
+    monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/v1/repos/stackchain/api/pulls/7/merge",
+            json={"expected_head_sha": "abc123"},
+        )
+
+    assert response.status_code == 503
+    assert response.json() == {
+        "error": "Security activity is temporarily unavailable. Nothing was merged."
+    }
+    assert merge_calls == []
+
+
+@pytest.mark.anyio
+async def test_assigned_pull_merge_reports_success_and_retains_pending_audit_when_finalize_fails(
+    monkeypatch,
+):
+    lifecycle = []
+
+    class InterruptedJournal:
+        def reserve(self, kind, *, target):
+            lifecycle.append(("reserve", kind, target))
+            return "merge-operation"
+
+        def finalize(self, operation_id):
+            lifecycle.append(("finalize", operation_id))
+            raise SecurityEventStoreError("unavailable")
+
+        def discard(self, operation_id):
+            lifecycle.append(("discard", operation_id))
+
+    async def assigned(*_args):
+        return True
+
+    async def merge(*_args):
+        return {"number": 7, "merged": True, "state": "closed"}
+
+    monkeypatch.setattr(main, "_security_event_store", lambda: InterruptedJournal())
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/v1/repos/stackchain/api/pulls/7/merge",
+            json={"expected_head_sha": "abc123"},
+        )
+
+    assert response.status_code == 200
+    assert response.json() == {"number": 7, "merged": True, "state": "closed"}
+    assert lifecycle == [
+        ("reserve", "pull_merged", "stackchain/api#7"),
+        ("finalize", "merge-operation"),
+    ]
+
+
 @pytest.mark.anyio
 async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypatch):
     calls = []
     merged = False
 
+    class LifecycleJournal:
+        def reserve(self, kind, *, target):
+            calls.append(("reserve", kind, target))
+            return "merge-operation"
+
+        def finalize(self, operation_id):
+            calls.append(("finalize", operation_id))
+
+        def discard(self, operation_id):
+            calls.append(("discard", operation_id))
+
     async def assigned(repository, number):
         return True
 
@@ -332,6 +420,7 @@ async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypa
         return merged
 
     monkeypatch.setattr(main, "ISSUE_ACTION_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr(main, "_security_event_store", lambda: LifecycleJournal())
     monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
     monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
     monkeypatch.setattr(main.gitea_proxy, "is_pull_merged_at_head", confirm, raising=False)
@@ -345,8 +434,10 @@ async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypa
     assert response.status_code == 200
     assert response.json() == {"number": 7, "merged": True, "state": "closed"}
     assert calls == [
+        ("reserve", "pull_merged", "stackchain/api#7"),
         ("merge", "stackchain/api", 7, "abc123"),
         ("confirm", "stackchain/api", 7, "abc123"),
+        ("finalize", "merge-operation"),
     ]
 
 
@@ -435,6 +526,47 @@ async def test_assigned_pull_merge_returns_conflict_without_mutating_stale_head(
     assert "New commits" in response.json()["error"]
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "rejection",
+    [gitea_proxy.StalePullError("changed"), gitea_proxy.PullNotMergeableError("unsafe")],
+)
+async def test_assigned_pull_merge_discards_audit_reservation_after_definite_rejection(
+    monkeypatch, rejection
+):
+    lifecycle = []
+
+    class LifecycleJournal:
+        def reserve(self, kind, *, target):
+            lifecycle.append(("reserve", kind, target))
+            return "merge-operation"
+
+        def discard(self, operation_id):
+            lifecycle.append(("discard", operation_id))
+
+    async def assigned(*_args):
+        return True
+
+    async def merge(*_args):
+        raise rejection
+
+    monkeypatch.setattr(main, "_security_event_store", lambda: LifecycleJournal())
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/v1/repos/stackchain/api/pulls/7/merge",
+            json={"expected_head_sha": "abc123"},
+        )
+
+    assert response.status_code == 409
+    assert lifecycle == [
+        ("reserve", "pull_merged", "stackchain/api#7"),
+        ("discard", "merge-operation"),
+    ]
+
+
 @pytest.mark.anyio
 async def test_gitea_merge_rejects_failed_ci_without_upstream_mutation():
     requests = []