From 5ca085a238f997610254ff37172bb18305f4fc1c Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Tue, 18 Aug 2026 12:05:42 +0500 Subject: [PATCH] . --- backend/.env.example | 15 ++ backend/inbox/app.py | 55 ++++- backend/inbox/models.py | 187 +++++++++++++++++ backend/inbox/serializers.py | 31 ++- backend/inbox/views.py | 245 ++++++++++++++++++++-- backend/inbox_classifier/agent_setup.py | 162 ++++++++++++++ backend/inbox_classifier/decorators.py | 200 ++++++++++++++++++ backend/inbox_classifier/enums.py | 71 +++++++ backend/inbox_classifier/execute_agent.py | 56 +++++ backend/inbox_classifier/models.py | 43 ++++ backend/inbox_classifier/plugins.py | 107 ++++++++++ backend/inbox_classifier/prompt.py | 95 +++++++++ backend/main.py | 7 + backend/tests/conftest.py | 40 ++++ pyproject.toml | 5 +- 15 files changed, 1292 insertions(+), 27 deletions(-) create mode 100644 backend/inbox_classifier/agent_setup.py create mode 100644 backend/inbox_classifier/decorators.py create mode 100644 backend/inbox_classifier/enums.py create mode 100644 backend/inbox_classifier/execute_agent.py create mode 100644 backend/inbox_classifier/models.py create mode 100644 backend/inbox_classifier/plugins.py create mode 100644 backend/inbox_classifier/prompt.py create mode 100644 backend/tests/conftest.py diff --git a/backend/.env.example b/backend/.env.example index e47fb09..20041c7 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -54,6 +54,21 @@ MAX_PDF_SIZE_MB=10 MAX_JD_CHARS=30000 MAX_RESUME_CHARS=60000 +# Inbox intake gate (inbox_classifier/): only mail judged to be a job application +# gets an inbox_messages row; every verdict is logged to inbox_message_triage. +# Model / token / effort / cache knobs are the OPENAI_* ones above. +# false restores the pre-gate behaviour exactly — the rollback lever. +INBOX_TRIAGE_ENABLED=true +# true: a provider outage or missing key ingests the mail and marks the verdict +# unclassified. false: skip it and leave it for a later /email/fetch. +INBOX_TRIAGE_FAIL_OPEN=true +INBOX_TRIAGE_CONCURRENCY=5 +INBOX_TRIAGE_MAX_SUBJECT_CHARS=300 +INBOX_TRIAGE_MAX_BODY_CHARS=4000 +# 0 disables the uncertainty branch; >0 routes low-confidence verdicts to the +# INBOX_TRIAGE_FAIL_OPEN policy. +INBOX_TRIAGE_MIN_CONFIDENCE=0 + REDIS_URL=redis://localhost:6379/0 TASKIQ_QUEUE_NAME=inbox TASKIQ_CV_QUEUE_NAME=cv_upload diff --git a/backend/inbox/app.py b/backend/inbox/app.py index ae1815b..e4953a0 100644 --- a/backend/inbox/app.py +++ b/backend/inbox/app.py @@ -25,6 +25,10 @@ class DuplicateBody(BaseModel): is_duplicate: bool +class TriageOverrideBody(BaseModel): + is_application: bool + + class EmailSendBody(BaseModel): to: str subject: str @@ -52,21 +56,28 @@ async def fetch_email( data=await service.service_email(top,skip) value=data.get("value") items_lst=[] + # Classify the whole page first, bounded-parallel, then replay it in upstream + # order: the inserts stay serial on the one request session and pending_match_ids + # keeps the sequence it has today. + decisions=await service.triage_round([item.get("id") for item in value]) for item in value: message_id=item.get("id") - service_per_email=await service.get_email_by_id(message_id,test_on) + service_per_email=await service.get_email_by_id(message_id,test_on,decision=decisions.get(str(message_id))) items_lst.append({"message_id":message_id,"email_contents":service_per_email}) if service.pending_match_ids: await service.enqueue_matching(list(service.pending_match_ids),force=False) + skipped=len(service.skipped_message_ids) + triage={"ingested":len(items_lst)-skipped,"skipped":skipped,"errors":len(service.triage_errors)} + account_setup=[] if test_on: - return JSONResponse(content={"data":items_lst,"status_code":200}) + return JSONResponse(content={"data":items_lst,"triage":triage,"status_code":200}) if service.pending_confirmation_emails: account_setup=await service.send_account_setup(list(service.pending_confirmation_emails)) - return JSONResponse(content={"data":items_lst,"account_setup":account_setup,"status_code":200}) + return JSONResponse(content={"data":items_lst,"account_setup":account_setup,"triage":triage,"status_code":200}) except HTTPException: raise @@ -214,6 +225,44 @@ async def get_inbox_counts( raise HTTPException(status_code=500,detail=str(e)) +@router.get("/inbox/triage") +async def fetch_triage( + search: str | None = Query(None), + is_application: bool | None = Query(None), + status: str | None = Query(None), + top: int = Query(100), + skip: int = Query(0, ge=0), + current_user: dict = Depends(require_permission(PermissionTag.INBOX_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=Email(session=session) + data=await service.get_triage_messages(top,skip,search,is_application,status) + total=await service.count_triage(search,is_application,status) + return JSONResponse(content={"data":data,"total":total,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + +@router.patch("/inbox/triage/{record_id}/override") +async def override_triage( + record_id: str, + payload: TriageOverrideBody, + current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)), + session: AsyncSession = Depends(get_session), +): + try: + service=Email(session=session) + data=await service.override_triage(record_id,payload.is_application,current_user) + return JSONResponse(content={"data":data,"total":1,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + @router.patch("/inbox/{record_id}/processing-state") async def set_processing_state( record_id: str, diff --git a/backend/inbox/models.py b/backend/inbox/models.py index 6cd863a..44d2832 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -671,6 +671,32 @@ class Inbox_Messages(SQLModel, table=True): await session.commit() return result.rowcount or 0 + @classmethod + async def get_by_upstream_id(cls, session: AsyncSession, message_id): + """Lookup by the UPSTREAM Graph id, not the local PK. + + get_inbox_message_by_id above takes the uuid primary key; the triage ledger is + keyed on the upstream id, so overturning a verdict needs this direction. + """ + result=await session.execute(select(cls).where(cls.message_id == str(message_id))) + return result.scalars().first() + + @classmethod + async def existing_message_ids(cls, session: AsyncSession, message_ids) -> set: + """The subset of upstream ids already persisted — the free half of the gate. + + A message already in this table was judged an application once, so the intake + classifier must never be paid for a second time; insert_email's upsert still + refreshes the row. One query per fetch round, columns only. + """ + ids=[str(m) for m in message_ids or [] if m] + if not ids: + return set() + result=await session.execute( + select(cls.message_id).where(cls.message_id.in_(ids)) + ) + return {row for (row,) in result.all() if row} + @classmethod async def mark_message_read(cls, session: AsyncSession, record_id): row=await cls.get_inbox_message_by_id(session,record_id) @@ -729,6 +755,167 @@ class Inbox_Messages(SQLModel, table=True): return row +class Inbox_Message_Triage(SQLModel, table=True): + """One intake verdict per upstream message id — the gate before inbox_messages. + + Rows land here for BOTH outcomes. Rejections are the point: inbox_messages stays + application-only, and a repeated /email/fetch never re-pays for the same + classification. Acceptances are recorded too, so a round that classified and then + failed to insert does not pay twice either. + + Deliberately no message_body column: the body is what this feature keeps out of the + database, and the override route re-reads the mail from upstream by message_id. + message_subject is kept (capped in inbox_classifier.decorators.triage_fields) + because a review screen without it is unusable — it is stored, never logged. + + server_default is load-bearing on every NOT NULL column: alembic_setup runs with + compare_server_default=True, so a model default without a matching server default + autogenerates a drift revision on every boot. + """ + + __tablename__ = "inbox_message_triage" + + id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) + # Upstream Graph id — the same key insert_email upserts on. + message_id: str = Field(index=True, unique=True) + is_application: bool = Field(default=False, sa_column_kwargs={"server_default": "false"}) + reason_code: str = Field(default="", sa_column_kwargs={"server_default": ""}) + confidence: float | None = Field(default=None) + evidence: str = Field(default="", sa_column_kwargs={"server_default": ""}) + # Triage_Status: classified | low_confidence | error. Plain text, not a PG enum — + # alembic autogenerate cannot see new enum labels, and the enum in + # inbox_classifier/enums.py already gates what code writes here. + status: str = Field(default="classified", sa_column_kwargs={"server_default": "classified"}) + error: str | None = Field(default=None) + model_name: str = Field(default="", sa_column_kwargs={"server_default": ""}) + message_subject: str = Field(default="", sa_column_kwargs={"server_default": ""}) + message_from: str = Field(default="", sa_column_kwargs={"server_default": ""}) + message_received_time: str = Field(default="", sa_column_kwargs={"server_default": ""}) + file_name: str = Field(default="", sa_column_kwargs={"server_default": ""}) + attachment: bool = Field(default=False, sa_column_kwargs={"server_default": "false"}) + ingested: bool = Field(default=False, sa_column_kwargs={"server_default": "false"}) + overridden_by_id: uuid.UUID | None = Field(default=None, foreign_key="users.id") + overridden_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) + # _now(), never datetime.now(): a naive local value bound to a timestamptz column + # is read back as UTC and silently backdates the row (see AtsResults below). + classified_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + + @staticmethod + def _as_uuid(record_id) -> uuid.UUID | None: + if record_id in (None, ""): + return None + try: + return uuid.UUID(str(record_id)) + except ValueError: + return None + + @classmethod + async def get_by_message_id(cls, session: AsyncSession, message_id): + result=await session.execute(select(cls).where(cls.message_id == str(message_id))) + return result.scalars().first() + + @classmethod + async def get_triage_by_id(cls, session: AsyncSession, record_id): + rid=cls._as_uuid(record_id) + if rid is None: + return None + result=await session.execute(select(cls).where(cls.id == rid)) + return result.scalars().first() + + @classmethod + async def verdicts_for_message_ids(cls, session: AsyncSession, message_ids) -> dict: + """{upstream message_id: is_application} for a whole fetch round, one query.""" + ids=[str(m) for m in message_ids or [] if m] + if not ids: + return {} + result=await session.execute( + select(cls.message_id, cls.is_application).where(cls.message_id.in_(ids)) + ) + return {message_id: bool(is_application) for message_id, is_application in result.all()} + + @classmethod + async def record_verdict(cls, session: AsyncSession, fields: dict): + """Upsert one verdict on message_id. + + Two fetch rounds can race the unique index, so IntegrityError rolls back and + re-reads rather than failing the round — same shape as _link_sender above. + """ + message_id=str(fields.get("message_id") or "") + if not message_id: + return None + existing=await cls.get_by_message_id(session, message_id) + if existing: + for key, value in fields.items(): + setattr(existing, key, value) + existing.classified_at=_now() + session.add(existing) + await session.commit() + await session.refresh(existing) + return existing + row=cls(**fields) + session.add(row) + try: + await session.commit() + except IntegrityError: + await session.rollback() + return await cls.get_by_message_id(session, message_id) + await session.refresh(row) + return row + + @classmethod + def _triage_filter(cls, statement, is_application, status, search): + if is_application is not None: + statement=statement.where(cls.is_application == bool(is_application)) + if status: + statement=statement.where(cls.status == str(status)) + if search: + pattern=f"%{search}%" + statement=statement.where( + or_(cls.message_subject.ilike(pattern), cls.message_from.ilike(pattern)) + ) + return statement + + @classmethod + async def list_triage(cls, session: AsyncSession, top, skip, is_application=None, + status=None, search=None): + statement=cls._triage_filter(select(cls), is_application, status, search) + statement=statement.order_by(cls.classified_at.desc()).offset(skip).limit(top) + result=await session.execute(statement) + return list(result.scalars().all()) + + @classmethod + async def count_triage(cls, session: AsyncSession, is_application=None, status=None, + search=None) -> int: + statement=cls._triage_filter(select(func.count(cls.id)), is_application, status, search) + result=await session.execute(statement) + return int(result.scalar() or 0) + + @classmethod + async def set_override(cls, session: AsyncSession, record_id, is_application, user_id=None): + row=await cls.get_triage_by_id(session, record_id) + if not row: + return None + row.is_application=bool(is_application) + row.overridden_by_id=cls._as_uuid(user_id) + row.overridden_at=_now() + session.add(row) + await session.commit() + await session.refresh(row) + return row + + @classmethod + async def mark_ingested(cls, session: AsyncSession, message_id, ingested: bool = True): + row=await cls.get_by_message_id(session, message_id) + if not row: + return None + row.ingested=bool(ingested) + session.add(row) + await session.commit() + await session.refresh(row) + return row + + class SourceChannels(SQLModel, table=True): __tablename__ = "source_channels" diff --git a/backend/inbox/serializers.py b/backend/inbox/serializers.py index 0bc19ee..6d368df 100644 --- a/backend/inbox/serializers.py +++ b/backend/inbox/serializers.py @@ -1,6 +1,6 @@ from pathlib import Path -from inbox.models import Inbox_Messages +from inbox.models import Inbox_Message_Triage, Inbox_Messages # match_status (inbox/tasks.py) -> the resume badge the inbox tabs render. _RESUME_STATUS = { @@ -128,3 +128,32 @@ def serialize_application(message: Inbox_Messages) -> dict: "processing_state": message.processing_state, "source_channel_id": message.source_channel_id, } + + +def serialize_triage(row: Inbox_Message_Triage) -> dict: + """inbox_message_triage row -> the intake gate's review shape. + + No body field exists to expose: the gate stores the verdict, never the mail. A + reviewer opens the original from the mailbox, or overturns the verdict and lets the + normal ingestion path re-fetch it. + """ + return { + "id": str(row.id), + "message_id": row.message_id, + "is_application": row.is_application, + "reason_code": row.reason_code, + "confidence": row.confidence, + "evidence": row.evidence, + "status": row.status, + "error": row.error, + "model_name": row.model_name or None, + "subject": row.message_subject, + "fromEmail": row.message_from, + "when": row.message_received_time, + "attachment": row.file_name or None, + "has_attachment": row.attachment, + "ingested": row.ingested, + "overridden_by": str(row.overridden_by_id) if row.overridden_by_id else None, + "overridden_at": row.overridden_at.isoformat() if row.overridden_at else None, + "classified_at": row.classified_at.isoformat() if row.classified_at else None, + } diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 712a529..822ba6d 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -1,11 +1,12 @@ +import asyncio import logging import uuid import httpx,os from fastapi import HTTPException from inbox.enums import Candidate_application_Status -from inbox.models import Inbox_Messages +from inbox.models import Inbox_Messages,Inbox_Message_Triage from inbox.file_decoder import decode_attachment -from inbox.serializers import serialize_application, serialize_message +from inbox.serializers import serialize_application, serialize_message, serialize_triage from inbox.plugins import ( EMAIL_API_TOKEN, fetch_message_read_status, @@ -13,12 +14,23 @@ from inbox.plugins import ( request_email_confirmation, send_mail, ) +from inbox_classifier.decorators import is_manual_upload,triage_fields +from inbox_classifier.execute_agent import classify_email +from inbox_classifier.plugins import ( + TRIAGE_CONCURRENCY, + TRIAGE_ENABLED, + TRIAGE_STATUSES, + sender_domain, + should_ingest, + triage_model_name, +) from dotenv import load_dotenv load_dotenv() from sqlalchemy.ext.asyncio import AsyncSession from datetime import datetime,timezone logger=logging.getLogger("inbox.match") +triage_logger=logging.getLogger("inbox.triage") class Email: @@ -28,6 +40,10 @@ class Email: self.token=token or EMAIL_API_TOKEN self.pending_match_ids:list[str]=[] self.pending_confirmation_emails:list[str]=[] + # Upstream ids the intake gate judged not to be job applications. They get a + # verdict row and no inbox_messages row. + self.skipped_message_ids:list[str]=[] + self.triage_errors:list[str]=[] # async def get_all_applications(self,app_id=None): # try: @@ -53,27 +69,163 @@ class Email: except Exception as e: raise HTTPException(status_code=500,detail=str(e)) - async def get_email_by_id(self,message_id,test_on=True): + async def fetch_message(self,message_id): + """GET /emails/{id} on the upstream Email API -> the Graph payload.""" async with httpx.AsyncClient() as client: - try: - response=await client.get(f"{self.get_url}/emails/{message_id}", - headers={"Authorization":f"Bearer {self.token}"} - ) - if response.status_code==200: - data=response.json() - re_create_file=await decode_attachment(data.get("attachments")) - row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) - if row.attachment and row.file_path and row.match_status is None: - self.pending_match_ids.append(str(row.id)) - if test_on: - return data - if new_user_email: - self.pending_confirmation_emails.append(new_user_email) - return data - else: - raise HTTPException(status_code=response.status_code,detail=response.text) - except Exception as e: - raise HTTPException(status_code=500,detail=str(e)) + response=await client.get(f"{self.get_url}/emails/{message_id}", + headers={"Authorization":f"Bearer {self.token}"} + ) + if response.status_code!=200: + raise HTTPException(status_code=response.status_code,detail=response.text) + return response.json() + + async def triage_round(self,message_ids): + """Fetch and classify a whole /email/fetch page, bounded by a semaphore. + + Returns {message_id: decision}. The caller replays the page in upstream order, + so pending_match_ids and pending_confirmation_emails keep the exact sequence + they have today. + + Only the upstream GET and the OpenAI call run concurrently, and nothing inside + the gather touches self.session — Depends(get_session) yields ONE AsyncSession, + which cannot be shared across tasks. All DB work stays in the serial replay. + + Two pre-filters run first and cost no tokens: a message already in + inbox_messages was judged an application once, and a message already in + inbox_message_triage has a stored verdict to replay. That is what makes a + repeated fetch free. + """ + ids=[str(m) for m in message_ids or [] if m] + decisions={} + if not ids: + return decisions + known=await Inbox_Messages.existing_message_ids(self.session,ids) + recorded=await Inbox_Message_Triage.verdicts_for_message_ids(self.session,ids) + pending=[] + for message_id in ids: + if message_id in known: + decisions[message_id]={"ingest":True,"status":"known","fresh":False} + elif message_id in recorded: + decisions[message_id]={"ingest":recorded[message_id],"status":"recorded","fresh":False} + else: + pending.append(message_id) + if not pending: + triage_logger.info("triage round: page=%s known=%s classified=0",len(ids),len(decisions)) + return decisions + + semaphore=asyncio.Semaphore(TRIAGE_CONCURRENCY) + + async def run(message_id): + async with semaphore: + data=await self.fetch_message(message_id) + if not TRIAGE_ENABLED or is_manual_upload(data): + return message_id,{"data":data,"ingest":True,"status":"disabled","fresh":False} + verdict,error=await classify_email(data) + ingest,status,reason=should_ingest(verdict,error) + return message_id,{"data":data,"verdict":verdict,"error":error,"ingest":ingest, + "status":status,"reason":reason,"fresh":True} + + results=await asyncio.gather(*(run(m) for m in pending),return_exceptions=True) + accepted=rejected=errors=0 + for result in results: + if isinstance(result,BaseException): + # First failure wins, preserving today's all-or-nothing behaviour for a + # failing upstream message. Never catch BaseException itself: a + # CancelledError must keep propagating. + raise result + message_id,decision=result + decisions[message_id]=decision + if decision.get("status")=="error": + errors+=1 + if decision.get("ingest"): + accepted+=1 + else: + rejected+=1 + triage_logger.info( + "triage round: page=%s known=%s classified=%s accepted=%s rejected=%s errors=%s", + len(ids),len(known)+len(recorded),len(pending),accepted,rejected,errors, + ) + return decisions + + async def record_triage(self,data,decision,ingested): + """Persist one verdict. Never raises into the ingestion path. + + A failed audit write must not cost us a candidate: the worst case is that the + next fetch re-classifies this message. + """ + try: + fields=triage_fields( + data, + decision.get("verdict"), + decision.get("status") or "classified", + decision.get("reason") or "", + error=decision.get("error") or "", + model_name=triage_model_name(), + ingested=ingested, + ) + await Inbox_Message_Triage.record_verdict(self.session,fields) + # Allowlisted keys only: sender DOMAIN not address, attachment COUNT not + # names, no subject or body text, no evidence text. + verdict=decision.get("verdict") + triage_logger.info( + "triage %s: application=%s reason=%s confidence=%s domain=%s attachments=%s", + fields["message_id"], + fields["is_application"], + fields["reason_code"], + getattr(verdict,"confidence",None), + sender_domain(fields["message_from"]), + len(data.get("attachments") or []), + ) + except Exception as e: + triage_logger.warning("triage record failed: %s",type(e).__name__) + + async def get_email_by_id(self,message_id,test_on=True,decision=None): + """Persist one upstream message, gated by the application classifier. + + `decision` is the pre-computed verdict from triage_round; without one this + classifies inline, so a single-message call still works. + + Ordering is deliberate. The verdict comes BEFORE decode_attachment: a rejected + mail must not write a file into decoded_attachments (nothing on this path ever + deletes one, and _write uses the basename only, so a vendor "resume.pdf" would + clobber a candidate's stored CV), and must not reach _link_sender, which would + create a candidate Users row and queue a confirmation mail for a stranger. + + The gate lives here, not in Inbox_Messages.insert_email, so + FileRead.ingest_upload bypasses it for free — that path fabricates an EMPTY body + and would be a guaranteed false negative under a subject+body classifier. + """ + try: + if decision is None: + decision=(await self.triage_round([message_id])).get(str(message_id)) or {} + data=decision.get("data") or await self.fetch_message(message_id) + + if not decision.get("ingest"): + if decision.get("fresh"): + await self.record_triage(data,decision,ingested=False) + self.skipped_message_ids.append(str(message_id)) + if decision.get("status")=="error": + self.triage_errors.append(str(message_id)) + return {"message_id":str(message_id),"skipped":"not_application", + "reason":decision.get("reason") or "","status":decision.get("status") or ""} + + re_create_file=await decode_attachment(data.get("attachments")) + row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) + if decision.get("fresh"): + await self.record_triage(data,decision,ingested=True) + if row.attachment and row.file_path and row.match_status is None: + self.pending_match_ids.append(str(row.id)) + if test_on: + return data + if new_user_email: + self.pending_confirmation_emails.append(new_user_email) + return data + except HTTPException: + # Was missing: the bare `except Exception` below caught the upstream-status + # HTTPException and re-raised every one of them as a 500. + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) async def get_inbox_messages(self,top,skip,search=None): messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search) @@ -241,6 +393,55 @@ class Email: raise HTTPException(status_code=404,detail="Message not found") return serialize_application(message) + async def get_triage_messages(self,top,skip,search=None,is_application=None,status=None): + """The intake gate's verdict log — mostly the mail that never became a row. + + A hard gate's only real risk is the silent false negative, so the rejections + have to be reviewable. + """ + if status is not None and status not in TRIAGE_STATUSES: + raise HTTPException(status_code=422,detail=f"status must be one of {', '.join(TRIAGE_STATUSES)}") + rows=await Inbox_Message_Triage.list_triage(self.session,top,skip,is_application,status,search) + return [serialize_triage(row) for row in rows] + + async def count_triage(self,search=None,is_application=None,status=None): + return await Inbox_Message_Triage.count_triage(self.session,is_application,status,search) + + async def override_triage(self,record_id,is_application,current_user=None): + """Overturn a verdict a recruiter disagrees with. + + false -> true re-fetches the mail from upstream and runs the normal ingestion + path, which is why the body was never stored. + + true -> false does NOT delete the inbox_messages row: inbox, ats_results, + assessments, notifications and application_stage_transitions all reference it, + so a purge would take candidate accounts and scores with it. It moves the row to + processing_state 'rejected' instead, an already-allowlisted value. + """ + if not isinstance(is_application,bool): + raise HTTPException(status_code=422,detail="is_application must be a boolean") + row=await Inbox_Message_Triage.get_triage_by_id(self.session,record_id) + if not row: + raise HTTPException(status_code=404,detail="Triage record not found") + + user_id=(current_user or {}).get("id") + if is_application and not row.ingested: + data=await self.fetch_message(row.message_id) + re_create_file=await decode_attachment(data.get("attachments")) + message,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) + await Inbox_Message_Triage.mark_ingested(self.session,row.message_id,True) + if message.attachment and message.file_path and message.match_status is None: + await self.enqueue_matching([str(message.id)],force=False) + if new_user_email: + await self.send_account_setup([new_user_email]) + elif not is_application and row.ingested: + message=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id) + if message: + await Inbox_Messages.set_processing_state(self.session,message.id,"rejected") + + updated=await Inbox_Message_Triage.set_override(self.session,record_id,is_application,user_id) + return serialize_triage(updated) + async def send_email(self,payload,current_user): to_email=(payload.get("to") or "").strip() subject=(payload.get("subject") or "").strip() diff --git a/backend/inbox_classifier/agent_setup.py b/backend/inbox_classifier/agent_setup.py new file mode 100644 index 0000000..a19f2d1 --- /dev/null +++ b/backend/inbox_classifier/agent_setup.py @@ -0,0 +1,162 @@ +"""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") diff --git a/backend/inbox_classifier/decorators.py b/backend/inbox_classifier/decorators.py new file mode 100644 index 0000000..8256b2b --- /dev/null +++ b/backend/inbox_classifier/decorators.py @@ -0,0 +1,200 @@ +"""HTML reduction, signal extraction, and triage column builders. + +Pure module: no FastAPI imports and no HTTPException. Plain functions despite the file +name, following agent/decorators.py. + +Stdlib only (html.parser + re). requirements.txt is deliberately untouched: a +dependency on an HTML library for one classifier prompt is not worth the pin. +""" + +from __future__ import annotations + +import re +from html.parser import HTMLParser + +from inbox_classifier.enums import Block_Tags, Drop_Tags + +_TAG=re.compile(r"<[^>]+>") +# \xa0 is listed explicitly:   unescapes to a NO-BREAK SPACE, which a plain \s +# collapse does not match, so an HTML mail would otherwise reach the prompt full of +# stray non-breaking spaces. Written as an escape, not the literal character, so it +# stays visible in a diff. +_SPACES=re.compile(r"[ \t\xa0\r\f\v]+") +_BLANK_LINES=re.compile(r"\n{3,}") + +# Quoted-history markers, in the order Outlook and Gmail actually emit them. +_QUOTE_MARKERS=( + re.compile(r"^-{2,}\s*original message\s*-{2,}", re.IGNORECASE | re.MULTILINE), + re.compile(r"^-{2,}\s*forwarded message\s*-{2,}", re.IGNORECASE | re.MULTILINE), + re.compile(r"^\s*on .{0,200}? wrote:\s*$", re.IGNORECASE | re.MULTILINE), + re.compile(r"^\s*from:\s.+$", re.IGNORECASE | re.MULTILINE), + re.compile(r"^\s*>", re.MULTILINE), +) + +# Below this many characters of new text, a "quoted" reply is really a bare forward +# with nothing above the line. Load-bearing: the prompt says to judge the quoted text +# in exactly that case, so it must not be trimmed away. +_MIN_NEW_TEXT=40 + +MANUAL_UPLOAD_PREFIX="manual-cv:" + + +class _TextExtractor(HTMLParser): + """Visible text only, block tags collapsed to newlines. + + convert_charrefs (default True) means handle_data already receives unescaped text, + so & /   / ' never reach the prompt as entities. handle_startendtag + dispatches to start+end by default, so
needs no special case. + """ + + def __init__(self): + super().__init__(convert_charrefs=True) + self._parts=[] + self._suppress=0 + + def _break(self): + """One line break per boundary, however many tags meet there. + + `

` is a single break, not two: closing and opening tags both mark a + boundary, and emitting a newline for each would turn every paragraph gap into a + blank line. Genuine blank lines in the source survive as data parts. + """ + if self._parts and self._parts[-1]=="\n": + return + self._parts.append("\n") + + def handle_starttag(self, tag, attrs): + if Drop_Tags.has(tag): + self._suppress+=1 + elif Block_Tags.has(tag): + self._break() + + def handle_endtag(self, tag): + if Drop_Tags.has(tag): + self._suppress=max(self._suppress-1,0) + elif Block_Tags.has(tag): + self._break() + + def handle_data(self, data): + if not self._suppress: + self._parts.append(data) + + def text(self) -> str: + return "".join(self._parts) + + +def _tidy(text, limit=None) -> str: + """Collapse runs of whitespace without destroying meaningful line breaks.""" + text=text.replace("\x00","") + text=_SPACES.sub(" ",text) + text="\n".join(line.strip() for line in text.split("\n")) + text=_BLANK_LINES.sub("\n\n",text).strip() + if limit is not None and len(text)>limit: + text=text[:limit].rstrip()+"\n[truncated]" + return text + + +def html_to_text(value, limit=None) -> str: + """Graph body HTML -> plain text. Empty in, empty out. + + message_body is stored as raw Graph HTML (inbox/models.py:353-360) and there is no + other html-to-text helper in backend/, so the reduction happens here. + """ + if not value or not isinstance(value,str): + return "" + if "<" not in value: + # Already plain text (Graph sends contentType "text" for some senders). + return _tidy(value,limit) + parser=_TextExtractor() + try: + parser.feed(value) + parser.close() + text=parser.text() + except Exception: + # Malformed markup should degrade, never fail a whole fetch round. + text="" + if not text.strip(): + text=_TAG.sub(" ",value) + return _tidy(text,limit) + + +def strip_quoted_reply(text) -> str: + """Trim at the first quoted-history marker, keeping only the newest message. + + Only trims when at least _MIN_NEW_TEXT characters precede the marker: a bare + forward whose new text is empty must reach the model whole. + """ + if not text: + return "" + cut=len(text) + for marker in _QUOTE_MARKERS: + match=marker.search(text) + if match is not None and match.start()=len(text): + return text + head=text[:cut].strip() + return head if len(head)>=_MIN_NEW_TEXT else text + + +def _raw_body(email_data) -> str: + """body dict -> body str -> bodyPreview, mirroring Inbox_Messages._body_text. + + The bodyPreview fallback matters: an image-only or malformed mail still carries its + preview line, which is often the only signal available. + """ + body=email_data.get("body") + if isinstance(body,dict): + return body.get("content") or "" + if isinstance(body,str): + return body + return email_data.get("bodyPreview") or "" + + +def email_signals(email_data, subject_limit, body_limit) -> tuple[str,str]: + """(subject, body_text) for the prompt. Subject and body only, by design.""" + subject=_tidy(str(email_data.get("subject") or ""),subject_limit) + body=html_to_text(_raw_body(email_data)) + body=_tidy(strip_quoted_reply(body),body_limit) + return subject,body + + +def is_manual_upload(email_data) -> bool: + """Recruiter CV upload (id "manual-cv:...") — an application by construction. + + Defence in depth: the gate lives in inbox.views.Email.get_email_by_id, which + FileRead.ingest_upload never calls, so the manual path already bypasses it. This + keeps the invariant testable and stops a future caller from re-introducing the + empty-body false negative (that path always sends body content ""). + """ + return str(email_data.get("id") or "").startswith(MANUAL_UPLOAD_PREFIX) + + +def triage_fields(email_data, verdict, status, reason_code, error="", model_name="", + ingested=False) -> dict: + """The inbox_message_triage column dict. + + No body key, ever: the body is what this feature keeps out of the database, and the + override route re-reads the mail from upstream by message_id. The subject is kept + (capped) because a review screen without it is unusable. + """ + attachments=email_data.get("attachments") or [] + file_names=",".join(str(a.get("name") or "") for a in attachments if a.get("name")) + return { + "message_id":str(email_data.get("id") or ""), + "is_application":bool(getattr(verdict,"is_application",False)), + "reason_code":str(reason_code or "")[:60], + "confidence":getattr(verdict,"confidence",None), + "evidence":(getattr(verdict,"evidence","") or "")[:200], + "status":str(status or "classified")[:30], + "error":(error or None), + "model_name":str(model_name or "")[:120], + "message_subject":str(email_data.get("subject") or "")[:300], + "message_from":( + email_data.get("from",{}).get("emailAddress",{}).get("address","") or "" + )[:320], + "message_received_time":str(email_data.get("receivedDateTime") or "")[:64], + "file_name":file_names[:1000], + "attachment":bool(email_data.get("hasAttachments")), + "ingested":bool(ingested), + } diff --git a/backend/inbox_classifier/enums.py b/backend/inbox_classifier/enums.py new file mode 100644 index 0000000..42d4a1c --- /dev/null +++ b/backend/inbox_classifier/enums.py @@ -0,0 +1,71 @@ +from enum import Enum + +# (str, Enum) like inbox/enums.py: the mixin keeps every member comparable to and +# usable as a plain string, which is what HTMLParser hands us and what the triage +# columns store. + + +class Block_Tags(str, Enum): + """Tags that imply a line break in the rendered mail.""" + + BR="br" + P="p" + DIV="div" + LI="li" + TR="tr" + TABLE="table" + BLOCKQUOTE="blockquote" + SECTION="section" + ARTICLE="article" + HR="hr" + H1="h1" + H2="h2" + H3="h3" + H4="h4" + H5="h5" + H6="h6" + + @classmethod + def has(cls, tag) -> bool: + # _value2member_map_ keeps this O(1) with no exception overhead. `tag in cls` + # would do the same on 3.12+ but raises TypeError on 3.11, and pyproject + # still allows 3.11. + return tag in cls._value2member_map_ + + +class Drop_Tags(str, Enum): + """Tags whose content is markup machinery, not readable text.""" + + SCRIPT="script" + STYLE="style" + HEAD="head" + TITLE="title" + META="meta" + LINK="link" + + @classmethod + def has(cls, tag) -> bool: + return tag in cls._value2member_map_ + + +class Triage_Reason_Code(str, Enum): + """Why the gate decided what it decided. + + Sent to the model as the schema's enum for `reason_code`, so these labels are + part of the prompt contract — renaming one changes model behaviour. + """ + + JOB_APPLICATION="job_application" + RECRUITER_OR_VENDOR="recruiter_or_vendor" + NEWSLETTER_OR_MARKETING="newsletter_or_marketing" + INTERNAL_OR_SCHEDULING="internal_or_scheduling" + AUTOMATED_NOTIFICATION="automated_notification" + OTHER="other" + + +class Triage_Status(str, Enum): + """How the verdict was reached, as stored on inbox_message_triage.status.""" + + CLASSIFIED="classified" + LOW_CONFIDENCE="low_confidence" + ERROR="error" diff --git a/backend/inbox_classifier/execute_agent.py b/backend/inbox_classifier/execute_agent.py new file mode 100644 index 0000000..a473588 --- /dev/null +++ b/backend/inbox_classifier/execute_agent.py @@ -0,0 +1,56 @@ +"""Intake-gate entrypoint — one Responses call per email. + +Pure module: no FastAPI imports and no HTTPException. +Called from inbox.views.Email; no HTTP surface of its own. + +Returns (verdict, error_code) and never raises, mirroring +inbox/plugins.extract_resume_text's (text, error) shape. A provider outage must be a +policy decision at the call site (INBOX_TRIAGE_FAIL_OPEN in plugins.should_ingest), not +a 500 on /email/fetch. +""" + +from __future__ import annotations + +import logging + +from app.core.errors import ATSError, classify_error + +from inbox_classifier.agent_setup import get_classifier +from inbox_classifier.decorators import email_signals +from inbox_classifier.plugins import TRIAGE_MAX_BODY_CHARS, TRIAGE_MAX_SUBJECT_CHARS + +logger=logging.getLogger("inbox.triage") + +# No subject and no body: there is nothing to judge, so this is unclassifiable rather +# than a "no". It routes through the fail policy, which under the default fail-open +# means the mail is ingested — a signal-free message is never silently dropped. +EMPTY_MESSAGE="empty_message" + + +async def classify_email(email_data) -> tuple: + """Judge one email from its subject and body. Never raises. + + (verdict, "") on success; (None, error_code) when the model could not be consulted + or returned something unusable. + """ + subject,body=email_signals(email_data,TRIAGE_MAX_SUBJECT_CHARS,TRIAGE_MAX_BODY_CHARS) + if not subject and not body: + return None,EMPTY_MESSAGE + + try: + # get_classifier() is inside the try on purpose: a missing OPENAI_API_KEY raises + # RuntimeError from llm_setup.get_client(), and a stale OPENAI_MODEL raises + # pydantic ValidationError from Settings. Both belong on the fail policy, not on + # a 500 for the whole fetch round. + classifier=get_classifier() + verdict=await classifier.classify(subject,body) + return verdict,"" + except ATSError as e: + logger.warning("triage failed: code=%s",e.error_code) + return None,e.error_code + except Exception as e: + # classify_error never returns provider text. Log the exception TYPE and the code + # only — never the message, which can carry prompt or body content. + code,_=classify_error(e) + logger.warning("triage failed: code=%s exc=%s",code,type(e).__name__) + return None,code diff --git a/backend/inbox_classifier/models.py b/backend/inbox_classifier/models.py new file mode 100644 index 0000000..be54a0a --- /dev/null +++ b/backend/inbox_classifier/models.py @@ -0,0 +1,43 @@ +"""The triage verdict exchanged with the intake-gate model. + +Pure module: no FastAPI imports and no HTTPException. + +``extra="forbid"`` is load-bearing — it emits ``additionalProperties: false``, which +the structured-outputs schema dialect requires (same reason as +app/models/scoring.py:22-23). Every field is required: structured outputs puts all +declared properties in ``required``, so a defaulted field buys nothing here. + +Mirrors agent/models.py — this package's models.py holds the shape the LLM pass +exchanges, not a SQLModel table. The triage TABLE lives in inbox/models.py beside +Inbox_Messages, because it is an inbox-domain fact and this package never opens a +session. +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + +from inbox_classifier.enums import Triage_Reason_Code + + +class EmailTriageVerdict(BaseModel): + """One intake decision about one email. + + ``confidence`` carries the model's doubt so the boolean does not have to. The + prompt tells it to answer the boolean the way a recruiter would want and to report + uncertainty here instead, which is what makes INBOX_TRIAGE_MIN_CONFIDENCE a usable + knob rather than a second, contradictory gate. + + reason_code is the Triage_Reason_Code enum rather than a Literal, so the labels + live in one place; Pydantic renders it as the same JSON-schema enum either way. + """ + + model_config = ConfigDict(extra="forbid") + + is_application: bool + reason_code: Triage_Reason_Code + confidence: float = Field(ge=0.0, le=1.0) + # One clause naming the signal used. Stored for the review screen, never logged: + # the model is told not to quote personal data, but it is still model-authored + # text derived from an email body. + evidence: str = Field(min_length=1, max_length=200) diff --git a/backend/inbox_classifier/plugins.py b/backend/inbox_classifier/plugins.py new file mode 100644 index 0000000..e71c0c8 --- /dev/null +++ b/backend/inbox_classifier/plugins.py @@ -0,0 +1,107 @@ +"""Intake-gate configuration, the fail policy, and log-safe digests. + +Pure module: no FastAPI imports and no HTTPException. + +Non-DB config is module-level load_dotenv() + os.getenv (house style). The model / +token / effort / cache knobs come from the bulk-ats Settings instead, exactly as +job/candidate/plugins.get_scoring_settings does, so OPENAI_MODEL and +OPENAI_MAX_OUTPUT_TOKENS keep one meaning per process. get_triage_settings() calls +get_settings() lazily, never at import: it validates OPENAI_MODEL and would otherwise +turn a stale env var into an import failure. +""" + +from __future__ import annotations + +import hashlib +import os + +from app.core.config import Settings, get_settings +from dotenv import load_dotenv + +from inbox_classifier.enums import Triage_Status +from inbox_classifier.prompt import PROMPT_VERSION + +load_dotenv() + + +def _flag(name, default) -> bool: + raw=(os.getenv(name) or "").strip().lower() + if not raw: + return default + return raw in ("1","true","yes","on") + + +# false restores the pre-gate behaviour exactly: every message is ingested and no +# triage row is written. The rollback lever — no code revert needed. +TRIAGE_ENABLED=_flag("INBOX_TRIAGE_ENABLED",True) + +# true: a provider outage or a missing key ingests the mail and stamps the verdict +# unclassified. The app already boots without OPENAI_API_KEY (main.py logs "llm startup +# skipped"), so fail-closed would silently make ingestion a no-op there. +TRIAGE_FAIL_OPEN=_flag("INBOX_TRIAGE_FAIL_OPEN",True) + +TRIAGE_CONCURRENCY=max(int(os.getenv("INBOX_TRIAGE_CONCURRENCY") or 5),1) +TRIAGE_MAX_SUBJECT_CHARS=max(int(os.getenv("INBOX_TRIAGE_MAX_SUBJECT_CHARS") or 300),1) +# ~1000 tokens. Application intent is always in the first screen of a mail, and this +# cap is what bounds cost and latency at 100 messages per fetch. +TRIAGE_MAX_BODY_CHARS=max(int(os.getenv("INBOX_TRIAGE_MAX_BODY_CHARS") or 4000),1) +# 0 disables the uncertainty branch entirely (0.0 < 0.0 is False). +TRIAGE_MIN_CONFIDENCE=float(os.getenv("INBOX_TRIAGE_MIN_CONFIDENCE") or 0) + +# One value for the whole deployment: the cacheable prefix is the system prompt, which +# does not vary per message or per batch. Versioned so a prompt edit never shares a +# cache route with the old text. +PROMPT_CACHE_KEY=f"inbox-triage-{PROMPT_VERSION}" + +UNCLASSIFIED_PREFIX="unclassified:" + +# For the review route's 422 check. Derived from the enum so the two never drift. +TRIAGE_STATUSES=tuple(item.value for item in Triage_Status) + + +def get_triage_settings() -> Settings: + """Validated OpenAI knobs (model family, token floor, effort, cache). + + Reads real env vars, which load_dotenv() above has populated from the nearest .env, + so OPENAI_MODEL / OPENAI_MAX_OUTPUT_TOKENS match what llm_setup uses. + """ + return get_settings() + + +def triage_model_name() -> str: + """The configured model, for the audit column. "" rather than raising. + + Reads settings, not the classifier: this is called while recording a verdict, and + building a client there would turn an audit field into an ingestion failure. + """ + try: + return get_triage_settings().openai_model + except Exception: + return "" + + +def subject_digest(subject) -> str: + """A stable, PII-safe handle for correlating log lines about one subject.""" + return hashlib.sha256((subject or "").encode("utf-8")).hexdigest()[:16] + + +def sender_domain(address) -> str: + """Domain only. The full address is PII and must never be logged.""" + address=(address or "").strip().lower() + return address.rsplit("@",1)[-1] if "@" in address else "" + + +def should_ingest(verdict, error_code="") -> tuple[bool,str,str]: + """(ingest, status, reason_code) — the entire fail policy, in one place. + + verdict None means the model could not be consulted: no API key, invalid config, + timeout, rate limit, refusal, truncation, or an email with no subject and no body to + judge. INBOX_TRIAGE_FAIL_OPEN decides, and the row is stamped unclassified: so + the review route can find every one of them. + """ + if verdict is None: + reason=f"{UNCLASSIFIED_PREFIX}{error_code or 'unknown'}"[:60] + return TRIAGE_FAIL_OPEN,Triage_Status.ERROR.value,reason + if verdict.confidence\n" + "{subject}\n" + "\n{body}\n\n" + "" +) + + +def build_email_block(subject, body) -> dict: + """The one content block. Delimiters are prompt text, not parsed markup. + + Nothing is escaped: there is no XML parser downstream, and the system prompt is what + defends against instruction-shaped content. Escaping here would only corrupt ordinary + resume punctuation. + """ + return { + "type":"input_text", + "text":_EMAIL_TEMPLATE.format(subject=subject,body=body), + } + + +def build_user_content(subject, body) -> list: + return [build_email_block(subject,body)] + + +def build_input(subject, body) -> list: + """The full ``input`` argument for ``responses.parse``.""" + return [ + { + "role":"user", + "content":build_user_content(subject,body), + } + ] diff --git a/backend/main.py b/backend/main.py index c7735e7..2c8bf64 100644 --- a/backend/main.py +++ b/backend/main.py @@ -63,6 +63,13 @@ async def lifespan(app): finally: if agent_ready and close_agent is not None: await close_agent() + # Before close_llm(): the classifier holds a reference to the shared client, + # which close_llm() disposes. No init counterpart — get_classifier() is lazy. + try: + from inbox_classifier.agent_setup import close_classifier + close_classifier() + except Exception as exc: + logger.warning("classifier close skipped: %s",exc) if llm_ready and close_llm is not None: await close_llm() if cv_broker_ready and cv_broker is not None: diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..d7c8d2e --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,40 @@ +"""Fixtures for the backend suite. + +The backend runs *from* `backend/` and has no __init__.py anywhere, so its modules are +top-level imports (`import inbox_classifier.prompt`). pytest is invoked from the repo +root, so `backend/` has to go on sys.path here — the root suite (tests/) imports the +installed `app` package instead and needs no such help. + +No live API calls anywhere: the adapter is exercised against a fake `responses` +resource, exactly as tests/unit/test_llm.py does. +""" + +from __future__ import annotations + +import os +import sys +from collections.abc import Iterator +from pathlib import Path + +import pytest + +_BACKEND = Path(__file__).resolve().parent.parent +if str(_BACKEND) not in sys.path: + # APPEND, never insert(0): backend/ contains a `tests` directory of its own, so + # putting it first would shadow the root `tests` package and break the root + # suite's `from tests.conftest import ...` imports. + sys.path.append(str(_BACKEND)) + + +@pytest.fixture(autouse=True) +def _hermetic_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Keep the suite hermetic. + + A real key must never leak in from the environment, and a developer's local + OPENAI_MODEL or INBOX_TRIAGE_* values must not change what the tests assert. + """ + for name in list(os.environ): + upper = name.upper() + if upper.startswith(("OPENAI_", "ANTHROPIC_", "INBOX_TRIAGE_", "SCORING_", "MAX_")): + monkeypatch.delenv(name, raising=False) + yield diff --git a/pyproject.toml b/pyproject.toml index 4d04d37..96b1d8f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,5 +67,8 @@ ignore_missing_imports = true [tool.pytest.ini_options] asyncio_mode = "auto" -testpaths = ["tests"] +# backend/tests covers the HRMS service (inbox intake gate); tests/ covers this +# package. backend/tests/conftest.py puts backend/ on sys.path, since that service +# runs from its own directory with top-level imports and no __init__.py. +testpaths = ["tests", "backend/tests"] addopts = "-q"