feat(ask): verify typed public semantic claims

backend/app/global_ask_queue.py

@@ -29,10 +29,22 @@
 from fastapi import HTTPException, status
 
 from lineageweave.ask_delivery import build_ask_delivery
+from lineageweave.claim_verification import (
+    CLAIM_NOT_ENOUGH_INFORMATION,
+    VERIFICATION_COMPLETED,
+    VERIFICATION_NO_PUBLIC_CLAIMS,
+    VERIFICATION_SKIPPED,
+    VERIFICATION_UNAVAILABLE,
+    ClaimVerificationClient,
+    ClaimVerificationResult,
+    NullClaimVerificationClient,
+    public_claim_candidates,
+)
 from lineageweave.embedding_client import EmbeddingClient, NullEmbeddingClient
 from lineageweave.http_client import HttpClientError
 from lineageweave.observability import record_server_failure
 from lineageweave.post_chat import (
+    ChatSourceDocument,
     PostChatClient,
     cited_post_evidence,
     cited_post_summaries,
@@ -92,6 +104,7 @@ async def enqueue_global_ask_job(
     *,
     requesting_account_id: str,
     question_text: str,
+    verify_external_requested: bool,
     corporate_entity_ids: frozenset[str],
     process_unit_ids: frozenset[str],
 ) -> str:
@@ -104,11 +117,13 @@ async def enqueue_global_ask_job(
     async with conn.transaction():
         job_id = await conn.fetchval(
             """
-            insert into global_ask_job (requesting_account_id, question_text)
-            values ($1, $2) returning global_ask_job_id
+            insert into global_ask_job
+                (requesting_account_id, question_text, verify_external_requested)
+            values ($1, $2, $3) returning global_ask_job_id
             """,
             requesting_account_id,
             question_text,
+            verify_external_requested,
         )
         await conn.executemany(
             """
@@ -141,6 +156,55 @@ async def enqueue_global_ask_job(
     return str(job_id)
 
 
+def _verification_next_action(status_code: str) -> str | None:
+    """Name the next evidence action without promoting web results to authority."""
+
+    return {
+        VERIFICATION_SKIPPED: None,
+        VERIFICATION_UNAVAILABLE: "Review the cited posts or try the public information check again later.",
+        VERIFICATION_NO_PUBLIC_CLAIMS: "Open the cited posts to review the available evidence.",
+        VERIFICATION_COMPLETED: "Compare the public sources with the cited posts before deciding what to do next.",
+        CLAIM_NOT_ENOUGH_INFORMATION: "Find another reliable source before relying on this claim.",
+    }.get(status_code, "Open the cited posts and review their evidence.")
+
+
+async def _verify_public_claims(
+    sources: list[ChatSourceDocument],
+    cited_post_ids: list[str],
+    *,
+    verify_external: bool,
+    client: ClaimVerificationClient,
+) -> tuple[str, tuple[ClaimVerificationResult, ...]]:
+    """Verify only cited claims explicitly marked safe for public egress."""
+
+    if not verify_external:
+        return VERIFICATION_SKIPPED, ()
+    cited_ids = frozenset(cited_post_ids)
+    claims = tuple(
+        claim
+        for claim in public_claim_candidates(
+            sources, allowed_source_post_ids=cited_ids
+        )
+    )
+    if not claims:
+        return VERIFICATION_NO_PUBLIC_CLAIMS, ()
+    if not client.available:
+        return VERIFICATION_UNAVAILABLE, ()
+    try:
+        results = tuple(
+            await asyncio.gather(
+                *(asyncio.to_thread(client.verify, claim) for claim in claims)
+            )
+        )
+    except (HttpClientError, IndexError, KeyError, OSError, TypeError, ValueError):
+        return VERIFICATION_UNAVAILABLE, ()
+    return VERIFICATION_COMPLETED, tuple(
+        result
+        for result in results
+        if set(result.source_post_ids).issubset(cited_ids)
+    )
+
+
 async def load_job_visibility(
     conn: asyncpg.Connection, job_id: str, account_id: str
 ) -> tuple[set[str], set[str], bool, bool]:
@@ -215,6 +279,8 @@ async def compute_global_ask_answer(
     process_scope_limited: bool,
     chat_client: PostChatClient,
     embedding_client: EmbeddingClient | None = None,
+    verify_external: bool = False,
+    claim_verification_client: ClaimVerificationClient | None = None,
 ) -> dict[str, Any]:
     """Assemble one complete Ask answer payload from authorized evidence.
 
@@ -254,7 +320,14 @@ def can_see(row: asyncpg.Record) -> bool:
             status.HTTP_503_SERVICE_UNAVAILABLE,
             "Ask Agent is unavailable: authorized evidence could not be assembled",
         ) from exc
+    verification_client = claim_verification_client or NullClaimVerificationClient()
     if not sources:
+        verification_status, external_claims = await _verify_public_claims(
+            sources,
+            [],
+            verify_external=verify_external,
+            client=verification_client,
+        )
         delivery = build_ask_delivery("", (), ())
         return {
             "answer_text": "",
@@ -264,6 +337,8 @@ def can_see(row: asyncpg.Record) -> bool:
             "cited_post_evidence": [],
             "lineage_graph": {"nodes": [], "edges": [], "truncated": False},
             "cited_post_images": [],
+            "external_verification_status": verification_status,
+            "external_claims": [claim.to_payload() for claim in external_claims],
             "next_action": "No authorized source posts are available for this question.",
             "delivery": delivery,
         }
@@ -300,6 +375,12 @@ def can_see(row: asyncpg.Record) -> bool:
             "Ask Agent is unavailable: contextual-orchestrator could not complete the answer",
         ) from exc
     cited_ids = list(answer.cited_post_ids)
+    verification_status, external_claims = await _verify_public_claims(
+        sources,
+        cited_ids,
+        verify_external=verify_external,
+        client=verification_client,
+    )
     async with pool.acquire() as conn:
         lineage_graph = await lineage_graphs_for_posts(conn, can_see, cited_ids)
         images = await cited_post_images(conn, cited_ids)
@@ -314,6 +395,11 @@ def can_see(row: asyncpg.Record) -> bool:
         "source_post_ids": [source.post_id for source in sources],
         "lineage_graph": lineage_graph,
         "delivery": build_ask_delivery(answer.answer_text, cited_posts, cited_evidence),
+        "external_verification_status": verification_status,
+        "external_claims": [claim.to_payload() for claim in external_claims],
+        "next_action": (
+            _verification_next_action(verification_status) if verify_external else None
+        ),
     }
 
 
@@ -350,6 +436,9 @@ async def process_global_ask_job(
     job_id: str,
     chat_factory: Callable[[], PostChatClient],
     embedding_factory: Callable[[], EmbeddingClient] = NullEmbeddingClient,
+    claim_verification_factory: Callable[
+        [], ClaimVerificationClient
+    ] = NullClaimVerificationClient,
 ) -> None:
     """Claim, answer, and settle one Ask job.
 
@@ -363,7 +452,7 @@ async def process_global_ask_job(
             """
             update global_ask_job set job_status_code = $2, updated_at = now()
             where global_ask_job_id = $1 and job_status_code = $3
-            returning requesting_account_id, question_text
+            returning requesting_account_id, question_text, verify_external_requested
             """,
             job_id,
             RUNNING,
@@ -397,6 +486,8 @@ async def process_global_ask_job(
                 process_scope_limited=process_scope_limited,
                 chat_client=chat_client,
                 embedding_client=embedding_factory(),
+                verify_external=bool(row["verify_external_requested"]),
+                claim_verification_client=claim_verification_factory(),
             ),
             timeout=JOB_DEADLINE_SECONDS,
         )
@@ -511,6 +602,7 @@ async def consume_global_ask_stream_once(
     last_id: str,
     chat_factory: Callable[[], PostChatClient],
     embedding_factory: Callable[[], EmbeddingClient] = NullEmbeddingClient,
+    claim_verification_factory: Callable[[], ClaimVerificationClient] = NullClaimVerificationClient,
     limiter: asyncio.Semaphore | None = None,
     tasks: set[asyncio.Task] | None = None,
 ) -> str:
@@ -535,6 +627,7 @@ async def consume_global_ask_stream_once(
                         job_id=job_id,
                         chat_factory=chat_factory,
                         embedding_factory=embedding_factory,
+                        claim_verification_factory=claim_verification_factory,
                     )
                 else:
                     await limiter.acquire()
@@ -544,6 +637,7 @@ async def consume_global_ask_stream_once(
                             job_id=job_id,
                             chat_factory=chat_factory,
                             embedding_factory=embedding_factory,
+                            claim_verification_factory=claim_verification_factory,
                             limiter=limiter,
                         )
                     )
@@ -560,6 +654,7 @@ async def _process_and_release(
     job_id: str,
     chat_factory: Callable[[], PostChatClient],
     embedding_factory: Callable[[], EmbeddingClient],
+    claim_verification_factory: Callable[[], ClaimVerificationClient],
     limiter: asyncio.Semaphore,
 ) -> None:
     """Run one dispatched job and free its concurrency slot afterwards."""
@@ -569,6 +664,7 @@ async def _process_and_release(
             job_id=job_id,
             chat_factory=chat_factory,
             embedding_factory=embedding_factory,
+            claim_verification_factory=claim_verification_factory,
         )
     finally:
         limiter.release()
@@ -590,6 +686,7 @@ async def run_global_ask_worker(
     *,
     chat_factory: Callable[[], PostChatClient],
     embedding_factory: Callable[[], EmbeddingClient] = NullEmbeddingClient,
+    claim_verification_factory: Callable[[], ClaimVerificationClient] = NullClaimVerificationClient,
 ) -> None:
     """Run the at-least-once Ask consumer with periodic queued-row recovery."""
     last_id = await _stream_tail(client)
@@ -609,6 +706,7 @@ async def run_global_ask_worker(
                     last_id=last_id,
                     chat_factory=chat_factory,
                     embedding_factory=embedding_factory,
+                    claim_verification_factory=claim_verification_factory,
                     limiter=limiter,
                     tasks=tasks,
                 )

backend/app/main.py

@@ -34,6 +34,11 @@
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 
+from lineageweave.claim_verification import (
+    NullClaimVerificationClient,
+    SearxngOrchestratedClaimVerificationClient,
+)
+
 from backend.app.activity_stream import (
     create_valkey_client,
     get_valkey,
@@ -295,6 +300,7 @@ async def lifespan(app: FastAPI):
                     timeout=load_settings().orchestrator_answer_timeout_seconds
                 ),
                 embedding_factory=_embedding_client,
+                claim_verification_factory=lambda: _claim_verification_client(),
             )
         )
         app.state.global_ask_worker = global_ask_worker
@@ -371,6 +377,23 @@ def _relation_verification_client():
     return SearxngRelationVerificationClient(base_url=settings.searxng_base_url)
 
 
+def _claim_verification_client():
+    """Return the public-evidence verifier, or its unavailable null channel."""
+
+    settings = load_settings()
+    if not (
+        settings.searxng_base_url
+        and settings.orchestrator_base_url
+        and settings.orchestrator_api_key
+    ):
+        return NullClaimVerificationClient()
+    return SearxngOrchestratedClaimVerificationClient(
+        settings.searxng_base_url,
+        settings.orchestrator_base_url,
+        settings.orchestrator_api_key,
+    )
+
+
 def _organization_name_resolution_client():
     """Live orchestrator client when configured; otherwise the unavailable null."""
     settings = load_settings()
@@ -2960,6 +2983,7 @@ class GlobalAskRequest(BaseModel):
     """JSON body for the buyer's source-grounded Global Ask Agent."""
 
     question: str
+    verify_external: bool = False
 
 
 @app.get("/api/posts/{post_id}/chat")
@@ -3117,6 +3141,7 @@ async def ask_agent(
             valkey,
             requesting_account_id=account.user_account_id,
             question_text=question,
+            verify_external_requested=request.verify_external,
             corporate_entity_ids=account.corporate_entity_ids,
             process_unit_ids=account.process_unit_ids,
         )