From 94ebb2c99d618158a86eb6971a3beb1c09acaa55 Mon Sep 17 00:00:00 2001 From: Alexander Gardiner Date: Sun, 12 Jul 2026 00:07:19 -0700 Subject: [PATCH] feat: end-to-end WhatsApp death-certificate verification + GiveLight handoff Wire the full flow: an inbound WhatsApp image is stored in a transient media store on arrival, the death-certificate agent invokes a context-aware verify tool that pulls the image out of band (never through the model), runs the full reliability pipeline (document, authenticity, consistency, score) and hands the case off to GiveLight when it clears the acceptance policy. - session_store: transient per-session media store (save/load latest image) - orchestrator: SessionContext deps + context-aware tool registration; thread deps through Session/run_stream; default Vertex region -> us-central1 - verify tool: run_pipeline + build_handoff_payload + deliver_to_gl (stub), replacing the base64-image-as-tool-arg design that could not work via an LLM - api: extract image/document messages; whatsapp_media.download_media resolves a Meta media ID to bytes (WHATSAPP_TOKEN); route bytes into chat() - chat: store uploaded image, build SessionContext, route image turns on the conversation narrative; the model sees only a text notice, never the bytes - fix: death-cert agent referenced a tool name that did not resolve - tests: media store, context tools, verify tool, api image path, and a TestModel integration test driving the tool + deps + handoff end to end --- agents/poc_deathCertParserAgent.yaml | 12 +- src/api.py | 84 +++++++--- src/chat.py | 84 +++++++++- src/orchestrator/agent.py | 147 ++++++++++++------ src/orchestrator/context.py | 22 +++ src/session.py | 11 +- src/session_store.py | 63 ++++++++ src/whatsapp_media.py | 65 ++++++++ tests/unit/test_api.py | 78 +++++++--- tests/unit/test_chat.py | 92 ++++++++++- tests/unit/test_session_store.py | 112 ++++++++++++- tests/unit/test_verify_flow_integration.py | 73 +++++++++ .../death_certificate_pipeline/test_verify.py | 91 +++++++++++ tools/deathCertificatePipeline.yaml | 22 ++- tools/death_certificate_pipeline/verify.py | 100 ++++++++++++ 15 files changed, 939 insertions(+), 117 deletions(-) create mode 100644 src/orchestrator/context.py create mode 100644 src/whatsapp_media.py create mode 100644 tests/unit/test_verify_flow_integration.py create mode 100644 tests/unit/tools/death_certificate_pipeline/test_verify.py create mode 100644 tools/death_certificate_pipeline/verify.py diff --git a/agents/poc_deathCertParserAgent.yaml b/agents/poc_deathCertParserAgent.yaml index 96d7561..85ab288 100644 --- a/agents/poc_deathCertParserAgent.yaml +++ b/agents/poc_deathCertParserAgent.yaml @@ -16,13 +16,17 @@ system_prompt: | You must track the number of times you have asked the user to provide the death certificate. If you ask for the document 3 times and the user still has not provided one, you must halt all normal conversational responses. From that point forward, reply to EVERY message from the user exclusively with this exact phrase: "I'm sorry, but I cannot proceed without proof". Continue doing this until a document is finally uploaded. PHASE 3: DOCUMENT PROCESSING & TOOL EXECUTION - The moment the user provides an image or PDF, you must immediately notify them by saying: "Your request is being processed." - Following that notification, review the document (which may be in French or Arabic), extract all written text, translate it into English, and immediately pass that data into the `death_certificate_consistency` tool. + The moment the user provides an image or PDF, your VERY FIRST action must be to call the `death_certificate_verification` + tool — call it before writing any reply. Do NOT respond with only "Your request is being processed" and stop; a standalone + acknowledgement without calling the tool is a failure. You do not pass the document in — the tool automatically retrieves + the user's most recent upload, verifies it, and forwards the case to GiveLight when it passes. + Only after the tool returns do you write your reply, and it must reflect the tool's outcome: if it was forwarded, reassure + them their request is now with GiveLight; if it needs review or a clearer image, gently explain the next step. skills: - # TBD + # TBD tools: - - death_certificate_consistency + - death_certificate_verification provider: type: gemini diff --git a/src/api.py b/src/api.py index 5926dcb..9d3605e 100644 --- a/src/api.py +++ b/src/api.py @@ -8,12 +8,14 @@ import logging import os +from dataclasses import dataclass from typing import Any from fastapi import FastAPI, Header, HTTPException, Request from starlette.concurrency import run_in_threadpool from .chat import chat +from .whatsapp_media import download_media logger = logging.getLogger(__name__) @@ -21,9 +23,26 @@ _WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "") +# WhatsApp media message types that carry a downloadable media ID. +_MEDIA_TYPES = ("image", "document") -def _extract_message(payload: dict[str, Any]) -> tuple[str | None, str | None]: - """Return (wa_id, text) for a text message, or (None, None) for anything else.""" + +@dataclass +class InboundMessage: + """A normalized inbound WhatsApp message.""" + + wa_id: str | None = None + text: str | None = None + media_id: str | None = None + mime_type: str | None = None + + +def _extract_message(payload: dict[str, Any]) -> InboundMessage: + """Normalize a WhatsApp webhook payload into an InboundMessage. + + Handles text and media (image/document) messages; returns an empty + InboundMessage for anything else (e.g. status updates). + """ try: entry = (payload.get("entry") or [])[0] value = (entry.get("changes") or [])[0].get("value", {}) @@ -31,20 +50,28 @@ def _extract_message(payload: dict[str, Any]) -> tuple[str | None, str | None]: messages = value.get("messages") or [] if not messages: - return None, None + return InboundMessage() msg = messages[0] - if msg.get("type") != "text": - return None, None - + msg_type = msg.get("type") wa_id = (contacts[0].get("wa_id") if contacts else None) or msg.get("from") - text = (msg.get("text") or {}).get("body", "").strip() - if not text: - return None, None - return wa_id, text + if msg_type == "text": + text = (msg.get("text") or {}).get("body", "").strip() + if not text: + return InboundMessage(wa_id=wa_id) + return InboundMessage(wa_id=wa_id, text=text) + + if msg_type in _MEDIA_TYPES: + media = msg.get(msg_type) or {} + media_id = media.get("id") + if not media_id: + return InboundMessage(wa_id=wa_id) + return InboundMessage(wa_id=wa_id, media_id=media_id, mime_type=media.get("mime_type")) + + return InboundMessage(wa_id=wa_id) except (IndexError, AttributeError, TypeError, KeyError): - return None, None + return InboundMessage() @app.post("/message") @@ -58,15 +85,32 @@ async def message_endpoint( payload = await request.json() logger.info("api.message received object=%s", payload.get("object")) - wa_id, text = _extract_message(payload) - if text is None: - logger.info("api.message skipped — no text payload") - return {"response": None} - - logger.info("api.message routing wa_id=%s chars=%d", wa_id, len(text)) - response = await run_in_threadpool(chat, text=text, session_id=wa_id) - logger.info("api.message done wa_id=%s response_chars=%d", wa_id, len(response)) - return {"response": response} + message = _extract_message(payload) + + if message.text is not None: + logger.info("api.message routing text wa_id=%s chars=%d", message.wa_id, len(message.text)) + response = await run_in_threadpool(chat, text=message.text, session_id=message.wa_id) + logger.info("api.message done wa_id=%s response_chars=%d", message.wa_id, len(response)) + return {"response": response} + + if message.media_id is not None: + logger.info("api.message media wa_id=%s media=%s", message.wa_id, message.media_id) + media = await download_media(message.media_id) + if media is None: + logger.info("api.message skipped — media download unavailable") + return {"response": None} + image_bytes, mime_type = media + response = await run_in_threadpool( + chat, + image_bytes=image_bytes, + image_media_type=mime_type, + session_id=message.wa_id, + ) + logger.info("api.message done wa_id=%s response_chars=%d", message.wa_id, len(response)) + return {"response": response} + + logger.info("api.message skipped — no actionable payload") + return {"response": None} @app.get("/health") diff --git a/src/chat.py b/src/chat.py index 94467f1..8e34d24 100644 --- a/src/chat.py +++ b/src/chat.py @@ -3,8 +3,17 @@ from collections.abc import Sequence from dotenv import load_dotenv -from pydantic_ai.messages import BinaryImage, ImageUrl, TextContent, UserContent - +from pydantic_ai.messages import ( + BinaryImage, + ImageUrl, + ModelMessage, + TextContent, + TextPart, + UserContent, + UserPromptPart, +) + +from .orchestrator.context import SessionContext from .router import AgentRouter from .session import Session from .session_store import FirestoreSessionStore @@ -13,11 +22,29 @@ DEFAULT_IMAGE_MEDIA_TYPE = "image/jpeg" IMAGE_PROMPT_TEXT = "The user sent this image on WhatsApp. Analyze it and provide a helpful response." -IMAGE_ROUTING_TEXT = "A WhatsApp user sent an image and needs help interpreting or responding to it." +IMAGE_ROUTING_TEXT = "A WhatsApp user uploaded a document image such as a death certificate for verification." +# Text notice given to the model when an image arrives. The image bytes go to the +# transient store (for tools to pull), never into the model's prompt. +IMAGE_ARRIVED_PROMPT = "The user has just uploaded a document image." _router: AgentRouter | None = None +def _render_history_text(history: Sequence[ModelMessage] | None) -> str: + """Flatten prior user/assistant turns into plain text for use as a narrative.""" + lines: list[str] = [] + for message in history or []: + for part in getattr(message, "parts", []): + content = getattr(part, "content", None) + if not isinstance(content, str) or not content.strip(): + continue + if isinstance(part, UserPromptPart): + lines.append(f"user: {content.strip()}") + elif isinstance(part, TextPart): + lines.append(f"assistant: {content.strip()}") + return "\n".join(lines) + + def _get_router() -> AgentRouter: global _router if _router is None: @@ -36,19 +63,24 @@ def chat( ) -> str: """Route one WhatsApp text or image message through the agent loop and return the response.""" - prompt = _build_prompt( + store = FirestoreSessionStore() if session_id is not None else None + history = store.load_history(session_id) if store is not None else None + history_text = _render_history_text(history) + + prompt, route_query = _prepare_turn( text=text, image_bytes=image_bytes, image_url=image_url, image_media_type=image_media_type, + history_text=history_text, + store=store, + session_id=session_id, ) - route_query = text.strip() if text is not None else IMAGE_ROUTING_TEXT router = _get_router() agent, _metadata = router.route_with_metadata(route_query) - store = FirestoreSessionStore() if session_id is not None else None - history = store.load_history(session_id) if store is not None else None - session = Session(agent, history=history) + deps = SessionContext(session_id=session_id, store=store, history_text=history_text) + session = Session(agent, history=history, deps=deps) response = "".join(session.send_stream(prompt)) if store is not None: @@ -61,6 +93,42 @@ def chat( return response +def _prepare_turn( + *, + text: str | None, + image_bytes: bytes | None, + image_url: str | None, + image_media_type: str, + history_text: str, + store: FirestoreSessionStore | None, + session_id: str | None, +) -> tuple[str | Sequence[UserContent], str]: + """Return (model_prompt, routing_query) and persist any uploaded image. + + An uploaded image (image_bytes) is saved to the transient store and the model + is handed only a text notice — the bytes are pulled later by a context-aware + tool and never enter the model's prompt. + """ + supplied = sum(value is not None for value in (text, image_bytes, image_url)) + if supplied != 1: + raise ValueError("Provide exactly one of text, image_bytes, or image_url") + + if image_bytes is not None: + if not image_bytes: + raise ValueError("image_bytes must not be empty") + if store is not None and session_id is not None: + store.save_media(session_id, image_bytes, mime_type=image_media_type) + return IMAGE_ARRIVED_PROMPT, (history_text or IMAGE_ROUTING_TEXT) + + if text is not None: + prompt = _build_prompt(text=text, image_bytes=None, image_url=None, image_media_type=image_media_type) + return prompt, text.strip() + + # image_url — legacy path: the referenced image is passed to the model directly. + prompt = _build_prompt(text=None, image_bytes=None, image_url=image_url, image_media_type=image_media_type) + return prompt, IMAGE_ROUTING_TEXT + + def _build_prompt( *, text: str | None, diff --git a/src/orchestrator/agent.py b/src/orchestrator/agent.py index c3b454a..09451bd 100644 --- a/src/orchestrator/agent.py +++ b/src/orchestrator/agent.py @@ -11,12 +11,14 @@ from pydantic import Field from pydantic_ai import Agent as PydanticAgent +from pydantic_ai import RunContext from pydantic_ai.messages import ModelMessage, UserContent from pydantic_ai.models import Model from pydantic_ai.models.google import GoogleModel from pydantic_ai.providers.google import GoogleProvider from . import prompts, tools +from .context import SessionContext logger = logging.getLogger(__name__) @@ -92,6 +94,7 @@ def __init__( model=model, system_prompt=enhanced_prompt, name=self.name, + deps_type=SessionContext, ) logger.info( "agent.model_init elapsed=%.3fs name=%s model=%s settings=%s", @@ -142,7 +145,7 @@ def _build_model(cls, provider: dict[str, Any]) -> tuple[Model, dict[str, Any]]: @classmethod def _build_gemini_model(cls, provider: dict[str, Any], model_name: str) -> tuple[GoogleModel, dict[str, Any]]: project = provider.get("project") or os.getenv("GOOGLE_CLOUD_PROJECT") - location = provider.get("location") or os.getenv("VERTEX_LOCATION", "us-east1") + location = provider.get("location") or os.getenv("VERTEX_LOCATION", "us-central1") model_settings = cls._model_settings_from_provider(provider, model_name) model = GoogleModel( @@ -180,13 +183,29 @@ def _json_schema_to_python_type(cls, schema: dict[str, Any]) -> Any: "boolean": bool, }.get(str(json_type), str) - def _apply_tool_signature(self, tool_func: Any, tool_def: tools.ToolDefinition) -> None: + def _apply_tool_signature( + self, + tool_func: Any, + tool_def: tools.ToolDefinition, + include_context: bool = False, + ) -> None: schema = tool_def.input_schema or {} properties = schema.get("properties", {}) or {} required = set(schema.get("required", []) or []) parameters: list[Parameter] = [] annotations: dict[str, Any] = {} + if include_context: + # pydantic-ai identifies the RunContext parameter by annotation and + # excludes it from the LLM-visible schema; the image is never a param. + parameters.append( + Parameter( + "ctx", + Parameter.POSITIONAL_OR_KEYWORD, + annotation=RunContext[SessionContext], + ) + ) + annotations["ctx"] = RunContext[SessionContext] for field_name, prop_schema in properties.items(): prop_schema = prop_schema or {} description = prop_schema.get("description", "") @@ -225,64 +244,99 @@ def _register_tool(self, tool_name: str) -> None: handler = self._resolve_handler(handler_path) - if iscoroutinefunction(handler): - async def _tool_fn(**kwargs: Any) -> Any: - tool_start = perf_counter() - args = {key: value for key, value in kwargs.items() if value is not None} - logger.info("agent.tool_call start tool=%s arg_keys=%s", tool_def.name, sorted(args.keys())) - try: - result = await handler(**args) - except Exception: - logger.error( - "agent.tool_call done tool=%s elapsed=%.3fs status=failed", - tool_def.name, - perf_counter() - tool_start, - ) - raise - logger.info( - "agent.tool_call done tool=%s elapsed=%.3fs status=success result_type=%s", - tool_def.name, - perf_counter() - tool_start, - type(result).__name__, - ) - return result + # Context-aware tools receive the run's SessionContext (via RunContext) so + # they can pull out-of-band state — e.g. the user's most recent image — + # without it passing through the model. Plain tools keep the original path. + needs_context = bool(tool_def.implementation.get("needs_context")) + is_async = iscoroutinefunction(handler) + + def _log_start(args: dict[str, Any]) -> float: + tool_start = perf_counter() + logger.info("agent.tool_call start tool=%s ctx=%s arg_keys=%s", tool_def.name, needs_context, sorted(args.keys())) + return tool_start + + def _log_done(tool_start: float, result: Any) -> None: + logger.info( + "agent.tool_call done tool=%s elapsed=%.3fs status=success result_type=%s", + tool_def.name, + perf_counter() - tool_start, + type(result).__name__, + ) + + def _log_failed(tool_start: float) -> None: + logger.error( + "agent.tool_call done tool=%s elapsed=%.3fs status=failed", + tool_def.name, + perf_counter() - tool_start, + ) + + if needs_context: + if is_async: + async def _tool_fn(ctx: RunContext[SessionContext], **kwargs: Any) -> Any: + args = {key: value for key, value in kwargs.items() if value is not None} + tool_start = _log_start(args) + try: + result = await handler(ctx, **args) + except Exception: + _log_failed(tool_start) + raise + _log_done(tool_start, result) + return result + else: + def _tool_fn(ctx: RunContext[SessionContext], **kwargs: Any) -> Any: + args = {key: value for key, value in kwargs.items() if value is not None} + tool_start = _log_start(args) + try: + result = handler(ctx, **args) + except Exception: + _log_failed(tool_start) + raise + _log_done(tool_start, result) + return result + + register = self.pydantic_ai_agent.tool else: - def _tool_fn(**kwargs: Any) -> Any: - tool_start = perf_counter() - args = {key: value for key, value in kwargs.items() if value is not None} - logger.info("agent.tool_call start tool=%s arg_keys=%s", tool_def.name, sorted(args.keys())) - try: - result = handler(**args) - except Exception: - logger.error( - "agent.tool_call done tool=%s elapsed=%.3fs status=failed", - tool_def.name, - perf_counter() - tool_start, - ) - raise - logger.info( - "agent.tool_call done tool=%s elapsed=%.3fs status=success result_type=%s", - tool_def.name, - perf_counter() - tool_start, - type(result).__name__, - ) - return result + if is_async: + async def _tool_fn(**kwargs: Any) -> Any: + args = {key: value for key, value in kwargs.items() if value is not None} + tool_start = _log_start(args) + try: + result = await handler(**args) + except Exception: + _log_failed(tool_start) + raise + _log_done(tool_start, result) + return result + else: + def _tool_fn(**kwargs: Any) -> Any: + args = {key: value for key, value in kwargs.items() if value is not None} + tool_start = _log_start(args) + try: + result = handler(**args) + except Exception: + _log_failed(tool_start) + raise + _log_done(tool_start, result) + return result + + register = self.pydantic_ai_agent.tool_plain _tool_fn.__name__ = f"tool_{tool_def.name}" _tool_fn.__doc__ = tool_def.description or f"Tool: {tool_def.name}" - self._apply_tool_signature(_tool_fn, tool_def) + self._apply_tool_signature(_tool_fn, tool_def, include_context=needs_context) - self.pydantic_ai_agent.tool_plain( + register( _tool_fn, name=tool_def.name, description=tool_def.description, ) - logger.info("agent.tool_registered elapsed=%.3fs tool=%s", perf_counter() - start, tool_name) + logger.info("agent.tool_registered elapsed=%.3fs tool=%s ctx=%s", perf_counter() - start, tool_name, needs_context) def run_stream( self, user_message: str | Sequence[UserContent], message_history: Optional[Sequence[ModelMessage]] = None, + deps: Optional[SessionContext] = None, ) -> Any: start = perf_counter() logger.info("agent.run_stream start name=%s history=%d", self.name, len(message_history or [])) @@ -290,6 +344,7 @@ def run_stream( user_message, message_history=message_history, model_settings=self.model_settings, + deps=deps if deps is not None else SessionContext(), ) logger.info("agent.run_stream returned elapsed=%.3fs name=%s", perf_counter() - start, self.name) return streamed diff --git a/src/orchestrator/context.py b/src/orchestrator/context.py new file mode 100644 index 0000000..e2bcf9e --- /dev/null +++ b/src/orchestrator/context.py @@ -0,0 +1,22 @@ +"""Per-run session context passed to context-aware tools. + +Threaded through pydantic-ai as the agent's ``deps`` so a tool can pull +out-of-band state — most importantly the user's most recent uploaded image — +without that data ever passing through the model. This keeps the mechanism +generic: any tool that declares ``needs_context`` receives this object and can +reach the transient store or the rendered conversation. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass +class SessionContext: + """Runtime context for a single agent turn.""" + + session_id: str | None = None + store: Any | None = None # FirestoreSessionStore (history + media) + history_text: str = "" # rendered prior conversation, used as narrative diff --git a/src/session.py b/src/session.py index ead229a..ec1c290 100644 --- a/src/session.py +++ b/src/session.py @@ -5,14 +5,21 @@ from pydantic_ai.messages import ModelMessage, UserContent from .orchestrator.agent import Agent +from .orchestrator.context import SessionContext logger = logging.getLogger(__name__) class Session: - def __init__(self, agent: Agent, history: Sequence[ModelMessage] | None = None): + def __init__( + self, + agent: Agent, + history: Sequence[ModelMessage] | None = None, + deps: SessionContext | None = None, + ): self.agent = agent self._history: list[ModelMessage] = list(history or []) + self._deps = deps @property def history(self) -> list[ModelMessage]: @@ -21,7 +28,7 @@ def history(self) -> list[ModelMessage]: def send_stream(self, user_message: str | Sequence[UserContent]): start = perf_counter() logger.info("session.stream start history=%d input_type=%s", len(self._history), type(user_message).__name__) - streamed = self.agent.run_stream(user_message, message_history=self._history) + streamed = self.agent.run_stream(user_message, message_history=self._history, deps=self._deps) logger.info("session.stream object returned elapsed=%.3fs", perf_counter() - start) first_chunk_at: float | None = None diff --git a/src/session_store.py b/src/session_store.py index e5910c1..fcd08d7 100644 --- a/src/session_store.py +++ b/src/session_store.py @@ -1,5 +1,7 @@ from __future__ import annotations +import base64 +import logging import os from collections.abc import Callable, Sequence from datetime import datetime, timedelta, timezone @@ -7,8 +9,14 @@ from pydantic_ai.messages import ModelMessage, ModelMessagesTypeAdapter +logger = logging.getLogger(__name__) + DEFAULT_SESSION_TTL = timedelta(hours=72) +# Firestore caps a document at ~1 MB; base64 inflates bytes by ~33%. Skip anything +# whose encoded form would risk the limit and log it rather than fail the write. +MAX_MEDIA_BYTES = 700_000 + class FirestoreSessionStore: def __init__( @@ -17,6 +25,7 @@ def __init__( client: Any | None = None, project: str | None = None, collection: str = "sessions", + media_collection: str = "session_media", server_timestamp: Any | None = None, ttl: timedelta = DEFAULT_SESSION_TTL, now: Callable[[], datetime] | None = None, @@ -24,6 +33,7 @@ def __init__( firestore = None if client is not None else _firestore_module() self._client = client or firestore.Client(project=project or os.getenv("GOOGLE_CLOUD_PROJECT")) self._collection = collection + self._media_collection = media_collection self._server_timestamp = ( server_timestamp if server_timestamp is not None @@ -73,9 +83,62 @@ def save_history( document.set(data, merge=True) + def save_media(self, session_id: str, image_bytes: bytes, *, mime_type: str) -> bool: + """Persist the most recent inbound image for a session. + + Stores one document per session (overwritten on each upload) so tools can + later pull the latest media by session_id without the image ever passing + through the model. Returns False (and skips the write) if the image is too + large for a Firestore document. + """ + if not image_bytes: + return False + if len(image_bytes) > MAX_MEDIA_BYTES: + logger.warning( + "session_store.save_media skipped — image too large bytes=%d limit=%d session=%.8s", + len(image_bytes), + MAX_MEDIA_BYTES, + session_id, + ) + return False + + document = self._media_document(session_id) + document.set( + { + "session_id": session_id, + "image_b64": base64.b64encode(image_bytes).decode("ascii"), + "mime_type": mime_type, + "created_at": self._server_timestamp, + "expires_at": self._now() + self._ttl, + } + ) + return True + + def load_latest_media(self, session_id: str) -> tuple[bytes, str] | None: + """Return (image_bytes, mime_type) for the most recent image, or None.""" + document = self._media_document(session_id) + snapshot = document.get() + if not snapshot.exists: + return None + + data = snapshot.to_dict() or {} + expires_at = data.get("expires_at") + if isinstance(expires_at, datetime) and expires_at <= self._now(): + document.delete() + return None + + encoded = data.get("image_b64") + if not encoded: + return None + + return base64.b64decode(encoded), data.get("mime_type") or "image/jpeg" + def _document(self, session_id: str) -> Any: return self._client.collection(self._collection).document(session_id) + def _media_document(self, session_id: str) -> Any: + return self._client.collection(self._media_collection).document(session_id) + def _firestore_module() -> Any: from google.cloud import firestore diff --git a/src/whatsapp_media.py b/src/whatsapp_media.py new file mode 100644 index 0000000..53568dd --- /dev/null +++ b/src/whatsapp_media.py @@ -0,0 +1,65 @@ +"""Resolve a WhatsApp media ID to raw bytes via the Meta Graph API. + +Inbound WhatsApp image/document messages carry only a media *ID*. Turning that +into bytes is a two-step authenticated call: + + 1. GET /{media-id} → JSON with a short-lived, authenticated URL + 2. GET → the media bytes + +Both require a WhatsApp access token (``WHATSAPP_TOKEN``). If the token is not +configured the function logs and returns None so the caller can degrade +gracefully rather than crash — this is the one external dependency to provision. + +This module is intentionally the single seam for media retrieval: if the upstream +webhook is later changed to pre-resolve media to bytes/URL, only this file and the +payload extractor in api.py need to change. +""" + +from __future__ import annotations + +import logging +import os + +import httpx + +logger = logging.getLogger(__name__) + +_GRAPH_BASE = "https://graph.facebook.com" +_DEFAULT_GRAPH_VERSION = "v21.0" +_TIMEOUT = 30.0 + + +async def download_media(media_id: str) -> tuple[bytes, str] | None: + """Return (bytes, mime_type) for a WhatsApp media ID, or None if unavailable.""" + token = os.getenv("WHATSAPP_TOKEN", "") + if not token: + logger.warning("whatsapp_media.download skipped — WHATSAPP_TOKEN not set") + return None + if not media_id: + return None + + version = os.getenv("WHATSAPP_GRAPH_VERSION", _DEFAULT_GRAPH_VERSION) + headers = {"Authorization": f"Bearer {token}"} + + try: + async with httpx.AsyncClient(timeout=_TIMEOUT) as client: + meta_resp = await client.get(f"{_GRAPH_BASE}/{version}/{media_id}", headers=headers) + meta_resp.raise_for_status() + meta = meta_resp.json() + + media_url = meta.get("url") + mime_type = meta.get("mime_type") or "image/jpeg" + if not media_url: + logger.warning("whatsapp_media.download no url in metadata media=%s", media_id) + return None + + # The media URL must be fetched with the same bearer token. + bytes_resp = await client.get(media_url, headers=headers) + bytes_resp.raise_for_status() + data = bytes_resp.content + except httpx.HTTPError as exc: + logger.error("whatsapp_media.download failed media=%s error=%s", media_id, exc) + return None + + logger.info("whatsapp_media.download ok media=%s bytes=%d mime=%s", media_id, len(data), mime_type) + return data, mime_type diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index db03131..c9f5975 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -112,30 +112,34 @@ def _make_client(secret: str = "", monkeypatch=None): def test_extract_message_text(): from src.api import _extract_message - wa_id, text = _extract_message(SAMPLE_TEXT_PAYLOAD) - assert wa_id == "16508106640" - assert text == "Hello B2, what is Givelight?" + msg = _extract_message(SAMPLE_TEXT_PAYLOAD) + assert msg.wa_id == "16508106640" + assert msg.text == "Hello B2, what is Givelight?" + assert msg.media_id is None -def test_extract_message_status_update_returns_none(): +def test_extract_message_status_update_returns_empty(): from src.api import _extract_message - wa_id, text = _extract_message(STATUS_UPDATE_PAYLOAD) - assert wa_id is None - assert text is None + msg = _extract_message(STATUS_UPDATE_PAYLOAD) + assert msg.text is None + assert msg.media_id is None -def test_extract_message_image_returns_none(): +def test_extract_message_image_returns_media_id(): from src.api import _extract_message - wa_id, text = _extract_message(IMAGE_PAYLOAD) - assert wa_id is None - assert text is None + msg = _extract_message(IMAGE_PAYLOAD) + assert msg.wa_id == "16508106640" + assert msg.text is None + assert msg.media_id == "media-123" + assert msg.mime_type == "image/jpeg" def test_extract_message_empty_payload(): from src.api import _extract_message - wa_id, text = _extract_message({}) - assert wa_id is None - assert text is None + msg = _extract_message({}) + assert msg.wa_id is None + assert msg.text is None + assert msg.media_id is None def test_extract_message_blank_text(): @@ -143,9 +147,9 @@ def test_extract_message_blank_text(): payload = { "entry": [{"changes": [{"value": {"messages": [{"type": "text", "text": {"body": " "}, "from": "123"}]}}]}] } - wa_id, text = _extract_message(payload) - assert wa_id is None - assert text is None + msg = _extract_message(payload) + assert msg.text is None + assert msg.media_id is None # --------------------------------------------------------------------------- @@ -180,10 +184,16 @@ def fake_chat(**kwargs): assert r.json() == {"response": None} -def test_message_image_returns_null(monkeypatch): +def test_message_image_download_unavailable_returns_null(monkeypatch): + """If media can't be downloaded (e.g. no WHATSAPP_TOKEN), respond with null.""" import src.api as api_module monkeypatch.setattr(api_module, "_WEBHOOK_SECRET", "") - monkeypatch.setattr(api_module, "chat", lambda **kw: "nope") + monkeypatch.setattr(api_module, "chat", lambda **kw: "should not be called") + + async def fake_download(media_id): + return None + + monkeypatch.setattr(api_module, "download_media", fake_download) client = TestClient(api_module.app) r = client.post("/message", json=IMAGE_PAYLOAD) @@ -191,6 +201,36 @@ def test_message_image_returns_null(monkeypatch): assert r.json() == {"response": None} +def test_message_image_downloads_and_calls_chat(monkeypatch): + """A downloadable image is fetched and routed to chat as image_bytes.""" + import src.api as api_module + monkeypatch.setattr(api_module, "_WEBHOOK_SECRET", "") + + captured = {} + + async def fake_download(media_id): + captured["media_id"] = media_id + return b"\xff\xd8jpeg-bytes", "image/jpeg" + + def fake_chat(*, image_bytes=None, image_media_type=None, session_id=None, **kwargs): + captured["image_bytes"] = image_bytes + captured["image_media_type"] = image_media_type + captured["session_id"] = session_id + return "Your request is being processed." + + monkeypatch.setattr(api_module, "download_media", fake_download) + monkeypatch.setattr(api_module, "chat", fake_chat) + + client = TestClient(api_module.app) + r = client.post("/message", json=IMAGE_PAYLOAD) + assert r.status_code == 200 + assert r.json() == {"response": "Your request is being processed."} + assert captured["media_id"] == "media-123" + assert captured["image_bytes"] == b"\xff\xd8jpeg-bytes" + assert captured["image_media_type"] == "image/jpeg" + assert captured["session_id"] == "16508106640" + + def test_message_text_calls_chat(monkeypatch): import src.api as api_module monkeypatch.setattr(api_module, "_WEBHOOK_SECRET", "") diff --git a/tests/unit/test_chat.py b/tests/unit/test_chat.py index 21171e1..be0923e 100644 --- a/tests/unit/test_chat.py +++ b/tests/unit/test_chat.py @@ -1,7 +1,21 @@ -from pydantic_ai.messages import BinaryImage, ImageUrl, TextContent +from pydantic_ai.messages import ( + BinaryImage, + ImageUrl, + ModelRequest, + ModelResponse, + TextContent, + TextPart, + UserPromptPart, +) from src import chat as chat_module -from src.chat import IMAGE_PROMPT_TEXT, _build_prompt, chat +from src.chat import ( + IMAGE_ARRIVED_PROMPT, + IMAGE_PROMPT_TEXT, + _build_prompt, + _render_history_text, + chat, +) def test_build_prompt_accepts_text_only() -> None: @@ -66,9 +80,10 @@ def route_with_metadata(self, query: str): return object(), {"score": 1.0} class FakeSession: - def __init__(self, agent: object, history=None): + def __init__(self, agent: object, history=None, deps=None): self.agent = agent self.history = list(history or []) + self.deps = deps created_sessions.append(self) def send_stream(self, prompt: str): @@ -114,9 +129,10 @@ def save_history(self, session_id: str, history, *, agent_name=None, channel=Non saved["channel"] = channel class FakeSession: - def __init__(self, agent: object, history=None): + def __init__(self, agent: object, history=None, deps=None): assert isinstance(agent, FakeAgent) assert history == loaded_history + self.deps = deps self.history = ["updated"] def send_stream(self, prompt: str): @@ -136,3 +152,71 @@ def send_stream(self, prompt: str): "agent_name": "support", "channel": "sms", } + + +def test_render_history_text_flattens_user_and_assistant_turns() -> None: + history = [ + ModelRequest(parts=[UserPromptPart(content="my mother passed away")]), + ModelResponse(parts=[TextPart(content="I'm so sorry for your loss")]), + ] + assert _render_history_text(history) == ( + "user: my mother passed away\nassistant: I'm so sorry for your loss" + ) + + +def test_chat_image_turn_stores_media_and_hides_image_from_model(monkeypatch) -> None: + """An uploaded image is saved to the store; the model only sees a text notice.""" + saved_media = {} + captured = {} + + class FakeAgent: + name = "death_certificate_poc_agent" + + class FakeRouter: + def route_with_metadata(self, query: str): + captured["route_query"] = query + return FakeAgent(), {"score": 1.0} + + class FakeStore: + def load_history(self, session_id: str): + # prior conversation establishes the death-certificate context + return [ModelRequest(parts=[UserPromptPart(content="my mother passed away")])] + + def save_media(self, session_id: str, image_bytes: bytes, *, mime_type: str) -> bool: + saved_media["session_id"] = session_id + saved_media["bytes"] = image_bytes + saved_media["mime_type"] = mime_type + return True + + def save_history(self, session_id, history, *, agent_name=None, channel=None) -> None: + pass + + class FakeSession: + def __init__(self, agent: object, history=None, deps=None): + captured["deps"] = deps + self.history = list(history or []) + + def send_stream(self, prompt): + captured["prompt"] = prompt + yield "Your request is being processed." + + monkeypatch.setattr(chat_module, "load_dotenv", lambda: None) + monkeypatch.setattr(chat_module, "_router", None) + monkeypatch.setattr(chat_module, "AgentRouter", FakeRouter) + monkeypatch.setattr(chat_module, "Session", FakeSession) + monkeypatch.setattr(chat_module, "FirestoreSessionStore", FakeStore) + + result = chat(image_bytes=b"\xff\xd8jpeg", image_media_type="image/jpeg", session_id="wa-1") + + assert result == "Your request is being processed." + # image persisted to the transient store + assert saved_media == {"session_id": "wa-1", "bytes": b"\xff\xd8jpeg", "mime_type": "image/jpeg"} + # the model prompt is a plain text notice — never the image bytes + assert captured["prompt"] == IMAGE_ARRIVED_PROMPT + assert not isinstance(captured["prompt"], list) + # routing used the prior conversation, biasing toward the death-cert agent + assert "my mother passed away" in captured["route_query"] + # deps carry the store + session so the tool can pull the image + assert captured["deps"].session_id == "wa-1" + assert captured["deps"].store is not None + assert "my mother passed away" in captured["deps"].history_text diff --git a/tests/unit/test_session_store.py b/tests/unit/test_session_store.py index 5e56814..3e81bfd 100644 --- a/tests/unit/test_session_store.py +++ b/tests/unit/test_session_store.py @@ -1,8 +1,9 @@ +import base64 from datetime import datetime, timedelta, timezone from pydantic_ai.messages import ModelMessagesTypeAdapter, ModelRequest, ModelResponse, TextPart, UserPromptPart -from src.session_store import FirestoreSessionStore +from src.session_store import MAX_MEDIA_BYTES, FirestoreSessionStore class FakeSnapshot: @@ -133,3 +134,112 @@ def test_save_history_serializes_messages_and_sets_metadata_on_create() -> None: assert data["channel"] == "whatsapp" assert data["history"][0]["kind"] == "request" assert data["history"][1]["kind"] == "response" + + +# --------------------------------------------------------------------------- +# transient media store +# --------------------------------------------------------------------------- + + +class FakeMediaDocument: + def __init__(self, snapshot: FakeSnapshot): + self.snapshot = snapshot + self.writes = [] + self.deletes = 0 + + def get(self): + return self.snapshot + + def set(self, data, merge: bool = False): + self.writes.append((data, merge)) + self.snapshot = FakeSnapshot(True, data) + + def delete(self): + self.deletes += 1 + self.snapshot = FakeSnapshot(False) + + +class FakeMediaClient: + def __init__(self, document: FakeMediaDocument): + self.document_ref = document + self.collection_names: list[str] = [] + + def collection(self, name: str): + self.collection_names.append(name) + return FakeCollection(self.document_ref) + + +def _media_store(document: FakeMediaDocument, now: datetime): + return FirestoreSessionStore( + client=FakeMediaClient(document), + server_timestamp="SERVER_TIME", + now=lambda: now, + ) + + +def test_save_media_stores_base64_and_metadata() -> None: + now = datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) + document = FakeMediaDocument(FakeSnapshot(False)) + client = FakeMediaClient(document) + store = FirestoreSessionStore(client=client, server_timestamp="SERVER_TIME", now=lambda: now) + + ok = store.save_media("session-1", b"\xff\xd8abc", mime_type="image/jpeg") + + assert ok is True + assert "session_media" in client.collection_names + data, _merge = document.writes[0] + assert data["session_id"] == "session-1" + assert data["image_b64"] == base64.b64encode(b"\xff\xd8abc").decode("ascii") + assert data["mime_type"] == "image/jpeg" + assert data["expires_at"] == now + timedelta(hours=72) + + +def test_load_latest_media_round_trips_bytes() -> None: + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + document = FakeMediaDocument( + FakeSnapshot( + True, + { + "image_b64": base64.b64encode(b"\xff\xd8abc").decode("ascii"), + "mime_type": "image/png", + "expires_at": now + timedelta(hours=1), + }, + ) + ) + store = _media_store(document, now) + + assert store.load_latest_media("session-1") == (b"\xff\xd8abc", "image/png") + + +def test_load_latest_media_deletes_expired_and_returns_none() -> None: + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + document = FakeMediaDocument( + FakeSnapshot( + True, + { + "image_b64": base64.b64encode(b"abc").decode("ascii"), + "expires_at": now, + }, + ) + ) + store = _media_store(document, now) + + assert store.load_latest_media("session-1") is None + assert document.deletes == 1 + + +def test_load_latest_media_returns_none_when_missing() -> None: + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + store = _media_store(FakeMediaDocument(FakeSnapshot(False)), now) + assert store.load_latest_media("session-1") is None + + +def test_save_media_skips_oversized_image() -> None: + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + document = FakeMediaDocument(FakeSnapshot(False)) + store = _media_store(document, now) + + ok = store.save_media("session-1", b"x" * (MAX_MEDIA_BYTES + 1), mime_type="image/jpeg") + + assert ok is False + assert document.writes == [] diff --git a/tests/unit/test_verify_flow_integration.py b/tests/unit/test_verify_flow_integration.py new file mode 100644 index 0000000..afa8e9a --- /dev/null +++ b/tests/unit/test_verify_flow_integration.py @@ -0,0 +1,73 @@ +"""Integration test: real orchestrator agent + verify tool, LLM faked with TestModel. + +Drives the novel mechanism end to end — the model calls the context-aware tool, +which pulls the user's image from the transient store (via SessionContext deps) +and runs the pipeline + handoff. Only the external boundaries (the LLM, the +reliability pipeline, and GiveLight delivery) are faked; the tool registration, +deps threading, and image-through-context path are all exercised for real. +""" + +from pathlib import Path + +from pydantic_ai.models.test import TestModel + +from src.orchestrator import agent as agent_module +from src.orchestrator.context import SessionContext +from src.session import Session +from tools.death_certificate_pipeline import verify as verify_module +from tools.death_certificate_pipeline.models import Band, ReliabilityResult + +_AGENT_YAML = Path(__file__).resolve().parents[2] / "agents" / "poc_deathCertParserAgent.yaml" + + +class _FakeStore: + def __init__(self, media): + self._media = media + self.pulled_for = None + + def load_latest_media(self, session_id): + self.pulled_for = session_id + return self._media + + +def test_model_call_pulls_image_and_hands_off(monkeypatch): + monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-proj") + + seen = {} + + async def fake_pipeline(submission): + seen["narrative"] = submission.narrative + seen["image"] = submission.image + return ReliabilityResult( + score=82, band=Band.HIGH, sub_scores={}, weights={}, flags=[], + justification="ok", extracted_fields={"full_name": "Jane Doe"}, + ) + + async def fake_deliver(payload): + seen["handoff"] = payload + return True + + monkeypatch.setattr(verify_module, "run_pipeline", fake_pipeline) + monkeypatch.setattr(verify_module, "deliver_to_gl", fake_deliver) + + definition = agent_module._load_agent_definition_from_file(_AGENT_YAML) + agent = agent_module.Agent(definition) + + store = _FakeStore((b"\xff\xd8jpeg-image-bytes", "image/jpeg")) + deps = SessionContext( + session_id="wa-1", + store=store, + history_text="My mother Jane Doe passed away last week.", + ) + + with agent.pydantic_ai_agent.override(model=TestModel()): + session = Session(agent, deps=deps) + "".join(session.send_stream("The user has just uploaded a document image.")) + + # the tool pulled the image for the right session, out of band from the model + assert store.pulled_for == "wa-1" + assert seen["image"] == b"\xff\xd8jpeg-image-bytes" + assert seen["narrative"] == "My mother Jane Doe passed away last week." + # and the passing result was forwarded to GiveLight + assert seen["handoff"]["score"] == 82 + assert len(seen["handoff"]["contact_identifier"]) == 64 diff --git a/tests/unit/tools/death_certificate_pipeline/test_verify.py b/tests/unit/tools/death_certificate_pipeline/test_verify.py new file mode 100644 index 0000000..34ff3fa --- /dev/null +++ b/tests/unit/tools/death_certificate_pipeline/test_verify.py @@ -0,0 +1,91 @@ +"""Unit tests for the context-aware death-certificate verification tool.""" + +from types import SimpleNamespace + +from tools.death_certificate_pipeline import verify as verify_module +from tools.death_certificate_pipeline.models import Band, ReliabilityResult +from tools.death_certificate_pipeline.verify import verify_death_certificate + + +class FakeStore: + def __init__(self, media): + self._media = media + + def load_latest_media(self, session_id): + return self._media + + +def _ctx(store, history_text="my mother Jane Doe passed away", session_id="wa-123"): + return SimpleNamespace( + deps=SimpleNamespace(session_id=session_id, store=store, history_text=history_text) + ) + + +def _result(band, flags=None): + return ReliabilityResult( + score=80 if band in (Band.HIGH, Band.MEDIUM) else 30, + band=band, + sub_scores={}, + weights={}, + flags=flags or [], + justification="ok", + extracted_fields={"full_name": "Jane Doe"}, + ) + + +async def test_verify_accepts_and_hands_off(monkeypatch): + seen = {} + + async def fake_pipeline(submission): + seen["narrative"] = submission.narrative + return _result(Band.HIGH) + + async def fake_deliver(payload): + seen["payload"] = payload + return True + + monkeypatch.setattr(verify_module, "run_pipeline", fake_pipeline) + monkeypatch.setattr(verify_module, "deliver_to_gl", fake_deliver) + + result = await verify_death_certificate(_ctx(FakeStore((b"\xff\xd8jpeg", "image/jpeg")))) + + assert result["status"] == "verified" + assert result["band"] == "high" + assert result["handed_off"] is True + # narrative came from the conversation history, not the model + assert "Jane Doe" in seen["narrative"] + # handoff payload carries a non-reversible contact id and the score + assert seen["payload"]["score"] == 80 + assert seen["payload"]["contact_identifier"] and len(seen["payload"]["contact_identifier"]) == 64 + + +async def test_verify_holds_back_on_hard_escalation(monkeypatch): + delivered = {"called": False} + + async def fake_pipeline(submission): + return _result(Band.ESCALATE, flags=["HARD_ESCALATION"]) + + async def fake_deliver(payload): + delivered["called"] = True + return True + + monkeypatch.setattr(verify_module, "run_pipeline", fake_pipeline) + monkeypatch.setattr(verify_module, "deliver_to_gl", fake_deliver) + + result = await verify_death_certificate(_ctx(FakeStore((b"\xff\xd8jpeg", "image/jpeg")))) + + assert result["status"] == "verified" + assert result["handed_off"] is False + assert delivered["called"] is False + + +async def test_verify_reports_no_document_when_store_empty(monkeypatch): + async def fake_pipeline(submission): # pragma: no cover - must not run + raise AssertionError("pipeline should not run without media") + + monkeypatch.setattr(verify_module, "run_pipeline", fake_pipeline) + + result = await verify_death_certificate(_ctx(FakeStore(None), history_text="")) + + assert result["status"] == "no_document" + assert result["handed_off"] is False diff --git a/tools/deathCertificatePipeline.yaml b/tools/deathCertificatePipeline.yaml index c6f87b2..3671ffe 100644 --- a/tools/deathCertificatePipeline.yaml +++ b/tools/deathCertificatePipeline.yaml @@ -1,17 +1,13 @@ -name: death_certificate_pipeline -description: "Extracts death certificate facts and checks them against the chat history" +name: death_certificate_verification +description: > + Verifies the death certificate the user most recently uploaded: runs the full + reliability pipeline (document, authenticity, narrative consistency, score) and, + if it passes, hands the case off to GiveLight. Call this once the user has sent a + document image — the image is pulled automatically and you do not pass it in. input_schema: type: object - properties: - chat_history: - type: string - description: "Chat history containing the narrative claims to compare" - image_bytes: - type: string - description: "Base64-encoded bytes of the death certificate image" - required: - - chat_history - - image_bytes + properties: {} implementation: type: function - handler: tools.death_certificate_pipeline.death_certificate_consistency.analyze_death_certificate_consistency_base64 + needs_context: true + handler: tools.death_certificate_pipeline.verify.verify_death_certificate diff --git a/tools/death_certificate_pipeline/verify.py b/tools/death_certificate_pipeline/verify.py new file mode 100644 index 0000000..e9e57e7 --- /dev/null +++ b/tools/death_certificate_pipeline/verify.py @@ -0,0 +1,100 @@ +"""Context-aware death-certificate verification tool. + +Registered as the ``death_certificate_verification`` agent tool. The model calls +it with no arguments once a user has sent a document; the image itself is never +seen by the model. This handler pulls the user's most recent image from the +transient session store (via the run's SessionContext), runs the full +reliability pipeline, and — when the result clears the acceptance policy — hands +the case off to GiveLight. + +Flow: SessionContext → latest image → run_pipeline → (policy) → handoff. +""" + +from __future__ import annotations + +import hashlib +import logging +from typing import Any + +from tools.death_certificate_pipeline.handoff import build_handoff_payload, deliver_to_gl +from tools.death_certificate_pipeline.models import Band, ReliabilityResult, Submission +from tools.death_certificate_pipeline.pipeline import run_pipeline + +logger = logging.getLogger(__name__) + +# Bands that are strong enough to forward to GiveLight automatically. Anything +# lower — or any hard escalation flag from the authenticity stage — is held back +# for human review instead. +_ACCEPT_BANDS = {Band.HIGH, Band.MEDIUM} +_NARRATIVE_FALLBACK = "No prior conversation details were provided by the claimant." + + +def _contact_identifier(session_id: str | None) -> str | None: + """Stable, non-reversible join key for GiveLight (placeholder until GL-20).""" + if not session_id: + return None + return hashlib.sha256(session_id.encode("utf-8")).hexdigest() + + +def _accepted(result: ReliabilityResult) -> bool: + return result.band in _ACCEPT_BANDS and "HARD_ESCALATION" not in result.flags + + +async def verify_death_certificate(ctx: Any) -> dict[str, Any]: + """Verify the user's most recent document and hand off to GiveLight if it passes.""" + deps = getattr(ctx, "deps", None) + session_id = getattr(deps, "session_id", None) + store = getattr(deps, "store", None) + + media = store.load_latest_media(session_id) if (store is not None and session_id) else None + if media is None: + logger.info("verify.no_media session=%.8s", str(session_id or "")) + return { + "status": "no_document", + "handed_off": False, + "summary": "No document has been received yet — ask the user to send a photo or PDF of the death certificate.", + } + + image_bytes, _mime = media + narrative = (getattr(deps, "history_text", "") or "").strip() or _NARRATIVE_FALLBACK + + submission = Submission( + image=image_bytes, + narrative=narrative, + case_fields={"channel": "whatsapp", "contact_identifier": _contact_identifier(session_id)}, + ) + + result = await run_pipeline(submission) + handed_off = False + + if _accepted(result): + payload = build_handoff_payload( + result, + contact_identifier=_contact_identifier(session_id), + case_fields=submission.case_fields, + ) + handed_off = await deliver_to_gl(payload) + + logger.info( + "verify.done session=%.8s score=%d band=%s accepted=%s handed_off=%s", + str(session_id or ""), + result.score, + result.band.value, + _accepted(result), + handed_off, + ) + + return { + "status": "verified", + "score": result.score, + "band": result.band.value, + "handed_off": handed_off, + "flags": result.flags, + "extracted_fields": result.extracted_fields, + "summary": ( + "Verification passed and the case was forwarded to GiveLight." + if handed_off + else "Verification did not meet the automatic-approval threshold; the case needs human review. " + "Consider asking the user for a clearer photo of the full certificate." + ), + }