diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index 2f230d226..f334a0c08 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/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,7 +139,7 @@ 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, @@ -145,21 +147,29 @@ async def verify_post_relations( 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 diff --git a/migrations/0165_global_ask_job.sql b/migrations/0165_global_ask_job.sql index 37fa4b568..95d28372a 100644 --- a/migrations/0165_global_ask_job.sql +++ b/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), diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 79eab4f47..9849d9326 100644 --- a/tests/test_relation_verification_internal.py +++ b/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" diff --git a/tests/test_schema.py b/tests/test_schema.py index 296690c41..a2960f873 100644 --- a/tests/test_schema.py +++ b/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", diff --git a/tests/test_server_diagnostics.py b/tests/test_server_diagnostics.py index a81aea95e..499858121 100644 --- a/tests/test_server_diagnostics.py +++ b/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