From 65d11fb0bc3a43594591546e4ee2186fc0ce3f29 Mon Sep 17 00:00:00 2001
From: FG Fernandez <26927750+frgfm@users.noreply.github.com>
Date: Sat, 10 Oct 2026 10:32:03 +0000
Subject: [PATCH 1/3] fix: commit multi-step CRUD workflows atomically

---
 src/app/api/api_v1/endpoints/alerts.py     |  12 +-
 src/app/api/api_v1/endpoints/detections.py |  38 ++++--
 src/app/api/api_v1/endpoints/sequences.py  |  23 ++--
 src/app/crud/base.py                       |  29 +++--
 src/app/crud/crud_sequence.py              |  10 +-
 src/app/services/alerts.py                 |   5 +-
 src/tests/endpoints/test_sequences.py      |   7 +-
 src/tests/test_crud_transactions.py        | 132 +++++++++++++++++++++
 8 files changed, 213 insertions(+), 43 deletions(-)
 create mode 100644 src/tests/test_crud_transactions.py

diff --git a/src/app/api/api_v1/endpoints/alerts.py b/src/app/api/api_v1/endpoints/alerts.py
index a2fd445a..41ee23e0 100644
--- a/src/app/api/api_v1/endpoints/alerts.py
+++ b/src/app/api/api_v1/endpoints/alerts.py
@@ -493,15 +493,14 @@ async def unmatch_alert_sequence(
         .where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     )
     await session.exec(delete_stmt)
-    await session.commit()
-
-    await refresh_alert_state(alert_id, session, alerts)
+    await refresh_alert_state(alert_id, session, alerts, commit=False)
 
     other_links_stmt: Any = (
         select(func.count()).select_from(AlertSequence).where(AlertSequence.sequence_id == sequence_id)
     )
     other_links = int((await session.exec(other_links_stmt)).one())
     if other_links > 0:
+        await session.commit()
         return None
 
     sequence = cast(Sequence, await sequences.get(sequence_id, strict=True))
@@ -518,7 +517,8 @@ async def unmatch_alert_sequence(
             last_seen_at=sequence.last_seen_at,
             lat=None,
             lon=None,
-        )
+        ),
+        commit=False,
     )
     session.add(AlertSequence(alert_id=new_alert.id, sequence_id=sequence_id))
     await session.commit()
@@ -543,6 +543,6 @@ async def delete_alert(
     # Delete associations
     delete_stmt: Any = delete(AlertSequence).where(AlertSequence.alert_id == cast(Any, alert_id))
     await session.exec(delete_stmt)
-    await session.commit()
     # Delete alert
-    await alerts.delete(alert_id)
+    await alerts.delete(alert_id, commit=False)
+    await session.commit()
diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py
index 37bcee36..91d110fb 100644
--- a/src/app/api/api_v1/endpoints/detections.py
+++ b/src/app/api/api_v1/endpoints/detections.py
@@ -145,7 +145,8 @@ async def _create_continuity_detection(
             bbox=EMPTY_BBOXES,
             sequence_id=sequence_id,
             recorded_at=recorded_at,
-        )
+        ),
+        commit=False,
     )
 
 
@@ -171,7 +172,7 @@ async def _attach_continuity_detections(
         created.append(
             await _create_continuity_detection(detections, camera_id, pose_id, bucket_key, seq.id, recorded_at)
         )
-        await sequences.enqueue_validation(seq.id)
+        await sequences.enqueue_validation(seq.id, commit=False)
     return created
 
 
@@ -297,6 +298,7 @@ async def _maybe_update_alert(
         await alerts.update(
             target_alert_id,
             AlertUpdate(lat=lat, lon=lon, started_at=new_start_at, last_seen_at=new_last_seen),
+            commit=False,
         )
 
 
@@ -357,7 +359,7 @@ async def _merge_alerts(
     await session.exec(delete_links_stmt)
     delete_alerts_stmt: Any = delete(Alert).where(cast(Any, Alert.id).in_(list(absorbed_ids)))
     await session.exec(delete_alerts_stmt)
-    await session.commit()
+    await session.flush()
 
 
 def _rewrite_after_merge(
@@ -435,7 +437,8 @@ async def _get_or_create_alert_id(
             lon=location[1] if isinstance(location, tuple) else None,
             started_at=start_at,
             last_seen_at=last_seen_at,
-        )
+        ),
+        commit=False,
     )
     return alert.id, set()
 
@@ -462,6 +465,8 @@ async def _attach_sequence_to_alert(
     sequences: SequenceCRUD,
     alerts: AlertCRUD,
     anchor_on_sequence: bool = False,
+    *,
+    commit: bool = True,
 ) -> Optional[int]:
     """Assign the given sequence to an alert based on cone/time overlap."""
     camera_by_id = await _get_camera_by_id(camera, cameras, sequence_.camera_id)
@@ -507,7 +512,10 @@ async def _attach_sequence_to_alert(
 
     if to_link:
         session.add_all(to_link)
+    if commit:
         await session.commit()
+    else:
+        await session.flush()
 
     return alert_id
 
@@ -589,6 +597,7 @@ async def create_detection(
         continuity_dets = await _attach_continuity_detections(
             detections, sequences, continuity_sequences, token_payload.sub, pose_id, bucket_key, effective_recorded_at
         )
+        await detections.session.commit()
         return DetectionRead(**continuity_dets[0].model_dump())
 
     # Upload media
@@ -619,7 +628,8 @@ async def create_detection(
                 bbox=single_bboxes,
                 others_bboxes=others_bboxes,
                 recorded_at=effective_recorded_at,
-            )
+            ),
+            commit=False,
         )
 
         det_bbox = _parse_bbox(bbox_str)
@@ -645,12 +655,12 @@ async def create_detection(
                 break
 
         if matched_sequence is not None:
-            await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at))
-            det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id))
+            await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at), commit=False)
+            det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id), commit=False)
             # Only the primary bbox tracks the sequence; siblings in others_bboxes are unrelated detections.
             det_max_conf = max_conf_from_bboxes(det.bbox)
             if det_max_conf is not None:
-                await sequences.bump_max_conf(matched_sequence.id, det_max_conf)
+                await sequences.bump_max_conf(matched_sequence.id, det_max_conf, commit=False)
             affected_sequences.add(matched_sequence.id)
         else:
             det_filters: List[tuple[str, Any]] = [
@@ -694,10 +704,13 @@ async def create_detection(
                         started_at=first_det.recorded_at,
                         last_seen_at=det.created_at,
                         max_conf=seq_max_conf,
-                    )
+                    ),
+                    commit=False,
                 )
                 for det_ in overlapping_dets:
-                    updated = await detections.update(det_.id, DetectionSequence(sequence_id=sequence_.id))
+                    updated = await detections.update(
+                        det_.id, DetectionSequence(sequence_id=sequence_.id), commit=False
+                    )
                     if det_.id == det.id:
                         det = updated
                 affected_sequences.add(sequence_.id)
@@ -724,7 +737,10 @@ async def create_detection(
     # claims due sequences from the DB and runs the gated pipeline: triangulation and ALL
     # notification channels (webhooks, Telegram, Slack) fire only once validated.
     for seq_id in affected_sequences:
-        await sequences.enqueue_validation(seq_id)
+        await sequences.enqueue_validation(seq_id, commit=False)
+
+    # Detection rows, sequence membership/confidence and validation jobs become visible together.
+    await detections.session.commit()
 
     first_det = cast(Detection, await detections.get(created[0].id, strict=True))
     return DetectionRead(**first_det.model_dump())
diff --git a/src/app/api/api_v1/endpoints/sequences.py b/src/app/api/api_v1/endpoints/sequences.py
index c2a18ee5..e097878b 100644
--- a/src/app/api/api_v1/endpoints/sequences.py
+++ b/src/app/api/api_v1/endpoints/sequences.py
@@ -259,16 +259,16 @@ async def delete_sequence(
     # Unset the sequence_id in the detections
     det_ids = await session.exec(select(Detection.id).where(Detection.sequence_id == sequence_id))
     for det_id in det_ids.all():
-        await detections.update(det_id, DetectionSequence(sequence_id=None))
+        await detections.update(det_id, DetectionSequence(sequence_id=None), commit=False)
     # Drop alert links for this sequence to avoid FK issues
     delete_stmt: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     await session.exec(delete_stmt)
-    await session.commit()
     # Delete the sequence
-    await sequences.delete(sequence_id)
+    await sequences.delete(sequence_id, commit=False)
     # Refresh affected alerts
     for aid in alert_ids:
-        await refresh_alert_state(aid, session, alerts)
+        await refresh_alert_state(aid, session, alerts, commit=False)
+    await session.commit()
 
 
 @router.patch("/{sequence_id}/label", status_code=status.HTTP_200_OK, summary="Label the nature of the sequence")
@@ -288,7 +288,7 @@ async def label_sequence(
         await verify_org_rights(token_payload.organization_id, sequence.camera_id, cameras)
 
     previous_label = sequence.is_wildfire
-    updated = await sequences.update(sequence_id, payload)
+    updated = await sequences.update(sequence_id, payload, commit=False)
 
     # Reverting a non-wildfire label: re-run cone matching. The sequence keeps its lonely alert
     # until the matching merges it into an overlapping one, so it is never left without an alert.
@@ -300,10 +300,14 @@ async def label_sequence(
     ):
         camera = cast(Camera, await cameras.get(sequence.camera_id, strict=True))
         async with _organization_alert_lock(camera.organization_id):
-            await _attach_sequence_to_alert(updated, camera, cameras, sequences, alerts, anchor_on_sequence=True)
+            await _attach_sequence_to_alert(
+                updated, camera, cameras, sequences, alerts, anchor_on_sequence=True, commit=False
+            )
+            await session.commit()
         return updated
 
     if payload.is_wildfire is None or payload.is_wildfire == AnnotationType.WILDFIRE_SMOKE:
+        await session.commit()
         return updated
 
     alert_ids_res = await session.exec(select(AlertSequence.alert_id).where(AlertSequence.sequence_id == sequence_id))
@@ -320,13 +324,13 @@ async def label_sequence(
         )
         siblings_res = await session.exec(siblings_stmt)
         if siblings_res.first() is None:
+            await session.commit()
             return updated
 
         delete_links: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
         await session.exec(delete_links)
-        await session.commit()
         for aid in alert_ids:
-            await refresh_alert_state(aid, session, alerts)
+            await refresh_alert_state(aid, session, alerts, commit=False)
 
     # Create a fresh alert for this sequence alone
     camera = cast(Camera, await cameras.get(sequence.camera_id, strict=True))
@@ -337,7 +341,8 @@ async def label_sequence(
             last_seen_at=sequence.last_seen_at,
             lat=None,
             lon=None,
-        )
+        ),
+        commit=False,
     )
     session.add(AlertSequence(alert_id=new_alert.id, sequence_id=sequence_id))
     await session.commit()
diff --git a/src/app/crud/base.py b/src/app/crud/base.py
index e7a79482..939b7bef 100644
--- a/src/app/crud/base.py
+++ b/src/app/crud/base.py
@@ -23,18 +23,27 @@ def __init__(self, session: AsyncSession, model: Type[ModelType]) -> None:
         self.session = session
         self.model = model
 
-    async def create(self, payload: CreateSchemaType) -> ModelType:
+    async def create(self, payload: CreateSchemaType, *, commit: bool = True) -> ModelType:
+        """Insert a row, optionally leaving the transaction to the caller.
+
+        Deferred writes are flushed so generated IDs/defaults and subsequent queries work.
+        The caller must commit the complete workflow; session cleanup rolls it back on error.
+        """
         entry = self.model(**payload.model_dump())
         try:
             self.session.add(entry)
-            await self.session.commit()
+            if commit:
+                await self.session.commit()
+            else:
+                await self.session.flush()
         except exc.IntegrityError as error:
             await self.session.rollback()
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
                 detail=f"An entry with the same index already exists : {error!s}",
             )
-        await self.session.refresh(entry)
+        if commit:
+            await self.session.refresh(entry)
 
         return entry
 
@@ -105,7 +114,7 @@ async def fetch_all(
         result = await self.session.exec(statement=statement)
         return [r for r in result]
 
-    async def update(self, entry_id: int, payload: UpdateSchemaType) -> ModelType:
+    async def update(self, entry_id: int, payload: UpdateSchemaType, *, commit: bool = True) -> ModelType:
         access = cast(ModelType, await self.get(entry_id, strict=True))
         values = payload.model_dump(exclude_unset=True)
 
@@ -113,17 +122,21 @@ async def update(self, entry_id: int, payload: UpdateSchemaType) -> ModelType:
             setattr(access, k, v)
 
         self.session.add(access)
-        await self.session.commit()
-        await self.session.refresh(access)
+        if commit:
+            await self.session.commit()
+            await self.session.refresh(access)
+        else:
+            await self.session.flush()
 
         return access
 
-    async def delete(self, entry_id: int) -> None:
+    async def delete(self, entry_id: int, *, commit: bool = True) -> None:
         await self.get(entry_id, strict=True)
         statement = delete(self.model).where(cast(Any, self.model).id == entry_id)  # ty: ignore[invalid-argument-type]
 
         await self.session.exec(statement=statement)  # type: ignore[call-overload]
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def get_in(self, list_: List[Any], field_name: str) -> List[ModelType]:
         statement: Any = select(self.model).where(getattr(self.model, field_name).in_(list_))
diff --git a/src/app/crud/crud_sequence.py b/src/app/crud/crud_sequence.py
index 8e8a70b6..1bc480ac 100644
--- a/src/app/crud/crud_sequence.py
+++ b/src/app/crud/crud_sequence.py
@@ -26,7 +26,7 @@ class SequenceCRUD(BaseCRUD[Sequence, Sequence, Union[SequenceUpdate, SequenceLa
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, Sequence)
 
-    async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
+    async def bump_max_conf(self, sequence_id: int, candidate: float, *, commit: bool = True) -> None:
         """Atomically raise sequences.max_conf to candidate if higher (or set if NULL).
 
         Uses a portable CASE expression so it runs on SQLite as well as Postgres.
@@ -38,7 +38,8 @@ async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
         )
         stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(max_conf=bumped)
         await self.session.exec(stmt)
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def set_temporal_score(
         self,
@@ -83,7 +84,7 @@ async def claim_validation(self, sequence_id: int, validation_status: Union[str,
         await self.session.commit()
         return bool(getattr(result, "rowcount", 0))
 
-    async def enqueue_validation(self, sequence_id: int) -> None:
+    async def enqueue_validation(self, sequence_id: int, *, commit: bool = True) -> None:
         """Mark the sequence as due for temporal validation (the DB-backed queue).
 
         Idempotent and FIFO-preserving: ``COALESCE`` keeps the oldest due timestamp, so a
@@ -101,7 +102,8 @@ async def enqueue_validation(self, sequence_id: int) -> None:
             .values(validation_due_at=func.coalesce(due_col, utcnow()))
         )
         await self.session.exec(stmt)
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def claim_due_validation(self, lease_seconds: float) -> Union[Sequence, None]:
         """Claim the oldest due sequence for validation, or None when nothing is due.
diff --git a/src/app/services/alerts.py b/src/app/services/alerts.py
index 952b71ac..de60f103 100644
--- a/src/app/services/alerts.py
+++ b/src/app/services/alerts.py
@@ -17,7 +17,7 @@
 __all__ = ["refresh_alert_state"]
 
 
-async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: AlertCRUD) -> None:
+async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: AlertCRUD, *, commit: bool = True) -> None:
     """Recompute an alert's bounds and location from its remaining sequences, or delete it if empty."""
     remaining_stmt: Any = (
         select(Sequence, Camera)
@@ -28,7 +28,7 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler
     remaining_res = await session.exec(remaining_stmt)
     rows = remaining_res.all()
     if not rows:
-        await alerts.delete(alert_id)
+        await alerts.delete(alert_id, commit=commit)
         return
 
     seqs = [row[0] for row in rows]
@@ -62,4 +62,5 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler
             lat=loc[0] if loc else None,
             lon=loc[1] if loc else None,
         ),
+        commit=commit,
     )
diff --git a/src/tests/endpoints/test_sequences.py b/src/tests/endpoints/test_sequences.py
index b4bd63ee..d74bfb7c 100644
--- a/src/tests/endpoints/test_sequences.py
+++ b/src/tests/endpoints/test_sequences.py
@@ -753,11 +753,11 @@ async def test_unit_label_sequence_as_other_smoke_refreshes_alert(
 
     # 3. Assertions
     mock_sequences_crud.get.assert_called_once_with(1, strict=True)
-    mock_sequences_crud.update.assert_called_once_with(1, payload)
+    mock_sequences_crud.update.assert_called_once_with(1, payload, commit=False)
 
     # Three session.exec calls: alert_ids lookup, siblings probe, delete links
     assert mock_session.exec.call_count == 3
-    mock_refresh_alert_state.assert_called_once_with(101, mock_session, mock_alerts_crud)
+    mock_refresh_alert_state.assert_called_once_with(101, mock_session, mock_alerts_crud, commit=False)
 
     # Verify a new alert was created for this sequence
     mock_alerts_crud.create.assert_called_once()
@@ -813,6 +813,7 @@ async def test_unit_relabel_sequence_to_wildfire_smoke_reattaches(
         mock_sequences_crud,
         mock_alerts_crud,
         anchor_on_sequence=True,
+        commit=False,
     )
     mock_session.exec.assert_not_called()
     mock_alerts_crud.create.assert_not_called()
@@ -899,7 +900,7 @@ async def test_unit_label_sequence_as_wildfire_smoke_does_not_refresh():
     )
 
     # 3. Assertions
-    mock_sequences_crud.update.assert_called_once_with(1, payload)
+    mock_sequences_crud.update.assert_called_once_with(1, payload, commit=False)
     mock_session.exec.assert_not_called()
     mock_alerts_crud.create.assert_not_called()
 
diff --git a/src/tests/test_crud_transactions.py b/src/tests/test_crud_transactions.py
new file mode 100644
index 00000000..b9426065
--- /dev/null
+++ b/src/tests/test_crud_transactions.py
@@ -0,0 +1,132 @@
+from datetime import timedelta
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlmodel import func, select
+
+from app.core.time import utcnow
+from app.crud import AlertCRUD, OrganizationCRUD, SequenceCRUD
+from app.db import session_factory
+from app.models import Alert, AlertSequence, Detection, Organization, Sequence
+from app.schemas.organizations import OrganizationCreate, SlackHook
+
+
+@pytest.mark.asyncio
+async def test_deferred_create_is_flushed_but_not_visible_until_commit(async_session):
+    crud = OrganizationCRUD(async_session)
+    organization = await crud.create(OrganizationCreate(name="deferred-org"), commit=False)
+    assert organization.id is not None
+    assert await crud.get(organization.id) is organization
+    async with session_factory() as observer:
+        assert await observer.get(Organization, organization.id) is None
+    await async_session.rollback()
+    async with session_factory() as observer:
+        assert await observer.get(Organization, organization.id) is None
+
+    committed = await crud.create(OrganizationCreate(name="committed-org"))
+    async with session_factory() as observer:
+        assert await observer.get(Organization, committed.id) is not None
+
+
+@pytest.mark.parametrize("operation", ["update", "delete"])
+@pytest.mark.asyncio
+async def test_deferred_mutations_roll_back(organization_session, operation):
+    crud = OrganizationCRUD(organization_session)
+    if operation == "update":
+        await crud.update(1, SlackHook(slack_hook="https://hooks.slack.com/services/TEST/TEST/test"), commit=False)
+    else:
+        await crud.delete(1, commit=False)
+    async with session_factory() as observer:
+        original = await observer.get(Organization, 1)
+        assert original is not None
+        assert original.slack_hook == pytest.organization_table[0]["slack_hook"]
+    await organization_session.rollback()
+    async with session_factory() as observer:
+        original = await observer.get(Organization, 1)
+        assert original is not None
+        assert original.slack_hook == pytest.organization_table[0]["slack_hook"]
+
+
+@pytest.mark.asyncio
+async def test_ingestion_rolls_back_rows_membership_confidence_and_queue_on_late_failure(
+    async_client, detection_session, monkeypatch
+):
+    from app.api.api_v1.endpoints import detections as endpoint
+
+    now = utcnow()
+    sequence = Sequence(camera_id=1, pose_id=1, camera_azimuth=45, started_at=now, last_seen_at=now, max_conf=0.1)
+    detection_session.add(sequence)
+    await detection_session.flush()
+    detection_session.add(
+        Detection(camera_id=1, pose_id=1, sequence_id=sequence.id, bucket_key="before", bbox="[(0.1,0.1,0.2,0.2,0.1)]")
+    )
+    await detection_session.commit()
+    original_count = (await detection_session.exec(select(func.count()).select_from(Detection))).one()
+    sequence_id = sequence.id
+
+    monkeypatch.setattr(endpoint, "upload_file", AsyncMock(return_value="new-frame"))
+    monkeypatch.setattr(SequenceCRUD, "enqueue_validation", AsyncMock(side_effect=RuntimeError("late queue failure")))
+    with pytest.raises(RuntimeError, match="late queue failure"):
+        await async_client.post(
+            "/detections",
+            headers=pytest.get_token(1, ["camera"], 1),
+            data={"pose_id": 1, "bboxes": "[(0.1,0.1,0.2,0.2,0.9),(0.7,0.7,0.8,0.8,0.8)]"},
+            files={"file": ("frame.jpg", b"frame", "image/jpeg")},
+        )
+    async with session_factory() as observer:
+        assert (await observer.exec(select(func.count()).select_from(Detection))).one() == original_count
+        persisted = await observer.get(Sequence, sequence_id)
+        assert persisted.last_seen_at == now
+        assert persisted.max_conf == pytest.approx(0.1)
+        assert persisted.validation_due_at is None
+
+
+@pytest.mark.parametrize("operation", ["label", "unmatch", "delete-sequence", "delete-alert"])
+@pytest.mark.asyncio
+async def test_alert_workflows_roll_back_on_late_failure(async_client, detection_session, monkeypatch, operation):
+    from app.api.api_v1.endpoints import sequences as endpoint
+
+    now = utcnow()
+    sequences = [
+        Sequence(camera_id=1, pose_id=1, camera_azimuth=45, started_at=now - timedelta(seconds=i), last_seen_at=now)
+        for i in range(2)
+    ]
+    alert = Alert(organization_id=1, started_at=now - timedelta(seconds=1), last_seen_at=now)
+    detection_session.add_all([alert, *sequences])
+    await detection_session.flush()
+    detection_session.add_all([AlertSequence(alert_id=alert.id, sequence_id=seq.id) for seq in sequences])
+    detection = Detection(camera_id=1, pose_id=1, sequence_id=sequences[0].id, bucket_key="before", bbox="[]")
+    detection_session.add(detection)
+    await detection_session.commit()
+    alert_id, sequence_id, detection_id = alert.id, sequences[0].id, detection.id
+    original_pairs = {(alert_id, seq.id) for seq in sequences}
+
+    fail = AsyncMock(side_effect=RuntimeError("late workflow failure"))
+
+    if operation in {"label", "unmatch"}:
+        monkeypatch.setattr(AlertCRUD, "create", fail)
+    elif operation == "delete-sequence":
+        monkeypatch.setattr(endpoint, "refresh_alert_state", fail)
+    else:
+        monkeypatch.setattr(AlertCRUD, "delete", fail)
+    auth = pytest.get_token(1, ["admin"], 1)
+    if operation == "label":
+        request = async_client.patch(
+            f"/sequences/{sequence_id}/label", headers=auth, json={"is_wildfire": "other_smoke"}
+        )
+    elif operation == "unmatch":
+        request = async_client.post(f"/alerts/{alert_id}/sequences/{sequence_id}/unmatch", headers=auth)
+    elif operation == "delete-sequence":
+        request = async_client.delete(f"/sequences/{sequence_id}", headers=auth)
+    else:
+        request = async_client.delete(f"/alerts/{alert_id}", headers=auth)
+    with pytest.raises(RuntimeError, match="late workflow failure"):
+        await request
+    async with session_factory() as observer:
+        assert await observer.get(Alert, alert_id) is not None
+        persisted = await observer.get(Sequence, sequence_id)
+        assert persisted is not None
+        assert persisted.is_wildfire is None
+        assert (await observer.get(Detection, detection_id)).sequence_id == sequence_id
+        links = (await observer.exec(select(AlertSequence).where(AlertSequence.alert_id == alert_id))).all()
+        assert {(link.alert_id, link.sequence_id) for link in links} == original_pairs

From cf9a8f750d1f4274692a38a0f1e4695351a5b697 Mon Sep 17 00:00:00 2001
From: FG Fernandez <26927750+frgfm@users.noreply.github.com>
Date: Thu, 15 Oct 2026 12:04:34 +0000
Subject: [PATCH 2/3] refactor: simplify atomic writes and reuse regression
 fixtures

---
 src/app/api/api_v1/endpoints/sequences.py |  10 +-
 src/app/crud/base.py                      |  33 ++----
 src/tests/endpoints/test_alerts.py        |  33 ++++++
 src/tests/endpoints/test_detections.py    |  29 +++++
 src/tests/test_crud_transactions.py       | 132 ----------------------
 5 files changed, 77 insertions(+), 160 deletions(-)
 delete mode 100644 src/tests/test_crud_transactions.py

diff --git a/src/app/api/api_v1/endpoints/sequences.py b/src/app/api/api_v1/endpoints/sequences.py
index e097878b..bc66775b 100644
--- a/src/app/api/api_v1/endpoints/sequences.py
+++ b/src/app/api/api_v1/endpoints/sequences.py
@@ -9,7 +9,7 @@
 from typing import Any, List, Union, cast
 
 from fastapi import APIRouter, Depends, HTTPException, Path, Query, Response, Security, status
-from sqlmodel import delete, func, select
+from sqlmodel import delete, func, select, update
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.api.api_v1.endpoints.detections import _attach_sequence_to_alert
@@ -19,7 +19,7 @@
 from app.db import get_session
 from app.models import AlertSequence, AnnotationType, Camera, Detection, Sequence, UserRole
 from app.schemas.alerts import AlertCreate
-from app.schemas.detections import DetectionSequence, DetectionWithUrl
+from app.schemas.detections import DetectionWithUrl
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceLabel, SequenceRead
 from app.services.alerts import refresh_alert_state
@@ -248,7 +248,6 @@ async def fetch_sequences_from_date(
 async def delete_sequence(
     sequence_id: int = Path(..., gt=0),
     sequences: SequenceCRUD = Depends(get_sequence_crud),
-    detections: DetectionCRUD = Depends(get_detection_crud),
     alerts: AlertCRUD = Depends(get_alert_crud),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN]),
@@ -257,9 +256,8 @@ async def delete_sequence(
     alert_ids_res = await session.exec(select(AlertSequence.alert_id).where(AlertSequence.sequence_id == sequence_id))
     alert_ids = list(alert_ids_res.all())
     # Unset the sequence_id in the detections
-    det_ids = await session.exec(select(Detection.id).where(Detection.sequence_id == sequence_id))
-    for det_id in det_ids.all():
-        await detections.update(det_id, DetectionSequence(sequence_id=None), commit=False)
+    unlink: Any = update(Detection).where(cast(Any, Detection.sequence_id) == sequence_id).values(sequence_id=None)
+    await session.exec(unlink)
     # Drop alert links for this sequence to avoid FK issues
     delete_stmt: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     await session.exec(delete_stmt)
diff --git a/src/app/crud/base.py b/src/app/crud/base.py
index 939b7bef..9a5d8a86 100644
--- a/src/app/crud/base.py
+++ b/src/app/crud/base.py
@@ -23,29 +23,25 @@ def __init__(self, session: AsyncSession, model: Type[ModelType]) -> None:
         self.session = session
         self.model = model
 
-    async def create(self, payload: CreateSchemaType, *, commit: bool = True) -> ModelType:
-        """Insert a row, optionally leaving the transaction to the caller.
+    async def _save(self, entry: ModelType, *, commit: bool) -> ModelType:
+        self.session.add(entry)
+        await self.session.flush()
+        if commit:
+            await self.session.commit()
+            await self.session.refresh(entry)
+        return entry
 
-        Deferred writes are flushed so generated IDs/defaults and subsequent queries work.
-        The caller must commit the complete workflow; session cleanup rolls it back on error.
-        """
+    async def create(self, payload: CreateSchemaType, *, commit: bool = True) -> ModelType:
+        """Flush a new row; pass commit=False to include it in a larger transaction."""
         entry = self.model(**payload.model_dump())
         try:
-            self.session.add(entry)
-            if commit:
-                await self.session.commit()
-            else:
-                await self.session.flush()
+            return await self._save(entry, commit=commit)
         except exc.IntegrityError as error:
             await self.session.rollback()
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
                 detail=f"An entry with the same index already exists : {error!s}",
             )
-        if commit:
-            await self.session.refresh(entry)
-
-        return entry
 
     async def get(self, entry_id: int, strict: bool = False) -> Union[ModelType, None]:
         entry: Union[ModelType, None] = await self.session.get(self.model, entry_id)
@@ -121,14 +117,7 @@ async def update(self, entry_id: int, payload: UpdateSchemaType, *, commit: bool
         for k, v in values.items():
             setattr(access, k, v)
 
-        self.session.add(access)
-        if commit:
-            await self.session.commit()
-            await self.session.refresh(access)
-        else:
-            await self.session.flush()
-
-        return access
+        return await self._save(access, commit=commit)
 
     async def delete(self, entry_id: int, *, commit: bool = True) -> None:
         await self.get(entry_id, strict=True)
diff --git a/src/tests/endpoints/test_alerts.py b/src/tests/endpoints/test_alerts.py
index 8b26a406..638ab28c 100644
--- a/src/tests/endpoints/test_alerts.py
+++ b/src/tests/endpoints/test_alerts.py
@@ -7,6 +7,7 @@
 import io
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Tuple, cast
+from unittest.mock import AsyncMock
 
 import pytest  # type: ignore
 from httpx import AsyncClient
@@ -16,6 +17,7 @@
 from app.api.api_v1.endpoints.alerts import _ALERT_EXPORT_COLUMNS, _iter_alerts_csv
 from app.core.config import settings
 from app.core.time import utcnow
+from app.db import session_factory
 from app.models import Alert, AlertSequence, AnnotationType, Camera, Detection, Organization, Pose, Sequence
 from app.services.overlap import compute_overlap
 
@@ -976,3 +978,34 @@ async def test_alerts_export_invalid_range(
 async def test_alerts_export_unauthenticated(async_client: AsyncClient, detection_session: AsyncSession):
     resp = await async_client.get("/alerts/export?from_date=2026-04-10&to_date=2026-04-12")
     assert resp.status_code == 401, resp.text
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "failure"),
+    [
+        ("PATCH", "/sequences/{sid}/label", "app.crud.AlertCRUD.create"),
+        ("POST", "/alerts/{aid}/sequences/{sid}/unmatch", "app.crud.AlertCRUD.create"),
+        ("DELETE", "/sequences/{sid}", "app.api.api_v1.endpoints.sequences.refresh_alert_state"),
+        ("DELETE", "/alerts/{aid}", "app.crud.AlertCRUD.delete"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_alert_changes_roll_back_on_late_failure(
+    async_client, detection_session, monkeypatch, method, path, failure
+):
+    alert, sequence_ids, counts = await _create_alert_with_sequences(detection_session, 1, 1, 48, 2)
+    sid = sequence_ids[0]
+    before = (await detection_session.get(Sequence, sid)).model_dump()
+    monkeypatch.setattr(failure, AsyncMock(side_effect=RuntimeError("late failure")))
+    kwargs = {"json": {"is_wildfire": "other_smoke"}} if method == "PATCH" else {}
+    with pytest.raises(RuntimeError, match="late failure"):
+        await async_client.request(
+            method, path.format(aid=alert.id, sid=sid), headers=pytest.get_token(1, ["admin"], 1), **kwargs
+        )
+    async with session_factory() as observer:
+        assert await observer.get(Alert, alert.id) is not None
+        assert (await observer.get(Sequence, sid)).model_dump() == before
+        links = (await observer.exec(select(AlertSequence.sequence_id).where(AlertSequence.alert_id == alert.id))).all()
+        assert set(links) == set(sequence_ids)
+        detections = (await observer.exec(select(Detection).where(Detection.sequence_id == sid))).all()
+        assert len(detections) == counts[0]
diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py
index fb3a107c..fb1a4a34 100644
--- a/src/tests/endpoints/test_detections.py
+++ b/src/tests/endpoints/test_detections.py
@@ -5,6 +5,7 @@
 from collections import Counter
 from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, List, Union
+from unittest.mock import AsyncMock
 
 import pytest  # type: ignore
 from fastapi import HTTPException, UploadFile
@@ -34,6 +35,7 @@
 from app.core.config import settings
 from app.core.time import utcnow
 from app.crud import AlertCRUD, CameraCRUD, DetectionCRUD, OrganizationCRUD, PoseCRUD, SequenceCRUD
+from app.db import session_factory
 from app.models import (
     Alert,
     AlertSequence,
@@ -2319,3 +2321,30 @@ async def fake_upload_file(file: UploadFile, organization_id: int, camera_id: in
     assert response.status_code == 403, response.text
     assert response.json()["detail"] == "Access forbidden."
     assert upload_calls == []
+
+
+@pytest.mark.asyncio
+async def test_ingestion_rolls_back_on_late_queue_failure(async_client, detection_session, monkeypatch):
+    sequence = await detection_session.get(Sequence, 1)
+    sequence.last_seen_at, sequence.max_conf, sequence.is_validated = utcnow(), 0.1, False
+    await detection_session.commit()
+    before = sequence.model_dump()
+    original_ids = (await detection_session.exec(select(Detection.id))).all()
+    enqueue = SequenceCRUD.enqueue_validation
+
+    async def fail_enqueue(self, sequence_id, **kwargs):
+        await enqueue(self, sequence_id, **kwargs)
+        raise RuntimeError("late queue failure")
+
+    monkeypatch.setattr(detections_api, "upload_file", AsyncMock(return_value="new-frame"))
+    monkeypatch.setattr(SequenceCRUD, "enqueue_validation", fail_enqueue)
+    with pytest.raises(RuntimeError, match="late queue failure"):
+        await async_client.post(
+            "/detections",
+            headers=pytest.get_token(1, ["camera"], 1),
+            data={"pose_id": 1, "bboxes": "[(0.1,0.1,0.2,0.2,0.9)]"},
+            files={"file": ("frame.jpg", b"frame", "image/jpeg")},
+        )
+    async with session_factory() as observer:
+        assert (await observer.exec(select(Detection.id))).all() == original_ids
+        assert (await observer.get(Sequence, 1)).model_dump() == before
diff --git a/src/tests/test_crud_transactions.py b/src/tests/test_crud_transactions.py
deleted file mode 100644
index b9426065..00000000
--- a/src/tests/test_crud_transactions.py
+++ /dev/null
@@ -1,132 +0,0 @@
-from datetime import timedelta
-from unittest.mock import AsyncMock
-
-import pytest
-from sqlmodel import func, select
-
-from app.core.time import utcnow
-from app.crud import AlertCRUD, OrganizationCRUD, SequenceCRUD
-from app.db import session_factory
-from app.models import Alert, AlertSequence, Detection, Organization, Sequence
-from app.schemas.organizations import OrganizationCreate, SlackHook
-
-
-@pytest.mark.asyncio
-async def test_deferred_create_is_flushed_but_not_visible_until_commit(async_session):
-    crud = OrganizationCRUD(async_session)
-    organization = await crud.create(OrganizationCreate(name="deferred-org"), commit=False)
-    assert organization.id is not None
-    assert await crud.get(organization.id) is organization
-    async with session_factory() as observer:
-        assert await observer.get(Organization, organization.id) is None
-    await async_session.rollback()
-    async with session_factory() as observer:
-        assert await observer.get(Organization, organization.id) is None
-
-    committed = await crud.create(OrganizationCreate(name="committed-org"))
-    async with session_factory() as observer:
-        assert await observer.get(Organization, committed.id) is not None
-
-
-@pytest.mark.parametrize("operation", ["update", "delete"])
-@pytest.mark.asyncio
-async def test_deferred_mutations_roll_back(organization_session, operation):
-    crud = OrganizationCRUD(organization_session)
-    if operation == "update":
-        await crud.update(1, SlackHook(slack_hook="https://hooks.slack.com/services/TEST/TEST/test"), commit=False)
-    else:
-        await crud.delete(1, commit=False)
-    async with session_factory() as observer:
-        original = await observer.get(Organization, 1)
-        assert original is not None
-        assert original.slack_hook == pytest.organization_table[0]["slack_hook"]
-    await organization_session.rollback()
-    async with session_factory() as observer:
-        original = await observer.get(Organization, 1)
-        assert original is not None
-        assert original.slack_hook == pytest.organization_table[0]["slack_hook"]
-
-
-@pytest.mark.asyncio
-async def test_ingestion_rolls_back_rows_membership_confidence_and_queue_on_late_failure(
-    async_client, detection_session, monkeypatch
-):
-    from app.api.api_v1.endpoints import detections as endpoint
-
-    now = utcnow()
-    sequence = Sequence(camera_id=1, pose_id=1, camera_azimuth=45, started_at=now, last_seen_at=now, max_conf=0.1)
-    detection_session.add(sequence)
-    await detection_session.flush()
-    detection_session.add(
-        Detection(camera_id=1, pose_id=1, sequence_id=sequence.id, bucket_key="before", bbox="[(0.1,0.1,0.2,0.2,0.1)]")
-    )
-    await detection_session.commit()
-    original_count = (await detection_session.exec(select(func.count()).select_from(Detection))).one()
-    sequence_id = sequence.id
-
-    monkeypatch.setattr(endpoint, "upload_file", AsyncMock(return_value="new-frame"))
-    monkeypatch.setattr(SequenceCRUD, "enqueue_validation", AsyncMock(side_effect=RuntimeError("late queue failure")))
-    with pytest.raises(RuntimeError, match="late queue failure"):
-        await async_client.post(
-            "/detections",
-            headers=pytest.get_token(1, ["camera"], 1),
-            data={"pose_id": 1, "bboxes": "[(0.1,0.1,0.2,0.2,0.9),(0.7,0.7,0.8,0.8,0.8)]"},
-            files={"file": ("frame.jpg", b"frame", "image/jpeg")},
-        )
-    async with session_factory() as observer:
-        assert (await observer.exec(select(func.count()).select_from(Detection))).one() == original_count
-        persisted = await observer.get(Sequence, sequence_id)
-        assert persisted.last_seen_at == now
-        assert persisted.max_conf == pytest.approx(0.1)
-        assert persisted.validation_due_at is None
-
-
-@pytest.mark.parametrize("operation", ["label", "unmatch", "delete-sequence", "delete-alert"])
-@pytest.mark.asyncio
-async def test_alert_workflows_roll_back_on_late_failure(async_client, detection_session, monkeypatch, operation):
-    from app.api.api_v1.endpoints import sequences as endpoint
-
-    now = utcnow()
-    sequences = [
-        Sequence(camera_id=1, pose_id=1, camera_azimuth=45, started_at=now - timedelta(seconds=i), last_seen_at=now)
-        for i in range(2)
-    ]
-    alert = Alert(organization_id=1, started_at=now - timedelta(seconds=1), last_seen_at=now)
-    detection_session.add_all([alert, *sequences])
-    await detection_session.flush()
-    detection_session.add_all([AlertSequence(alert_id=alert.id, sequence_id=seq.id) for seq in sequences])
-    detection = Detection(camera_id=1, pose_id=1, sequence_id=sequences[0].id, bucket_key="before", bbox="[]")
-    detection_session.add(detection)
-    await detection_session.commit()
-    alert_id, sequence_id, detection_id = alert.id, sequences[0].id, detection.id
-    original_pairs = {(alert_id, seq.id) for seq in sequences}
-
-    fail = AsyncMock(side_effect=RuntimeError("late workflow failure"))
-
-    if operation in {"label", "unmatch"}:
-        monkeypatch.setattr(AlertCRUD, "create", fail)
-    elif operation == "delete-sequence":
-        monkeypatch.setattr(endpoint, "refresh_alert_state", fail)
-    else:
-        monkeypatch.setattr(AlertCRUD, "delete", fail)
-    auth = pytest.get_token(1, ["admin"], 1)
-    if operation == "label":
-        request = async_client.patch(
-            f"/sequences/{sequence_id}/label", headers=auth, json={"is_wildfire": "other_smoke"}
-        )
-    elif operation == "unmatch":
-        request = async_client.post(f"/alerts/{alert_id}/sequences/{sequence_id}/unmatch", headers=auth)
-    elif operation == "delete-sequence":
-        request = async_client.delete(f"/sequences/{sequence_id}", headers=auth)
-    else:
-        request = async_client.delete(f"/alerts/{alert_id}", headers=auth)
-    with pytest.raises(RuntimeError, match="late workflow failure"):
-        await request
-    async with session_factory() as observer:
-        assert await observer.get(Alert, alert_id) is not None
-        persisted = await observer.get(Sequence, sequence_id)
-        assert persisted is not None
-        assert persisted.is_wildfire is None
-        assert (await observer.get(Detection, detection_id)).sequence_id == sequence_id
-        links = (await observer.exec(select(AlertSequence).where(AlertSequence.alert_id == alert_id))).all()
-        assert {(link.alert_id, link.sequence_id) for link in links} == original_pairs

From db1882089cd54c495fd4f9a0b71d15070ca55dc0 Mon Sep 17 00:00:00 2001
From: FG Fernandez <26927750+frgfm@users.noreply.github.com>
Date: Fri, 23 Oct 2026 15:07:58 +0000
Subject: [PATCH 3/3] fix: serialize ingestion per pose within atomic
 transactions

---
 src/app/api/api_v1/endpoints/detections.py    | 13 ++-
 src/app/crud/crud_sequence.py                 | 19 ++--
 src/tests/endpoints/test_detections.py        | 97 ++++++++++++++++++-
 .../endpoints/test_temporal_validation.py     |  8 ++
 4 files changed, 124 insertions(+), 13 deletions(-)

diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py
index ffc432a2..363fda38 100644
--- a/src/app/api/api_v1/endpoints/detections.py
+++ b/src/app/api/api_v1/endpoints/detections.py
@@ -158,7 +158,7 @@ async def _attach_continuity_detections(
         created.append(
             await _create_continuity_detection(detections, camera_id, pose_id, bucket_key, seq.id, recorded_at)
         )
-        await sequences.enqueue_validation(seq.id, commit=False)
+        await sequences.enqueue_validation(seq.id)
     return created
 
 
@@ -580,6 +580,11 @@ async def create_detection(
         if not continuity_sequences:
             return Response(status_code=status.HTTP_204_NO_CONTENT)
         bucket_key = await upload_file(file, token_payload.organization_id, token_payload.sub)
+        await sequences.lock_pose(pose_id)
+        # Uploading and waiting may change which sequences still need continuity.
+        continuity_sequences = await _get_continuity_sequences(sequences, token_payload.sub, pose_id)
+        if not continuity_sequences:
+            return Response(status_code=status.HTTP_204_NO_CONTENT)
         continuity_dets = await _attach_continuity_detections(
             detections, sequences, continuity_sequences, token_payload.sub, pose_id, bucket_key, effective_recorded_at
         )
@@ -596,6 +601,8 @@ async def create_detection(
             crop, token_payload.organization_id, token_payload.sub, key_prefix=f"crop_{idx}_"
         )
 
+    # Keep media I/O outside the lock; serialize matching and every write for this pose.
+    await sequences.lock_pose(pose_id)
     created: List[Detection] = []
     camera = cast(Camera, await cameras.get(token_payload.sub, strict=True))
     # sequences touched by this request, to mark due for validation (DB-backed queue).
@@ -648,7 +655,7 @@ async def create_detection(
             # Only the primary bbox tracks the sequence; siblings in others_bboxes are unrelated detections.
             det_max_conf = max_conf_from_bboxes(det.bbox)
             if det_max_conf is not None:
-                await sequences.bump_max_conf(matched_sequence.id, det_max_conf, commit=False)
+                await sequences.bump_max_conf(matched_sequence.id, det_max_conf)
             affected_sequences.add(matched_sequence.id)
         else:
             det_filters: List[tuple[str, Any]] = [
@@ -725,7 +732,7 @@ async def create_detection(
     # claims due sequences from the DB and runs the gated pipeline: triangulation and ALL
     # notification channels (webhooks, Telegram, Slack) fire only once validated.
     for seq_id in affected_sequences:
-        await sequences.enqueue_validation(seq_id, commit=False)
+        await sequences.enqueue_validation(seq_id)
 
     # Detection rows, sequence membership/confidence and validation jobs become visible together.
     await detections.session.commit()
diff --git a/src/app/crud/crud_sequence.py b/src/app/crud/crud_sequence.py
index 1bc480ac..37567ca5 100644
--- a/src/app/crud/crud_sequence.py
+++ b/src/app/crud/crud_sequence.py
@@ -8,7 +8,7 @@
 from datetime import timedelta
 from typing import Any, Union, cast
 
-from sqlalchemy import case, distinct, func, null, or_, select, update
+from sqlalchemy import case, distinct, func, null, or_, select, text, update
 from sqlmodel import select as select_model
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -26,8 +26,13 @@ class SequenceCRUD(BaseCRUD[Sequence, Sequence, Union[SequenceUpdate, SequenceLa
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, Sequence)
 
-    async def bump_max_conf(self, sequence_id: int, candidate: float, *, commit: bool = True) -> None:
-        """Atomically raise sequences.max_conf to candidate if higher (or set if NULL).
+    async def lock_pose(self, pose_id: int) -> None:
+        """Serialize sequence assembly for this pose until the caller commits or rolls back."""
+        # Advisory-lock namespace 7342 is reserved for ingestion (7341 is alert attachment).
+        await self.session.exec(text("SELECT pg_advisory_xact_lock(7342, :pose_id)"), params={"pose_id": pose_id})
+
+    async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
+        """Raise sequences.max_conf if higher (or set if NULL), in the caller's transaction.
 
         Uses a portable CASE expression so it runs on SQLite as well as Postgres.
         """
@@ -38,8 +43,6 @@ async def bump_max_conf(self, sequence_id: int, candidate: float, *, commit: boo
         )
         stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(max_conf=bumped)
         await self.session.exec(stmt)
-        if commit:
-            await self.session.commit()
 
     async def set_temporal_score(
         self,
@@ -84,8 +87,8 @@ async def claim_validation(self, sequence_id: int, validation_status: Union[str,
         await self.session.commit()
         return bool(getattr(result, "rowcount", 0))
 
-    async def enqueue_validation(self, sequence_id: int, *, commit: bool = True) -> None:
-        """Mark the sequence as due for temporal validation (the DB-backed queue).
+    async def enqueue_validation(self, sequence_id: int) -> None:
+        """Mark the sequence due for validation (the DB-backed queue), in the caller's transaction.
 
         Idempotent and FIFO-preserving: ``COALESCE`` keeps the oldest due timestamp, so a
         sequence already queued is NOT re-queued (one entry per sequence, whichever worker
@@ -102,8 +105,6 @@ async def enqueue_validation(self, sequence_id: int, *, commit: bool = True) ->
             .values(validation_due_at=func.coalesce(due_col, utcnow()))
         )
         await self.session.exec(stmt)
-        if commit:
-            await self.session.commit()
 
     async def claim_due_validation(self, lease_seconds: float) -> Union[Sequence, None]:
         """Claim the oldest due sequence for validation, or None when nothing is due.
diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py
index e9bea2d5..e2b4416a 100644
--- a/src/tests/endpoints/test_detections.py
+++ b/src/tests/endpoints/test_detections.py
@@ -10,7 +10,7 @@
 import pytest  # type: ignore
 from fastapi import HTTPException, UploadFile
 from httpx import AsyncClient
-from sqlmodel import select
+from sqlmodel import select, text
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.api.api_v1.endpoints import detections as detections_api
@@ -2271,3 +2271,98 @@ async def fail_enqueue(self, sequence_id, **kwargs):
     async with session_factory() as observer:
         assert (await observer.exec(select(Detection.id))).all() == original_ids
         assert (await observer.get(Sequence, 1)).model_dump() == before
+        await observer.exec(text("SET LOCAL lock_timeout = '1s'"))
+        await SequenceCRUD(observer).lock_pose(1)  # Rollback released the transaction lock.
+
+
+@pytest.mark.parametrize(
+    "second_boxes",
+    [
+        "[(0.8,0.8,0.9,0.9,0.8),(0.1,0.1,0.2,0.2,0.9)]",
+        "[(0.8,0.8,0.9,0.9,0.8)]",
+        "[]",
+    ],
+    ids=["opposite-order", "matched-and-continuity", "empty-continuity"],
+)
+@pytest.mark.asyncio
+async def test_concurrent_ingestion_serializes_only_the_same_pose(
+    async_client, detection_session, monkeypatch, second_boxes
+):
+    sequence = await detection_session.get(Sequence, 1)
+    sequence.last_seen_at, sequence.is_validated = utcnow(), False
+    other = Sequence(**(sequence.model_dump() | {"id": None}))
+    detection_session.add(other)
+    await detection_session.flush()
+    detection_session.add(
+        Detection(
+            camera_id=1,
+            pose_id=1,
+            sequence_id=other.id,
+            bucket_key="other-seed",
+            bbox="[(0.8,0.8,0.9,0.9,0.8)]",
+        )
+    )
+    await detection_session.commit()
+    monkeypatch.setattr(settings, "SEQUENCE_BBOX_TOLERANCE", 0.0)
+    monkeypatch.setattr(
+        detections_api,
+        "upload_file",
+        AsyncMock(side_effect=["first-frame", "second-frame", "other-pose"]),
+    )
+    paused, resume = asyncio.Event(), asyncio.Event()
+    update_sequence = SequenceCRUD.update
+    holder_pid = 0
+
+    async def pause_after_first_update(self, sequence_id, payload, **kwargs):
+        nonlocal holder_pid
+        result = await update_sequence(self, sequence_id, payload, **kwargs)
+        if sequence_id == 1 and not paused.is_set():
+            holder_pid = (await self.session.exec(text("SELECT pg_backend_pid()"))).scalar_one()
+            paused.set()
+            await resume.wait()
+        return result
+
+    monkeypatch.setattr(SequenceCRUD, "update", pause_after_first_update)
+
+    async def post(boxes, pose_id=1):
+        return await async_client.post(
+            "/detections",
+            headers=pytest.get_token(1, ["camera"], 1),
+            data={"pose_id": pose_id, "bboxes": boxes},
+            files={"file": ("frame.jpg", b"frame", "image/jpeg")},
+        )
+
+    tasks = []
+    try:
+        async with asyncio.timeout(10):
+            tasks.append(asyncio.create_task(post("[(0.1,0.1,0.2,0.2,0.9),(0.8,0.8,0.9,0.9,0.8)]")))
+            await paused.wait()  # First request holds sequence A while the second starts with B.
+            tasks.append(asyncio.create_task(post(second_boxes)))
+            async with session_factory() as observer:
+                # Wait for a real database lock wait, not a scheduler-dependent sleep.
+                while not (
+                    await observer.exec(
+                        text(
+                            "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'advisory' "
+                            "AND NOT granted AND :holder = ANY(pg_blocking_pids(pid)))"
+                        ),
+                        params={"holder": holder_pid},
+                    )
+                ).scalar_one():
+                    pass
+                assert not tasks[1].done()
+                assert (await post("[(0.1,0.1,0.2,0.2,0.9)]", pose_id=2)).status_code == 201
+            resume.set()
+            assert [r.status_code for r in await asyncio.gather(*tasks)] == [201, 201]
+    finally:
+        resume.set()
+        for task in tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+    async with session_factory() as observer:
+        for key in ("first-frame", "second-frame"):
+            rows = (await observer.exec(select(Detection).where(Detection.bucket_key == key))).all()
+            assert len(rows) == 2
+            assert {row.sequence_id for row in rows} == {1, other.id}
diff --git a/src/tests/endpoints/test_temporal_validation.py b/src/tests/endpoints/test_temporal_validation.py
index 0cd99579..a1ed2250 100644
--- a/src/tests/endpoints/test_temporal_validation.py
+++ b/src/tests/endpoints/test_temporal_validation.py
@@ -78,6 +78,7 @@ async def _seed_sequence(
 
 async def _enqueue(session: AsyncSession, sequence_id: int) -> None:
     await SequenceCRUD(session).enqueue_validation(sequence_id)
+    await session.commit()
 
 
 async def _backdate_due(seconds: float = 60.0) -> None:
@@ -159,12 +160,14 @@ async def test_enqueue_is_idempotent_and_fifo_preserving(detection_session: Asyn
     seq = await _seed_sequence(detection_session, 5)
     crud = SequenceCRUD(detection_session)
     await crud.enqueue_validation(cast(int, seq.id))
+    await detection_session.commit()
     await detection_session.refresh(seq)
     first_due = seq.validation_due_at
     assert first_due is not None
 
     await asyncio.sleep(0.01)
     await crud.enqueue_validation(cast(int, seq.id))  # second detection while queued
+    await detection_session.commit()
     await detection_session.refresh(seq)
     assert seq.validation_due_at == first_due  # one entry per sequence, oldest due kept
 
@@ -175,12 +178,14 @@ async def test_enqueue_skips_validated_and_window_exhausted(detection_session: A
     crud = SequenceCRUD(detection_session)
     await crud.claim_validation(cast(int, seq.id))
     await crud.enqueue_validation(cast(int, seq.id))
+    await detection_session.commit()
     await detection_session.refresh(seq)
     assert seq.validation_due_at is None  # validated: nothing to do
 
     seq2 = await _seed_sequence(detection_session, 5)
     await crud.finish_validation_job(cast(int, seq2.id), validation_status=WINDOW_EXHAUSTED)
     await crud.enqueue_validation(cast(int, seq2.id))
+    await detection_session.commit()
     await detection_session.refresh(seq2)
     assert seq2.validation_due_at is None  # terminal: never resurrected
 
@@ -377,6 +382,7 @@ async def test_process_window_exhausted_when_scored_and_past_max_frames(detectio
 
     # Terminal: a later fail-open or enqueue can't resurrect it.
     await crud.enqueue_validation(cast(int, seq.id))
+    await detection_session.commit()
     await detection_session.refresh(seq)
     assert seq.validation_due_at is None
 
@@ -585,6 +591,7 @@ async def test_poison_job_dead_letters_after_max_attempts(detection_session: Asy
 
     # Terminal: new detections can't resurrect it, the worker sees nothing due.
     await crud.enqueue_validation(cast(int, seq.id))
+    await detection_session.commit()
     await detection_session.refresh(seq)
     assert seq.validation_due_at is None
     assert await _process_one_due() is False
@@ -897,6 +904,7 @@ async def test_first_score_past_window_below_threshold_is_terminal(detection_ses
 
     # Terminal: a later enqueue can't resurrect it.
     await SequenceCRUD(detection_session).enqueue_validation(cast(int, seq.id))
+    await detection_session.commit()
     await detection_session.refresh(seq)
     assert seq.validation_due_at is None