fix: roll back unauthorized Global Ask turns (v2.20.3)

CHANGELOG.d/2.20.3-global-ask-atomic-rollback.md

@@ -0,0 +1,4 @@
+### Fixed
+
+- Roll back a Global Ask turn and its citations when final authorization fails,
+  so a rejected answer cannot poison the existing session.

backend/app/main.py

@@ -179,6 +179,7 @@
     load_global_ask_context,
     persist_global_ask_summary,
     persist_global_ask_turn,
+    persist_global_ask_turn_in_transaction,
     persist_post_chat,
 )
 from backend.app.post_summary_ingestion import (
@@ -2787,25 +2788,26 @@ async def chat_about_post(
         ) from exc
     cited_ids = list(answer.cited_post_ids)
     async with pool.acquire() as conn:
-        await persist_post_chat(
-            conn,
-            post_id,
-            question,
-            answer.answer_text,
-            cited_ids,
-            knowledge_cutoff=knowledge_cutoff,
-        )
-        answer_evidence = await read_authorized_ask_evidence(
-            conn,
-            cited_post_ids=cited_ids,
-            corporate_entity_ids=account.corporate_entity_ids,
-            knowledge_cutoff=knowledge_cutoff,
-        )
-    if not answer_evidence.all_citations_visible:
-        raise HTTPException(
-            status.HTTP_503_SERVICE_UNAVAILABLE,
-            "Post chat evidence changed before the answer could be returned",
-        )
+        async with conn.transaction():
+            await persist_post_chat(
+                conn,
+                post_id,
+                question,
+                answer.answer_text,
+                cited_ids,
+                knowledge_cutoff=knowledge_cutoff,
+            )
+            answer_evidence = await read_authorized_ask_evidence(
+                conn,
+                cited_post_ids=cited_ids,
+                corporate_entity_ids=account.corporate_entity_ids,
+                knowledge_cutoff=knowledge_cutoff,
+            )
+            if not answer_evidence.all_citations_visible:
+                raise HTTPException(
+                    status.HTTP_503_SERVICE_UNAVAILABLE,
+                    "Post chat evidence changed before the answer could be returned",
+                )
     await publish_activity_event(
         valkey,
         post_id,
@@ -2946,24 +2948,25 @@ async def ask_agent(
         ) from exc
     cited_ids = list(answer.cited_post_ids)
     async with pool.acquire() as conn:
-        await persist_global_ask_turn(
-            conn,
-            conversation.session_id,
-            question,
-            answer.answer_text,
-            cited_ids,
-        )
-        answer_evidence = await read_authorized_ask_evidence(
-            conn,
-            cited_post_ids=cited_ids,
-            corporate_entity_ids=account.corporate_entity_ids,
-            knowledge_cutoff=knowledge_cutoff,
-        )
-    if not answer_evidence.all_citations_visible:
-        raise HTTPException(
-            status.HTTP_503_SERVICE_UNAVAILABLE,
-            "Global Ask evidence changed before the answer could be returned",
-        )
+        async with conn.transaction():
+            await persist_global_ask_turn_in_transaction(
+                conn,
+                conversation.session_id,
+                question,
+                answer.answer_text,
+                cited_ids,
+            )
+            answer_evidence = await read_authorized_ask_evidence(
+                conn,
+                cited_post_ids=cited_ids,
+                corporate_entity_ids=account.corporate_entity_ids,
+                knowledge_cutoff=knowledge_cutoff,
+            )
+            if not answer_evidence.all_citations_visible:
+                raise HTTPException(
+                    status.HTTP_503_SERVICE_UNAVAILABLE,
+                    "Global Ask evidence changed before the answer could be returned",
+                )
     await publish_operation_event(
         valkey,
         account.user_account_id,

backend/app/post_chat_ingestion.py

@@ -193,46 +193,80 @@ async def persist_global_ask_summary(
     )
 
 
-async def persist_global_ask_turn(
+async def _persist_global_ask_turn_rows(
     conn: asyncpg.Connection,
     session_id: str,
     question: str,
     answer: str,
     cited_post_ids: Iterable[str],
 ) -> int:
-    """Append one serialized turn and its normalized citation references."""
+    """Append one turn while the caller owns the surrounding transaction."""
     citations = list(dict.fromkeys(str(post_id) for post_id in cited_post_ids))
-    async with conn.transaction():
-        await conn.fetchrow(
-            "select global_ask_session_id from global_ask_session where global_ask_session_id = $1 for update",
+    await conn.fetchrow(
+        "select global_ask_session_id from global_ask_session where global_ask_session_id = $1 for update",
+        session_id,
+    )
+    ordinal = int(
+        await conn.fetchval(
+            "select coalesce(max(turn_ordinal), 0) + 1 from global_ask_turn where global_ask_session_id = $1",
             session_id,
         )
-        ordinal = int(
-            await conn.fetchval(
-                "select coalesce(max(turn_ordinal), 0) + 1 from global_ask_turn where global_ask_session_id = $1",
-                session_id,
-            )
-        )
+    )
+    await conn.execute(
+        "insert into global_ask_turn (global_ask_session_id, turn_ordinal, question_text, answer_text) values ($1, $2, $3, $4)",
+        session_id,
+        ordinal,
+        question,
+        answer,
+    )
+    for citation_ordinal, post_id in enumerate(citations):
         await conn.execute(
-            "insert into global_ask_turn (global_ask_session_id, turn_ordinal, question_text, answer_text) values ($1, $2, $3, $4)",
+            "insert into global_ask_turn_citation (global_ask_session_id, turn_ordinal, citation_ordinal, cited_post_id) values ($1, $2, $3, $4)",
             session_id,
             ordinal,
-            question,
-            answer,
+            citation_ordinal,
+            post_id,
         )
-        for citation_ordinal, post_id in enumerate(citations):
-            await conn.execute(
-                "insert into global_ask_turn_citation (global_ask_session_id, turn_ordinal, citation_ordinal, cited_post_id) values ($1, $2, $3, $4)",
-                session_id,
-                ordinal,
-                citation_ordinal,
-                post_id,
-            )
-        await conn.execute(
-            "update global_ask_session set updated_at = now() where global_ask_session_id = $1",
+    await conn.execute(
+        "update global_ask_session set updated_at = now() where global_ask_session_id = $1",
+        session_id,
+    )
+    return ordinal
+
+
+async def persist_global_ask_turn(
+    conn: asyncpg.Connection,
+    session_id: str,
+    question: str,
+    answer: str,
+    cited_post_ids: Iterable[str],
+) -> int:
+    """Append one serialized turn and its normalized citation references."""
+    async with conn.transaction():
+        return await _persist_global_ask_turn_rows(
+            conn,
             session_id,
+            question,
+            answer,
+            cited_post_ids,
         )
-    return ordinal
+
+
+async def persist_global_ask_turn_in_transaction(
+    conn: asyncpg.Connection,
+    session_id: str,
+    question: str,
+    answer: str,
+    cited_post_ids: Iterable[str],
+) -> int:
+    """Append one turn while the API owns the reauthorization transaction."""
+    return await _persist_global_ask_turn_rows(
+        conn,
+        session_id,
+        question,
+        answer,
+        cited_post_ids,
+    )
 
 
 async def _normalize_post_body_text(