fix(sessions): compute turn_index per turn and 409 duplicate appends

Author: mmabrouk

api/oss/src/dbs/postgres/sessions/turns/dao.py

@@ -2,6 +2,7 @@
 from uuid import UUID
 
 from sqlalchemy import delete as sa_delete, select
+from sqlalchemy.exc import IntegrityError
 
 from oss.src.core.sessions.turns.dtos import (
     HarnessKind,
@@ -11,6 +12,7 @@
 )
 from oss.src.core.sessions.turns.interfaces import SessionTurnsDAOInterface
 from oss.src.core.shared.dtos import Windowing
+from oss.src.core.shared.exceptions import EntityCreationConflict
 from oss.src.dbs.postgres.sessions.turns.dbes import SessionTurnDBE
 from oss.src.dbs.postgres.sessions.turns.mappings import (
     map_turn_dbe_to_dto,
@@ -44,9 +46,26 @@ async def append(
             turn=turn,
         )
         async with self.engine.session() as session:
-            session.add(dbe)
-            await session.commit()
-            await session.refresh(dbe)
+            try:
+                session.add(dbe)
+                await session.commit()
+                await session.refresh(dbe)
+            except IntegrityError as e:
+                await session.rollback()
+                error_str = str(e.orig) if e.orig else str(e)
+                if "ix_session_turns_project_id_session_id_turn_index" in error_str:
+                    raise EntityCreationConflict(
+                        entity="Session turn",
+                        message=(
+                            f"Session turn {turn.turn_index} already exists for "
+                            f"session {turn.session_id}."
+                        ),
+                        conflict={
+                            "session_id": turn.session_id,
+                            "turn_index": turn.turn_index,
+                        },
+                    ) from e
+                raise
         return map_turn_dbe_to_dto(turn_dbe=dbe)
 
     async def fetch_turn(

api/oss/tests/pytest/unit/sessions/test_turns_dao_conflict.py

@@ -0,0 +1,77 @@
+"""A duplicate session turn maps to HTTP 409 instead of an uncaught database error."""
+
+from contextlib import asynccontextmanager
+from uuid import uuid4
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from oss.src.core.sessions.turns.dtos import HarnessKind, SessionTurnCreate
+from oss.src.dbs.postgres.sessions.turns.dao import SessionTurnsDAO
+from oss.src.utils.exceptions import ConflictException, intercept_exceptions
+
+
+class _FakeSession:
+    def __init__(self):
+        self.rollback_count = 0
+
+    def add(self, _dbe):
+        pass
+
+    async def commit(self):
+        raise IntegrityError(
+            "INSERT INTO session_turns ...",
+            {},
+            Exception(
+                "duplicate key value violates unique constraint "
+                '"ix_session_turns_project_id_session_id_turn_index"'
+            ),
+        )
+
+    async def refresh(self, _dbe):
+        raise AssertionError("a failed insert must not refresh")
+
+    async def rollback(self):
+        self.rollback_count += 1
+
+
+class _FakeEngine:
+    def __init__(self):
+        self.session_handle = _FakeSession()
+
+    @asynccontextmanager
+    async def session(self):
+        yield self.session_handle
+
+
+@pytest.mark.anyio
+async def test_duplicate_turn_returns_409_and_rolls_back(anyio_backend):
+    assert anyio_backend == "asyncio"
+    engine = _FakeEngine()
+    dao = SessionTurnsDAO(engine=engine)
+    session_id = "sess-duplicate-1"
+    turn = SessionTurnCreate(
+        session_id=session_id,
+        stream_id=uuid4(),
+        turn_index=3,
+        harness_kind=HarnessKind.PI,
+    )
+
+    @intercept_exceptions(verbose=False)
+    async def append_turn():
+        return await dao.append(project_id=uuid4(), user_id=None, turn=turn)
+
+    with pytest.raises(ConflictException) as exc_info:
+        await append_turn()
+
+    assert engine.session_handle.rollback_count == 1
+    assert exc_info.value.status_code == 409
+    assert exc_info.value.detail == {
+        "message": f"Session turn 3 already exists for session {session_id}.",
+        "conflict": {"session_id": session_id, "turn_index": 3},
+    }
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"