src/main.py

@@ -5918,31 +5918,60 @@ async def merge_assigned_pull(
         target=f"{repository}#{number}",
     )
 
-    async def merge_pull():
-        if not await gitea_proxy.is_assigned_pull(repository, number):
-            raise HTTPException(status_code=404, detail="Assigned pull request not found")
-        return await gitea_proxy.merge_assigned_pull(
-            repository, number, submission.expected_head_sha
+    try:
+        assigned = await asyncio.wait_for(
+            gitea_proxy.is_assigned_pull(repository, number),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
+    except Exception:
+        return JSONResponse(
+            {"error": "The pull request assignment could not be verified. Nothing was merged."},
+            status_code=503,
+            headers={"Retry-After": "1"},
+        )
+    if not assigned:
+        raise HTTPException(status_code=404, detail="Assigned pull request not found")
+
+    journal = _security_event_store()
+    try:
+        operation_id = await asyncio.to_thread(
+            journal.reserve,
+            "pull_merged",
+            target=f"{repository}#{number}",
+        )
+    except SecurityEventStoreError:
+        return JSONResponse(
+            {"error": "Security activity is temporarily unavailable. Nothing was merged."},
+            status_code=503,
+            headers={"Retry-After": "1"},
         )
 
     try:
         result = await asyncio.wait_for(
-            merge_pull(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
-        )
-        await asyncio.to_thread(
-            _security_event_store().record,
-            "pull_merged",
-            target=f"{repository}#{number}",
+            gitea_proxy.merge_assigned_pull(
+                repository, number, submission.expected_head_sha
+            ),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
         )
+        try:
+            await asyncio.to_thread(journal.finalize, operation_id)
+        except SecurityEventStoreError:
+            pass
         return result
-    except HTTPException:
-        raise
     except gitea_proxy.StalePullError:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         return JSONResponse(
             {"error": "New commits were pushed. Refresh before merging."},
             status_code=409,
         )
     except gitea_proxy.PullNotMergeableError:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         return JSONResponse(
             {"error": "This pull request is not currently safe to merge. Refresh its status."},
             status_code=409,
@@ -5958,6 +5987,10 @@ async def merge_assigned_pull(
         except Exception:
             merged = False
         if merged:
+            try:
+                await asyncio.to_thread(journal.finalize, operation_id)
+            except SecurityEventStoreError:
+                pass
             return {"number": number, "merged": True, "state": "closed"}
         return JSONResponse(
             {

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 = []