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"