diff --git a/CHANGELOG.d/2.20.3-global-ask-atomic-rollback.md b/CHANGELOG.d/2.20.3-global-ask-atomic-rollback.md new file mode 100644 index 000000000..91d36685c --- /dev/null +++ b/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. diff --git a/backend/app/main.py b/backend/app/main.py index 06f6d2457..15f6aa22f 100644 --- a/backend/app/main.py +++ b/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, diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 8daf01b1f..f728ea89a 100644 --- a/backend/app/post_chat_ingestion.py +++ b/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( diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 12d60ca98..c1a00d7af 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -3474,6 +3474,197 @@ def answer(self, question: str, sources) -> object: assert "raw-global-provider-secret" not in response.text +def test_global_ask_failed_reauthorization_rolls_back_and_session_continues( + client, demo_analyst_token, seeded_db, monkeypatch +) -> None: + """A citation race rolls back only the new turn and preserves the session.""" + from lineageweave.post_chat import ChatAnswer + from backend.app.ask_project_history import AskEvidenceProjection + + class _FakeAskClient: + available = True + + def answer(self, question: str, sources, *, conversation_context: str = "") -> ChatAnswer: + del question, sources, conversation_context + return ChatAnswer( + answer_text="Synthetic answer that must be rolled back", + cited_post_ids=(seeded_db["public_post_id"],), + ) + + async def hidden_after_persist(*_args, **_kwargs) -> AskEvidenceProjection: + return AskEvidenceProjection( + all_citations_visible=False, + cited_posts=(), + project_histories=(), + project_histories_truncated=False, + knowledge_cutoff="2026-08-21T00:00:00Z", + ) + + monkeypatch.setattr("backend.app.main._post_chat_client", lambda: _FakeAskClient()) + monkeypatch.setattr("backend.app.main.read_authorized_ask_evidence", hidden_after_persist) + + session_id = str(uuid.uuid4()) + admin_conn = psycopg2.connect(seeded_db["dsn"]) + admin_conn.autocommit = True + try: + with admin_conn.cursor() as cur: + cur.execute( + """ + insert into global_ask_session (global_ask_session_id, user_account_id) + select %s, user_account_id + from user_account + where email_address = 'test.analyst@example.test' + """, + (session_id,), + ) + cur.execute( + "insert into global_ask_turn " + "(global_ask_session_id, turn_ordinal, question_text, answer_text) " + "values (%s, 1, 'Prior question', 'Prior answer')", + (session_id,), + ) + finally: + admin_conn.close() + + failed = client.post( + "/api/ask", + json={ + "question": "What happened before the authorization race?", + "session_id": session_id, + }, + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert failed.status_code == 503 + assert "Synthetic answer" not in failed.text + + admin_conn = psycopg2.connect(seeded_db["dsn"]) + admin_conn.autocommit = True + try: + with admin_conn.cursor() as cur: + cur.execute( + "update source_post set source_draft_code = 'race-draft' where post_id = %s", + (seeded_db["public_post_id"],), + ) + finally: + admin_conn.close() + + async def no_authorized_sources(*_args, **_kwargs): + return [] + + monkeypatch.setattr("backend.app.main.gather_global_chat_sources", no_authorized_sources) + follow_up = client.post( + "/api/ask", + json={"question": "Continue after the race", "session_id": session_id}, + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert follow_up.status_code == 200, follow_up.text + assert follow_up.json()["answer_text"] == "" + assert follow_up.json()["cited_post_ids"] == [] + + verify_conn = psycopg2.connect(seeded_db["dsn"]) + verify_conn.autocommit = True + try: + with verify_conn.cursor() as cur: + cur.execute( + "select question_text, answer_text from global_ask_turn " + "where global_ask_session_id = %s order by turn_ordinal", + (session_id,), + ) + assert cur.fetchall() == [ + ("Prior question", "Prior answer"), + ("Continue after the race", ""), + ] + cur.execute( + "select count(*) from global_ask_turn_citation where global_ask_session_id = %s", + (session_id,), + ) + assert cur.fetchone()[0] == 0 + finally: + verify_conn.close() + + +def test_post_chat_failed_reauthorization_rolls_back_and_retry_succeeds( + client, demo_analyst_token, seeded_db, monkeypatch +) -> None: + """A post-chat citation race rolls back its cached answer before retry.""" + from lineageweave.post_chat import ChatAnswer + from backend.app.ask_project_history import AskEvidenceProjection + + class _FakePostChatClient: + available = True + + def __init__(self) -> None: + self.calls = 0 + + def answer(self, question: str, sources) -> ChatAnswer: + del question, sources + self.calls += 1 + if self.calls == 1: + return ChatAnswer( + answer_text="Synthetic answer that must be rolled back", + cited_post_ids=(seeded_db["public_post_id"],), + ) + return ChatAnswer(answer_text="Recovered post-chat answer", cited_post_ids=()) + + chat_client = _FakePostChatClient() + authorization_checks = 0 + + async def reauthorization(*_args, **_kwargs) -> AskEvidenceProjection: + nonlocal authorization_checks + authorization_checks += 1 + return AskEvidenceProjection( + all_citations_visible=authorization_checks > 1, + cited_posts=(), + project_histories=(), + project_histories_truncated=False, + knowledge_cutoff="2026-08-21T00:00:00Z", + ) + + monkeypatch.setattr("backend.app.main._post_chat_client", lambda: chat_client) + monkeypatch.setattr("backend.app.main.read_authorized_ask_evidence", reauthorization) + + post_id = seeded_db["own_private_post_id"] + first_question = "Post chat authorization race" + failed = client.post( + f"/api/posts/{post_id}/chat", + json={"question": first_question}, + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert failed.status_code == 503 + assert "Synthetic answer" not in failed.text + + retry = client.post( + f"/api/posts/{post_id}/chat", + json={"question": "Retry after post chat race"}, + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert retry.status_code == 200, retry.text + assert retry.json()["answer_text"] == "Recovered post-chat answer" + + verify_conn = psycopg2.connect(seeded_db["dsn"]) + verify_conn.autocommit = True + try: + with verify_conn.cursor() as cur: + cur.execute( + "select count(*) from post_chat_result where post_id = %s and question_text = %s", + (post_id, first_question), + ) + assert cur.fetchone()[0] == 0 + cur.execute( + "select question_text, answer_text from post_chat_result " + "where post_id = %s order by computed_at, question_norm", + (post_id,), + ) + assert cur.fetchall() == [("Retry after post chat race", "Recovered post-chat answer")] + cur.execute( + "select count(*) from post_chat_citation where post_id = %s", + (post_id,), + ) + assert cur.fetchone()[0] == 0 + finally: + verify_conn.close() + + def test_keymen_provider_error_does_not_leak_raw_error( client, demo_analyst_token, seeded_db, monkeypatch ) -> None: diff --git a/docs/adr/0129-global-ask-atomic-reauthorization.md b/docs/adr/0129-global-ask-atomic-reauthorization.md new file mode 100644 index 000000000..fb70f35eb --- /dev/null +++ b/docs/adr/0129-global-ask-atomic-reauthorization.md @@ -0,0 +1,48 @@ +# ADR 0129 — Ask turn persistence and final citation reauthorization share one transaction + +- Status: Accepted on this feature branch; protected-main adoption requires the normal stacked PR gates +- Date: 2026-08-21 +- Depends on: ADR 0113 and ADR 0126 + +## Context + +Global Ask and post-scoped chat persist a new answer and its citation rows +before performing the final tenant, publication, and knowledge-cutoff +reauthorization. If a cited post changes between retrieval and that final +check, the API correctly returns a generic 503, but a committed rejected turn +can poison the session or leave an invalid cached answer. + +## Decision + +1. The `/api/ask` cited-answer path acquires one PostgreSQL connection and + opens one outer transaction. +2. The `/api/posts/{post_id}/chat` cited-answer path uses the same transaction + boundary for the result and normalized citation rows. +3. Global Ask allocates its turn ordinal while the session row is locked; both + paths run final citation reauthorization on the same connection, with the + same tenant scope and knowledge cutoff, before the transaction exits. +4. A failed final authorization raises the existing generic 503 inside the + transaction. PostgreSQL then rolls back only the new answer and citations; + previously committed turns and cached exchanges remain unchanged. +5. The no-authorized-source path keeps its existing empty-turn behavior and + does not invent evidence. + +## Consequences + +- A visibility/publication race cannot leave a rejected citation in Global Ask + session history or post-scoped chat storage. +- A follow-up request can reuse the same Global Ask session after a rejected + answer, and a post-scoped question can be retried without a poisoned cache. +- The answer, citation persistence, and final authorization share one + database connection and lock lifetime; the transaction must not include the + provider call or Valkey publication. +- Real PostgreSQL integration coverage is required for both rollback and + session-continuity behavior. + +## References — APA 7th + +PostgreSQL Global Development Group. (2026). *Transactions*. PostgreSQL +documentation. https://www.postgresql.org/docs/current/tutorial-transactions.html + +OWASP Foundation. (2025). *Error handling*. OWASP Application Security +Verification Standard. https://owasp.org/www-project-application-security-verification-standard/ diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index fccd061b2..a58bfa72f 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -347,3 +347,19 @@ runtime note into a shipped/live claim. and evidence navigation do not identify why a VOC occurred. *This document is continuously updated by the hourly automated agent loop.* + +## Global Ask atomic rollback gap (2026-08-21) + +- **Observed gap**: the final citation reauthorization ran after the new + Global Ask turn committed, so a visibility/publication race could leave a + rejected citation in session history and force a later 409 restart. +- **Active remediation**: issue #362 and the stacked v2.20.3 feature branch + move turn persistence and final authorization into one outer PostgreSQL + transaction while keeping tenant ABAC, publication eligibility, knowledge + cutoff, session continuity, and the generic 503 boundary. +- **Closure evidence**: a live PostgreSQL integration test forces the final + authorization to fail, verifies the rejected turn/citations are absent, then + makes the cited source ineligible and proves the same session can complete a + safe no-source follow-up with HTTP 200. Backend API and focused Ask suites + pass on the exact branch head; hosted Checks and independent approval remain + external gates.