perf(web): release provider work and bound landing reads

backend/app/relation_verification_ingestion.py

@@ -30,6 +30,7 @@ class VerifiedRelation:
 @dataclass(frozen=True)
 class _PendingRelation:
     counterparty_entity_name: str
+    relationship_type_code: str
     relationship_label: str
     internal_evidence_post_id: str | None
 
@@ -114,7 +115,8 @@ async def verify_post_relations(
     """
     rows = await conn.fetch(
         """
-        select c.counterparty_entity_name, v.lookup_label as relationship_label
+        select c.counterparty_entity_name, c.relationship_type_code,
+               v.lookup_label as relationship_label
         from post_counterparty_entity c
         join common_lookup_value v on v.lookup_code = c.relationship_type_code
         where c.post_id = $1 and c.verification_status_code = 'verify_pending'
@@ -137,29 +139,37 @@ async def verify_post_relations(
             row["counterparty_entity_name"],
             row["relationship_label"],
         )
-        await conn.execute(
+        update_status = await conn.execute(
             """
             update post_counterparty_entity
             set verification_status_code = $3,
                 verification_evidence_url = $4,
                 verification_evidence_post_id = $5,
                 verification_checked_at = now()
             where post_id = $1 and counterparty_entity_name = $2
+              and verification_status_code = 'verify_pending'
+              and relationship_type_code = $6
+              and ($5::uuid is null or exists (
+                  select 1 from source_post evidence
+                  where evidence.post_id = $5::uuid
+              ))
             """,
             post_id,
             row["counterparty_entity_name"],
             result.status_code,
             result.evidence_url,
             internal_evidence_post_id,
+            row["relationship_type_code"],
         )
-        verified.append(
-            VerifiedRelation(
-                counterparty_entity_name=row["counterparty_entity_name"],
-                verification_status_code=result.status_code,
-                verification_evidence_url=result.evidence_url,
-                verification_evidence_post_id=internal_evidence_post_id,
+        if update_status == "UPDATE 1":
+            verified.append(
+                VerifiedRelation(
+                    counterparty_entity_name=row["counterparty_entity_name"],
+                    verification_status_code=result.status_code,
+                    verification_evidence_url=result.evidence_url,
+                    verification_evidence_post_id=internal_evidence_post_id,
+                )
             )
-        )
     return verified
 
 
@@ -173,7 +183,8 @@ async def verify_post_relations_from_pool(
     async with pool.acquire() as conn:
         rows = await conn.fetch(
             """
-            select c.counterparty_entity_name, v.lookup_label as relationship_label
+            select c.counterparty_entity_name, c.relationship_type_code,
+                   v.lookup_label as relationship_label
             from post_counterparty_entity c
             join common_lookup_value v on v.lookup_code = c.relationship_type_code
             where c.post_id = $1 and c.verification_status_code = 'verify_pending'
@@ -184,6 +195,7 @@ async def verify_post_relations_from_pool(
         pending = [
             _PendingRelation(
                 str(row["counterparty_entity_name"]),
+                str(row["relationship_type_code"]),
                 str(row["relationship_label"]),
                 await _find_internal_evidence_post(
                     conn,
@@ -203,18 +215,13 @@ async def verify_post_relations_from_pool(
             relation.counterparty_entity_name,
             relation.relationship_label,
         )
-        verified.append(
-            VerifiedRelation(
-                relation.counterparty_entity_name,
-                result.status_code,
-                result.evidence_url,
-                relation.internal_evidence_post_id,
-            )
+        completed = VerifiedRelation(
+            relation.counterparty_entity_name,
+            result.status_code,
+            result.evidence_url,
+            relation.internal_evidence_post_id,
         )
-
-    persisted = []
-    async with pool.acquire() as conn, conn.transaction():
-        for relation in verified:
+        async with pool.acquire() as conn:
             update_status = await conn.execute(
                 """
                 update post_counterparty_entity
@@ -224,13 +231,19 @@ async def verify_post_relations_from_pool(
                     verification_checked_at = now()
                 where post_id = $1 and counterparty_entity_name = $2
                   and verification_status_code = 'verify_pending'
+                  and relationship_type_code = $6
+                  and ($5::uuid is null or exists (
+                      select 1 from source_post evidence
+                      where evidence.post_id = $5::uuid
+                  ))
                 """,
                 post_id,
-                relation.counterparty_entity_name,
-                relation.verification_status_code,
-                relation.verification_evidence_url,
-                relation.verification_evidence_post_id,
+                completed.counterparty_entity_name,
+                completed.verification_status_code,
+                completed.verification_evidence_url,
+                completed.verification_evidence_post_id,
+                relation.relationship_type_code,
             )
-            if update_status == "UPDATE 1":
-                persisted.append(relation)
-    return persisted
+        if update_status == "UPDATE 1":
+            verified.append(completed)
+    return verified

migrations/0165_global_ask_job.sql

@@ -7,6 +7,53 @@
 -- Mirrors the durable-row-plus-stream design post_content_job already
 -- uses, so a lost stream entry is recovered from the queued rows.
 
+-- Existing volumes must not silently retain a differently-shaped queue table.
+-- `IF NOT EXISTS` is idempotent only when the existing object is compatible;
+-- fail before any insert path can observe a partial schema.
+do $$
+declare
+    account_index regclass;
+    queued_index regclass;
+begin
+    account_index := to_regclass('public.global_ask_job_account_idx');
+    queued_index := to_regclass('public.global_ask_job_queued_idx');
+    if to_regclass('public.global_ask_job') is not null
+       and exists (
+           select 1
+           from (values
+               ('global_ask_job_id', 'uuid'),
+               ('requesting_account_id', 'uuid'),
+               ('question_text', 'text'),
+               ('job_status_code', 'text'),
+               ('answer_payload', 'jsonb'),
+               ('failure_detail', 'text'),
+               ('created_at', 'timestamp with time zone'),
+               ('updated_at', 'timestamp with time zone')
+           ) as required(column_name, data_type)
+           where not exists (
+               select 1
+               from information_schema.columns column_info
+               where column_info.table_schema = 'public'
+                 and column_info.table_name = 'global_ask_job'
+                 and column_info.column_name = required.column_name
+                 and column_info.data_type = required.data_type
+           )
+       ) then
+        raise exception 'global_ask_job exists with an incompatible schema';
+    end if;
+    if account_index is not null
+       and pg_get_indexdef(account_index)
+           not ilike '%(requesting_account_id, created_at DESC)%' then
+        raise exception 'global_ask_job_account_idx exists with an incompatible definition';
+    end if;
+    if queued_index is not null
+       and pg_get_indexdef(queued_index)
+           not ilike '%(created_at)%where%job_status_code%' then
+        raise exception 'global_ask_job_queued_idx exists with an incompatible definition';
+    end if;
+end
+$$;
+
 create table if not exists global_ask_job (
     global_ask_job_id uuid primary key default uuid_generate_v4(),
     requesting_account_id uuid not null references user_account (user_account_id),

tests/test_relation_verification_internal.py

@@ -2,6 +2,8 @@
 
 import asyncio
 
+import pytest
+
 from backend.app.relation_verification_ingestion import (
     verify_post_relations,
     verify_post_relations_from_pool,
@@ -18,12 +20,14 @@ def __init__(self, evidence_post_id: str | None, update_status: str = "UPDATE 1"
         self.update_status = update_status
         self.fetchrow_args: tuple[object, ...] | None = None
         self.execute_args: tuple[object, ...] | None = None
+        self.execute_calls: list[tuple[object, ...]] = []
 
     async def fetch(self, query: str, post_id: str):
         assert "verification_status_code = 'verify_pending'" in query
         return [
             {
                 "counterparty_entity_name": "Example Partner",
+                "relationship_type_code": "partner",
                 "relationship_label": "Partner",
             }
         ]
@@ -36,7 +40,9 @@ async def fetchrow(self, query: str, *args: object):
 
     async def execute(self, query: str, *args: object):
         assert "verification_evidence_post_id = $5" in query
+        assert "$5::uuid is null or exists" in query
         self.execute_args = args
+        self.execute_calls.append(args)
         return self.update_status
 
     def transaction(self):
@@ -52,7 +58,7 @@ async def __aexit__(self, exc_type, exc, traceback):
 
 
 class _Acquire:
-    def __init__(self, pool: "_Pool") -> None:
+    def __init__(self, pool: _Pool) -> None:
         self.pool = pool
 
     async def __aenter__(self):
@@ -99,6 +105,7 @@ def test_relation_verification_persists_authorized_internal_evidence() -> None:
         STATUS_CORROBORATED,
         "https://example.test/evidence",
         "internal-post",
+        "partner",
     )
 
 
@@ -109,7 +116,8 @@ def test_relation_verification_keeps_external_result_when_internal_search_misses
 
     assert verified[0].verification_evidence_post_id is None
     assert conn.execute_args is not None
-    assert conn.execute_args[-1] is None
+    assert conn.execute_args[-2] is None
+    assert conn.execute_args[-1] == "partner"
 
 
 def test_pool_connection_is_released_during_external_verification() -> None:
@@ -139,3 +147,45 @@ def test_pool_verification_counts_only_rows_settled_by_this_worker() -> None:
     )
 
     assert verified == []
+
+
+def test_pool_verification_persists_completed_rows_before_provider_failure() -> None:
+    """A later provider failure does not roll back an earlier completed row."""
+
+    class _TwoRelationConnection(_Connection):
+        async def fetch(self, query: str, post_id: str):
+            assert "verification_status_code = 'verify_pending'" in query
+            return [
+                {
+                    "counterparty_entity_name": "Example Partner",
+                    "relationship_type_code": "partner",
+                    "relationship_label": "Partner",
+                },
+                {
+                    "counterparty_entity_name": "Example Supplier",
+                    "relationship_type_code": "supplier",
+                    "relationship_label": "Supplier",
+                },
+            ]
+
+    class _FailingSecondVerifier:
+        def verify(
+            self, organization_name: str, relationship_label: str
+        ) -> RelationVerificationResult:
+            if organization_name == "Example Supplier":
+                raise RuntimeError("synthetic provider failure")
+            return RelationVerificationResult(
+                STATUS_CORROBORATED, "https://example.test/evidence"
+            )
+
+    conn = _TwoRelationConnection(None)
+
+    with pytest.raises(RuntimeError, match="synthetic provider failure"):
+        asyncio.run(
+            verify_post_relations_from_pool(
+                _Pool(conn), _FailingSecondVerifier(), "origin-post"
+            )
+        )
+
+    assert len(conn.execute_calls) == 1
+    assert conn.execute_calls[0][-1] == "partner"

tests/test_schema.py

@@ -154,6 +154,14 @@
     / "migrations"
     / "0182_report_leftover_map_unexplained.sql"
 )
+_GLOBAL_ASK_JOB_MIGRATION = (
+    Path(__file__).resolve().parents[1] / "migrations" / "0165_global_ask_job.sql"
+)
+_GLOBAL_ASK_SCOPE_MIGRATION = (
+    Path(__file__).resolve().parents[1]
+    / "migrations"
+    / "0203_global_ask_authorization_scope.sql"
+)
 
 
 def _postgres_available() -> bool:
@@ -218,6 +226,12 @@ def schema_db():
                 cur.execute(_LEFTOVER_MAP_RECONSTRUCTION_MIGRATION.read_text())
                 cur.execute(_LEFTOVER_MAP_UNEXPLAINED_SHARE_MIGRATION.read_text())
                 cur.execute(_SOURCE_EVENT_TIME_MIGRATION.read_text())
+                cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text())
+                cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text())
+                # Exercise the production replay contract against the same
+                # PostgreSQL objects instead of merely inspecting SQL text.
+                cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text())
+                cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text())
                 # Match ADR 0166's production migration executor instead of
                 # maintaining a fixture-owned SQL parser.
                 subprocess.run(
@@ -288,6 +302,9 @@ def test_migration_applies_cleanly(schema_db) -> None:
         "post_summary_action",
         "post_chat_result",
         "post_chat_citation",
+        "global_ask_job",
+        "global_ask_job_corporate_entity_scope",
+        "global_ask_job_process_unit_scope",
         "occupational_construct_vocabulary",
         "occupational_construct",
         "post_occupational_construct_assertion",

tests/test_server_diagnostics.py

@@ -39,6 +39,16 @@ def answer(self, question: str, sources: object) -> object:
         raise self._exc
 
 
+class _EmbeddingClient:
+    """Deterministic available embedding channel for Ask diagnostics."""
+
+    available = True
+    resolved_model = "synthetic-embedding"
+
+    def embed(self, _text: str) -> list[float]:
+        return [1.0, 0.0]
+
+
 def _call_ask(monkeypatch: pytest.MonkeyPatch, exc: BaseException) -> None:
     async def _sources(*args: object, **kwargs: object) -> list[object]:
         return [SimpleNamespace(post_id="synthetic-post-1")]
@@ -53,6 +63,7 @@ async def _sources(*args: object, **kwargs: object) -> list[object]:
                 process_unit_ids=set(),
                 process_scope_limited=False,
                 chat_client=_FailingClient(exc),
+                embedding_client=_EmbeddingClient(),
             )
         )
     assert raised.value.status_code == 503
@@ -150,6 +161,7 @@ async def _sources(*args: object, **kwargs: object) -> list[object]:
                 process_unit_ids=set(),
                 process_scope_limited=False,
                 chat_client=_FailingClient(RuntimeError("unused")),
+                embedding_client=_EmbeddingClient(),
             )
         )
     assert raised.value.status_code == 503