"""Intake-gate adapter and its process-wide instance. Pure module: no FastAPI imports and no HTTPException. Owns construction and lifecycle only — get_classifier() / close_classifier() — mirroring agent/agent_setup.py. The prompt lives in prompt.py, the verdict shape in models.py, the run entrypoint in execute_agent.py. responses.parse rather than a hand-built JSON schema, for the reason app/services/llm.py:1-10 gives: parse derives a conforming schema and validates the reply back into the Pydantic model, so extra="forbid" still gates every verdict. Not llm_setup.llm_call(json_mode=True), which is schema-free — a gate that decides whether a row exists at all needs a validated bool, and needs the delivery-status branch below to tell "the model said no" from "the model could not answer". """ from __future__ import annotations import logging from app.core.config import supports_reasoning from app.core.errors import ModelRefusedError, ModelResponseInvalidError, ModelUnavailableError from openai import AsyncOpenAI from inbox_classifier.models import EmailTriageVerdict from inbox_classifier.plugins import PROMPT_CACHE_KEY, get_triage_settings from inbox_classifier.prompt import SYSTEM_PROMPT, build_input logger=logging.getLogger("inbox.triage") # Reasons the provider can return on an incomplete response. _TRUNCATED="max_output_tokens" _FILTERED="content_filter" def _first_refusal(response): """Return the refusal text if the model declined, else None. A refusal arrives as a content part inside an output message, not as an error, so it has to be walked for explicitly before the parsed output is trusted. Duplicated from app/services/llm.py:42-53 rather than imported: it is private there, and each domain owning its own copy is the same call inbox/plugins.py:384-386 already makes. """ for item in getattr(response,"output",None) or []: for part in getattr(item,"content",None) or []: if getattr(part,"type",None)=="refusal": refusal=getattr(part,"refusal",None) return str(refusal) if refusal else "refused" return None class EmailClassifier: def __init__(self, client:AsyncOpenAI, model, max_output_tokens, effort, enable_cache=True): self._client=client self._model=model self._max_output_tokens=max_output_tokens self._effort=effort self._enable_cache=enable_cache self._supports_reasoning=supports_reasoning(model) @property def model(self) -> str: return self._model async def classify(self, subject, body) -> EmailTriageVerdict: kwargs={ "model":self._model, "instructions":SYSTEM_PROMPT, "input":build_input(subject,body), "text_format":EmailTriageVerdict, "max_output_tokens":self._max_output_tokens, } # No temperature and no top_p: reasoning models reject them, and sampling was # never the right lever for a classification task. if self._supports_reasoning: kwargs["reasoning"]={"effort":self._effort} if self._enable_cache: kwargs["prompt_cache_key"]=PROMPT_CACHE_KEY response=await self._client.responses.parse(**kwargs) status=getattr(response,"status",None) self._log_usage(response,status) # Branch on delivery status before trusting any output. if status=="failed": raise ModelUnavailableError("provider reported a failed response") if status=="incomplete": reason=getattr(getattr(response,"incomplete_details",None),"reason",None) if reason==_FILTERED: raise ModelRefusedError("content filter blocked the response") if reason==_TRUNCATED: raise ModelResponseInvalidError("response truncated at max_output_tokens") raise ModelResponseInvalidError(f"incomplete response: {reason}") if _first_refusal(response) is not None: raise ModelRefusedError("model declined to classify this email") parsed=getattr(response,"output_parsed",None) if not isinstance(parsed,EmailTriageVerdict): raise ModelResponseInvalidError("response did not parse into EmailTriageVerdict") return parsed def _log_usage(self, response, status): """Token and cache visibility. %-args, not extra={}: main.py:21 configures format="%(levelname)-8s %(name)s: %(message)s", which renders no extra keys — the ATS adapter's structured fields are invisible in this process today. """ usage=getattr(response,"usage",None) input_details=getattr(usage,"input_tokens_details",None) output_details=getattr(usage,"output_tokens_details",None) logger.info( "triage upstream: model=%s status=%s request_id=%s in=%s out=%s cached=%s reasoning=%s", self._model, status, getattr(response,"id",None), getattr(usage,"input_tokens",None), getattr(usage,"output_tokens",None), getattr(input_details,"cached_tokens",None), getattr(output_details,"reasoning_tokens",None), ) _classifier=None def get_classifier() -> EmailClassifier: """Process-wide classifier over llm_setup's shared AsyncOpenAI client. Lazy so a missing OPENAI configuration surfaces on the first /email/fetch, not at import; llm_setup.init_llm() in the app lifespan has normally created and verified the client already. Mirrors job/candidate/plugins.get_scorer(). """ global _classifier if _classifier is None: from llm_setup import get_client settings=get_triage_settings() _classifier=EmailClassifier( get_client(), model=settings.openai_model, max_output_tokens=settings.openai_max_output_tokens, effort=settings.openai_effort, enable_cache=settings.openai_enable_prompt_cache, ) return _classifier def close_classifier(): """Drop the cached instance. Hooked into main.py's lifespan beside close_llm(), which disposes the shared client — a retained reference would otherwise point at a closed pool on an in-process restart. """ global _classifier _classifier=None logger.info("classifier closed")