fix: roll back unauthorized Global Ask turns

CHANGELOG.md

@@ -18,6 +18,10 @@ All notable changes to this project are documented here. Format follows
 
 ### Fixed
 
+- Global Ask now re-authorizes cited posts under row locks inside the same
+  persistence transaction, rolling back a raced turn without poisoning the
+  conversation or exposing an unauthorized answer (Issue #362).
+
 - Tenant settings now refresh their audit timestamp on every successful update,
   keep the year field visibly blank while it is edited, and disable no-op
   whitespace-only saves.
@@ -46,9 +50,9 @@ All notable changes to this project are documented here. Format follows
 - All OpenAI-compatible chat-completion consumers now validate the shared
   response envelope before parsing it, preventing malformed provider bodies
   from escaping as raw `KeyError` or response-shape details.
-- Customer Master integration fixtures no longer reference an unshipped
-  Global Ask history migration; Global Ask remains stateless as documented by
-  ADR 0090, while the shipped scope-facet migration is applied directly.
+- Customer Master integration fixtures apply only the migrations needed by
+  their scope; persisted Global Ask history is covered by its own 0105
+  migration and the real API fixture.
 - The static SQL review contract now counts the Customer Master evidence query
   that uses closed schema fragments and bound entity ids.
 

backend/app/global_ask_history.py

@@ -14,6 +14,10 @@ class GlobalAskConversationNotFound(LookupError):
     """The requested conversation is absent or owned by another account."""
 
 
+class GlobalAskEvidenceChanged(RuntimeError):
+    """A cited post became unauthorized before the new turn could commit."""
+
+
 def conversation_title(question: str) -> str:
     """Use the first question as a bounded, readable transcript label."""
     compact = " ".join(question.strip().split())
@@ -250,6 +254,32 @@ async def fetch_conversation(
     }
 
 
+async def _ensure_citations_visible(
+    conn: asyncpg.Connection,
+    conversation_id: UUID,
+    turn_ordinal: int,
+    cited_post_count: int,
+    can_see_post: Callable[[asyncpg.Record], bool],
+) -> None:
+    """Lock and re-authorize new citations before their transaction commits."""
+    rows = await conn.fetch(
+        """
+        select relation.cited_post_id::text as post_id,
+               post.post_title, post.visibility_code, post.corporate_entity_id,
+               post.author_account_id, post.source_detail_state_code
+          from global_ask_turn_citation relation
+          join source_post post on post.post_id = relation.cited_post_id
+         where relation.global_ask_session_id = $1
+           and relation.turn_ordinal = $2
+         for share of post
+        """,
+        conversation_id,
+        turn_ordinal,
+    )
+    if len(rows) != cited_post_count or any(not can_see_post(row) for row in rows):
+        raise GlobalAskEvidenceChanged
+
+
 async def persist_turn(
     conn: asyncpg.Connection,
     user_account_id: str,
@@ -260,6 +290,7 @@ async def persist_turn(
     source_post_ids: Iterable[str],
     cited_post_ids: Iterable[str],
     cited_post_evidence: Iterable[dict[str, Any]],
+    can_see_post: Callable[[asyncpg.Record], bool] | None = None,
 ) -> UUID:
     source_ids = list(dict.fromkeys(str(post_id) for post_id in source_post_ids))
     source_set = set(source_ids)
@@ -348,5 +379,13 @@ async def persist_turn(
             "update global_ask_session set updated_at = now() where global_ask_session_id = $1",
             conversation_id,
         )
+        if can_see_post is not None:
+            await _ensure_citations_visible(
+                conn,
+                conversation_id,
+                ordinal,
+                len(cited_ids),
+                can_see_post,
+            )
     assert conversation_id is not None
     return conversation_id

backend/app/main.py

@@ -137,6 +137,7 @@
 from backend.app.five_w1h_ingestion import load_five_w1h_slots
 from backend.app.global_ask_history import (
     GlobalAskConversationNotFound,
+    GlobalAskEvidenceChanged,
     conversation_exists,
     fetch_conversation,
     list_conversations,
@@ -3254,7 +3255,13 @@ async def ask_agent(
                 response["source_post_ids"],
                 response["cited_post_ids"],
                 response["cited_post_evidence"],
+                can_see_post=lambda row: _can_use_post_for_analysis(account, row),
             )
+        except GlobalAskEvidenceChanged as exc:
+            raise HTTPException(
+                status.HTTP_503_SERVICE_UNAVAILABLE,
+                "Ask Agent is unavailable: authorized evidence changed; retry the question",
+            ) from exc
         except GlobalAskConversationNotFound as exc:
             raise HTTPException(status.HTTP_404_NOT_FOUND, "conversation not found") from exc
     response["conversation_id"] = str(persisted_conversation_id)

backend/tests/test_api.py

@@ -127,6 +127,11 @@
     / "migrations"
     / "0104_two_word_database_identifiers.sql"
 )
+_GLOBAL_ASK_MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "0105_global_ask_conversation_history.sql"
+)
 _TENANT_IDENTITY_METADATA_MIGRATION = (
     Path(__file__).resolve().parents[2]
     / "migrations"
@@ -288,6 +293,7 @@ def seeded_db(demo_analyst_token):
             cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text())
             cur.execute(_TENANT_SETTINGS_MIGRATION.read_text())
             cur.execute(_IDENTIFIER_MIGRATION.read_text())
+            cur.execute(_GLOBAL_ASK_MIGRATION.read_text())
             cur.execute(_TENANT_IDENTITY_METADATA_MIGRATION.read_text())
             cur.execute(_AFFILIATION_SCOPE_FACET_MIGRATION.read_text())
             cur.execute(_EVENT_CLUE_MIGRATION.read_text())
@@ -3960,6 +3966,104 @@ def answer(self, question: str, sources) -> object:
     assert "raw-global-provider-secret" not in response.text
 
 
+def test_global_ask_rejects_raced_citation_without_poisoning_session(
+    client, demo_analyst_token, seeded_db, monkeypatch
+) -> None:
+    """A citation that becomes W is rolled back, while the session remains usable."""
+    from lineageweave.post_chat import ChatAnswer
+
+    class _RaceChatClient:
+        available = True
+        calls = 0
+        raced_post_id: str | None = None
+
+        def answer(self, question: str, sources) -> ChatAnswer:
+            self.calls += 1
+            if self.calls == 2:
+                cited_post_id = sources[0].post_id
+                self.raced_post_id = cited_post_id
+                race_conn = psycopg2.connect(seeded_db["dsn"])
+                race_conn.autocommit = True
+                try:
+                    with race_conn.cursor() as cur:
+                        cur.execute(
+                            "update source_post set source_detail_state_code = 'W' where post_id = %s",
+                            (cited_post_id,),
+                        )
+                finally:
+                    race_conn.close()
+                return ChatAnswer("answer must be rolled back", (cited_post_id,))
+            return ChatAnswer(f"safe answer {self.calls}")
+
+    fake_client = _RaceChatClient()
+    monkeypatch.setattr("backend.app.main._post_chat_client", lambda: fake_client)
+    monkeypatch.setattr(
+        "backend.app.main.cited_post_evidence",
+        lambda sources, cited_ids: [
+            {"post_id": cited_ids[0], "facts": [{"kind": "source_field", "text": "synthetic evidence"}]}
+        ]
+        if cited_ids
+        else [],
+    )
+    headers = {"Authorization": f"Bearer {demo_analyst_token}"}
+
+    first = client.post("/api/ask", json={"question": "Public post"}, headers=headers)
+    assert first.status_code == 200, first.text
+    conversation_id = first.json()["conversation_id"]
+
+    raced = client.post(
+        "/api/ask",
+        json={"question": "Public post", "conversation_id": conversation_id},
+        headers=headers,
+    )
+    assert raced.status_code == 503
+    assert "answer must be rolled back" not in raced.text
+    assert "retry" in raced.text
+
+    with psycopg2.connect(seeded_db["dsn"]) as check_conn, check_conn.cursor() as cur:
+        cur.execute(
+            "select count(*) from global_ask_turn where global_ask_session_id = %s",
+            (conversation_id,),
+        )
+        assert cur.fetchone()[0] == 1
+        cur.execute(
+            "select count(*) from global_ask_turn_citation where global_ask_session_id = %s",
+            (conversation_id,),
+        )
+        assert cur.fetchone()[0] == 0
+        cur.execute(
+            "select count(*) from global_ask_turn_evidence where global_ask_session_id = %s",
+            (conversation_id,),
+        )
+        assert cur.fetchone()[0] == 0
+
+    restore_conn = psycopg2.connect(seeded_db["dsn"])
+    restore_conn.autocommit = True
+    try:
+        with restore_conn.cursor() as cur:
+            cur.execute(
+                "update source_post set source_detail_state_code = null where post_id = %s",
+                (fake_client.raced_post_id,),
+            )
+    finally:
+        restore_conn.close()
+
+    follow_up = client.post(
+        "/api/ask",
+        json={"question": "Public post", "conversation_id": conversation_id},
+        headers=headers,
+    )
+    assert follow_up.status_code == 200, follow_up.text
+    assert follow_up.json()["conversation_id"] == conversation_id
+
+    with psycopg2.connect(seeded_db["dsn"]) as check_conn, check_conn.cursor() as cur:
+        cur.execute(
+            "select count(*) from global_ask_turn where global_ask_session_id = %s",
+            (conversation_id,),
+        )
+        assert cur.fetchone()[0] == 2
+
+
 def test_keymen_provider_error_does_not_leak_raw_error(
     client, demo_analyst_token, seeded_db, monkeypatch
 ) -> None:

docs/adr/0126-global-ask-conversation-history.md

@@ -27,6 +27,13 @@ path owns only the requesting account's conversations and re-applies the
 current post visibility rule before returning source titles, citations, or
 evidence.
 
+Before the persistence transaction commits, cited `source_post` rows are
+locked with `FOR SHARE` and the same analysis-visibility predicate is applied
+again. If a citation became unauthorized or disappeared, the transaction
+raises a stable retryable failure and rolls back the new session, turn,
+citations, and evidence together. This closes the race between evidence
+retrieval and transcript persistence without exposing the discarded answer.
+
 This is transcript persistence, not a new long-context prompt contract. The
 orchestrator continues to receive the current question and its bounded,
 authorized evidence set. Conversation summarization or cross-turn reasoning
@@ -38,5 +45,7 @@ requires a separate ADR and upstream orchestrator contract.
 * A user cannot read another account's conversation by changing a UUID.
 * Revoked post visibility removes that post's source/citation projection from
   history; the stored answer remains account-owned transcript data.
+* A visibility change during a turn cannot leave a partially persisted or
+  unusable conversation; the reader retries after the source boundary settles.
 * The UI can select an existing conversation or start a new one without
   changing the existing `/api/ask` evidence contract.