"""Client lookup Tools. All AUTOMATIC: they read, and every read is scoped by the CMS against this agent's identity (ADR-0003). Being wrong wastes a call; it cannot corrupt a record. """ from __future__ import annotations from typing import Any from app.api.client import AgentApiError, CmsClient, NotFound from app.tools.base import ExecutionPolicy, Tool, ToolResult from app.documents.identifiers import CNIC_TYPE, FILE_NUMBER_TYPE def _exactly_numbered(matches: list[dict[str, Any]], term: str) -> list[dict[str, Any]]: """The clients whose file number *is* the one asked for. Compared as text after trimming and folding case, not as integers: file numbers here carry single-letter suffixes ("16 A"), so a numeric comparison would either fail on them or, worse, read "16 A" as 16 and file a document against the wrong client. """ wanted = str(term or "").strip().casefold() if not wanted: return [] return [ client for client in matches if str(client.get("file_number") or "").strip().casefold() == wanted ] class SearchClientTool(Tool): """The AI's first move when a document or message arrives: whose is this?""" name = "search_client" description = ( "Find clients by name, file number, CNIC, mobile, email or city. " "Returns matches with their id, file number and status." ) policy = ExecutionPolicy.AUTOMATIC def __init__(self, cms: CmsClient) -> None: self._cms = cms def run(self, **kwargs: Any) -> ToolResult: query = (kwargs.get("query") or "").strip() if not query: # Refused rather than executed: an empty search returns the whole # client base, which is not a search, and a model that meant to # search for something would rather be told than handed everything. return ToolResult.failure("A search term is required.") try: response = self._cms.search_clients( search=query, search_type=kwargs.get("search_type", "name"), per_page=kwargs.get("limit", 10), ) except AgentApiError as exc: return ToolResult.failure(str(exc)) matches = response.get("data", []) return ToolResult.success( { "matches": matches, "total": response.get("meta", {}).get("total", len(matches)), # Stated explicitly so a workflow can branch on ambiguity rather # than assuming the first hit is the right one. Picking silently # is how a document ends up filed against the wrong client. "unambiguous": len(matches) == 1, } ) class FindClientByIdentifierTool(Tool): """Whose document this is, when a person said so rather than a document. The self-chat forwarding path. Nothing has read the file — a bank statement's contents never leave the installation and are not opened here at all (ADR-0002) — so the caption's identifier is the entire basis for the client, and the rules for resolving it are the whole safety story. ## The precedence, and why it is not "try one, then the other" A file number is preferred because it is this firm's own key: exact, one per client, and what somebody types when they mean a specific file. A CNIC is exact too, but it belongs to the person rather than to the record, and the two can legitimately disagree — a family filing under one number, or a CNIC copied from the wrong card. So when both are supplied, both are looked up. Preferring the file number and never running the second search would discard the one piece of evidence capable of showing that the caption is wrong. When they name different clients, nothing is chosen: both go to the reviewer as candidates and the disagreement is stated, because a contradicted file number is not "probably right" — the contradiction is itself the finding. """ name = "find_client_by_identifier" description = ( "Find the one client a supplied file number or CNIC names. " "Returns candidates rather than choosing when the answer is not exactly one." ) policy = ExecutionPolicy.AUTOMATIC def __init__(self, cms: CmsClient) -> None: self._cms = cms def run(self, **kwargs: Any) -> ToolResult: file_number = (kwargs.get("file_number") or "").strip() cnic = (kwargs.get("cnic") or "").strip() mobile = (kwargs.get("mobile") or "").strip() if not file_number and not cnic and not mobile: # Nothing to look up. Answered, not refused. # # This used to fail the step, on the reasoning that an empty search # returns the whole client base. That protection is real — and is # achieved by not searching, which is what happens here. Failing as # well threw the document away. # # Found by real traffic on staging: a bank statement recognised from # its filename, with no caption, reached the forwarding workflow # with no identifier and the run FAILED. No proposal, no queue # entry, nothing for anybody to look at — a document that would # previously have reached a reviewer with the client left blank # simply vanished. # # Zero matches is the same answer the CMS gives for an identifier # matching nobody, and every workflow already carries that to a # reviewer rather than failing on it. return ToolResult.success( { "matches": [], "unambiguous": False, "identifier": { "type": None, "value": None, "both_supplied": False, "agreed": None, }, "conflict": False, } ) try: by_file = self._search(file_number, FILE_NUMBER_TYPE) if file_number else None by_cnic = self._search(cnic, CNIC_TYPE) if cnic else None except AgentApiError as exc: # The CMS being unreachable is an operational fact, not a finding # about this document. Failing here leaves the run retryable with the # file still on disk — better than proposing a filing with no client # and making a person sort out a network problem by hand. return ToolResult.failure(str(exc)) resolved = self._resolve(file_number, cnic, by_file, by_cnic) if resolved["matches"] or not mobile: return ToolResult.success(resolved) return ToolResult.success(self._by_mobile(mobile)) # ── Internals ───────────────────────────────────────────────────────── def _search(self, term: str, search_type: str) -> list[dict[str, Any]]: # A small page, deliberately. More than a handful of matches for an # exact identifier means the identifier was not exact, and a reviewer # cannot usefully choose from fifty names anyway. response = self._cms.search_clients(search=term, search_type=search_type, per_page=10) matches = list(response.get("data", [])) if search_type == FILE_NUMBER_TYPE: # The CMS searches file numbers by substring, and it should: a # person typing "27" into the box is browsing. Nobody who sends # "File No 279" is browsing. # # Without this, searching 279 returns 3279, 2799, 2798 … — ten # near-misses, no two the same client, so the result reads as an # ambiguity and the reviewer is asked to choose between ten people # who all have the wrong number. Narrowing to the client whose file # number *is* 279 is not a guess; it is what the identifier means. # # Falls back to the full list when nothing matches exactly, because # near-misses are genuinely useful to a person deciding, and when # several match exactly, because two clients sharing a file number # is a data problem for a human rather than a choice for this code. if exact := _exactly_numbered(matches, term): return exact return matches def _resolve( self, file_number: str, cnic: str, by_file: list[dict[str, Any]] | None, by_cnic: list[dict[str, Any]] | None, ) -> dict[str, Any]: """Turn one or two searches into a client, or into a question.""" both = by_file is not None and by_cnic is not None agreed: bool | None = None if both: file_ids = {c.get("id") for c in by_file} cnic_ids = {c.get("id") for c in by_cnic} if file_ids and cnic_ids: agreed = bool(file_ids & cnic_ids) # One client, named by both. The strongest answer this workflow can # produce, and the only case where a second identifier makes the # result more certain rather than merely redundant. if len(file_ids) == 1 and file_ids == cnic_ids: return self._answer(by_file, FILE_NUMBER_TYPE, file_number, both=True, agreed=True) # Two identifiers, two different people. if file_ids and cnic_ids and not (file_ids & cnic_ids): merged = by_file + [c for c in by_cnic if c.get("id") not in file_ids] return { "matches": merged, "unambiguous": False, "identifier": { "type": FILE_NUMBER_TYPE, "value": file_number, "both_supplied": True, "agreed": False, }, "conflict": True, } # One of them found something. The file number is preferred wherever it # produced a result at all. if by_file: return self._answer(by_file, FILE_NUMBER_TYPE, file_number, both=both, agreed=agreed) if by_cnic: return self._answer(by_cnic, CNIC_TYPE, cnic, both=both, agreed=agreed) # Nothing matched. A real answer, carried to the reviewer rather than # failing the run: the document is on disk, the caption is on screen, # and a person can find the client in seconds. return { "matches": [], "unambiguous": False, "identifier": { "type": FILE_NUMBER_TYPE if file_number else CNIC_TYPE, "value": file_number or cnic, "both_supplied": both, "agreed": agreed, }, "conflict": False, } def _by_mobile(self, mobile: str) -> dict[str, Any]: """The last resort, and the weakest by a long way. Reached only when no exact identifier was supplied or none matched. A phone number is not an identity: phones are shared within families, an accountant forwards on a client's behalf, and a number changes hands. So the result is *never* treated as unambiguous however many rows come back — a single match here still goes to a reviewer to confirm. That is the difference between this tier and the two above it. A file number matching one client is an answer; a mobile matching one client is a suggestion. """ try: matches = self._search(mobile, "mobile") except AgentApiError: # Swallowed rather than failed: the exact lookups already ran and # already found nothing, and losing their answer to a failure on the # weakest tier would turn "no client matched" into "the run broke". matches = [] return { "matches": matches, "unambiguous": False, "identifier": { "type": "mobile", "value": mobile, "both_supplied": False, "agreed": None, }, "conflict": False, "weak": True, } @staticmethod def _answer( matches: list[dict[str, Any]], identifier_type: str, value: str, *, both: bool, agreed: bool | None, ) -> dict[str, Any]: return { "matches": matches, # Stated so the workflow branches rather than assuming the first hit # is right. Two clients sharing a file number is a data problem in # the CMS, not a reason to pick one of them. "unambiguous": len(matches) == 1, "identifier": { "type": identifier_type, "value": value, "both_supplied": both, "agreed": agreed, }, "conflict": False, } class GetClientTool(Tool): """Everything the agent may know about one client.""" name = "get_client" description = "Fetch a single client's details and record counts by id." policy = ExecutionPolicy.AUTOMATIC def __init__(self, cms: CmsClient) -> None: self._cms = cms def run(self, **kwargs: Any) -> ToolResult: client_id = kwargs.get("client_id") if not isinstance(client_id, int) or client_id <= 0: return ToolResult.failure("A positive integer client_id is required.") try: return ToolResult.success({"client": self._cms.get_client(client_id)}) except NotFound: # The CMS answers 404 for both "absent" and "not visible to you", # deliberately. Repeating that distinction here would invent # information the API refused to give. return ToolResult.failure(f"No client {client_id} is visible to this agent.") except AgentApiError as exc: return ToolResult.failure(str(exc))