HR-ATS-Portal/backend/inbox_classifier/agent_setup.py

"""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")