HR-ATS-Portal/backend/README.md

HR-ATS-Portal — Backend

FastAPI service behind TalentFlow, an HR / Applicant-Tracking portal. It pulls candidate applications out of a mailbox, decodes CV attachments, runs an LLM agent to match each CV against live job posts, publishes job ads to social channels through Buffer, and serves the whole thing to the frontend behind JWT auth and a database-driven RBAC model.

Everything in this document describes backend/ only.

---

Table of contents

• Answering your questions
• Architecture
• Tech stack
• Directory layout
• House style — what lives in which file
• Domains
• Data model
• API reference
• Authentication and RBAC
• Background jobs
• The matching agent
• The ATS scoring engine
• External integrations
• Configuration
• Running locally
• Database migrations
• Docker
• Response conventions
• Known gaps and gotchas

---

Answering your questions

A short orientation on the ATS work that arrived with the dashboard branch, for anyone opening this repo for the first time. Every claim links to the section that carries the detail.

Is the ATS linked to the tables, or just an agentic flow?

Both, and the split is the important part. The engine is app/ at the repo root — not under backend/ — and it is stateless: no SQLAlchemy, no session, no table. It is imported as a library (pip install -e ..), not called over HTTP; app/api/routes.py and app/main.py are dead weight here. The linkage is job/candidate/views.py::CandidateScoring, which builds the JD, calls the engine, and persists everything to candidates.

So: agentic scoring, fully relational output. Full detail in The ATS scoring engine.

What it gets from the engine

ATSScore, eight validated fields: candidate_name, job_title, current_company, years_experience (0–60), match_score (0–100, required), matched_keywords / missing_keywords (≤30, deduplicated), summary_critique (1–500 chars). See What the engine gives back.

What it requires from the system

A job_description string built only from JobPosts columns in a fixed order (build_job_description, which excludes post_text and salary to stay byte-stable for prompt caching), plus résumé bytes from either an upload or inbox_messages.file_path. See What the system gives the engine.

Where the values land

1:1 into candidates, plus context the engine never sees (job_id, source, inbox_message_id, content_sha256, created_by, model). Results merge by slot index, never by filename, since inbox attachments routinely collide on resume.pdf. See Where the values land.

Routing

Two manual routes (/candidate/score, /candidate/score_inbox) and two automatic triggers (after every CV match in inbox/tasks.py, and on PATCH /inbox/{id}/assign-job-post), all idempotent, both automatic paths wrapped so a scoring failure never fails the match. See Routing in code.

What is missing — the part worth acting on

Gap	Effect
ats_results is a dead table	Declared with a full supersede chain, migrated, 0 rows, no reader, no writer anywhere in backend/
inbox_messages.ats_score / ats_band are read but never written	serialize_application returns them, so the Applications tab shows null even for candidates that do have a completed score
Re-scoring destroys history	upsert_candidate updates in place on (job_id, content_sha256)
.gitignore line 56 (**_**_**.py) ignores generated migrations	14 of 18 on disk are untracked, so a fresh clone cannot reach head; with DB_AUTOGENERATE=true every developer invents their own revision ids for the same change

The full list, including the DOC/DOCX limitation and the missing wrapper tests, is under What is missing and Known gaps and gotchas.

How this document was verified

The route tables were written from source and then checked against the running service: all 59 documented route rows match /openapi.json, every internal anchor resolves, and all 13 permission tags resolve against PermissionTag. That check caught four real errors worth repeating, since the same mistakes are easy to make from reading alone: /candidate/stage/fetch does not exist (it is /pipeline/transitions/fetch), /offers/update is PATCH not PUT, /offers/issue was missing entirely, and the pipeline and assignment guards are pipeline.* / jobs.*, not candidates.* / job_board.*.

---

Architecture

flowchart TB
    FE["Frontend (Vite, :5173)"] -->|JWT Bearer| API

    subgraph API["FastAPI — main.py :8000"]
        R1["users / role / forget_password / notifications"]
        R2["inbox"]
        R3["job (job_post + candidate)"]
    end

    API --> PG[("PostgreSQL — schema app")]
    API -->|enqueue| REDIS[("Redis Streams")]

    REDIS --> W["Taskiq worker\ninbox.tasks + inbox.sync_tasks"]
    REDIS --> WCV["Taskiq CV worker\ninbox.cv_tasks"]
    SCHED["Taskiq scheduler\ncron"] --> REDIS
    SCHEDCV["Taskiq CV scheduler\nretries"] --> REDIS
    W --> PG
    WCV --> PG
    W --> AGENT["LangGraph agent\nagent/"]
    WCV --> AGENT
    AGENT --> OAI["OpenAI"]

    API -->|GET /emails, /sync/read-status| MAILAPI["Email API (MS Graph proxy)"]
    W --> MAILAPI
    API -->|multipart send| TEAMS["Teams Mail API"]
    API -->|GraphQL| BUF["Buffer"]

The application flow, end to end:

1. GET /email/fetch pulls messages from the external Email API, decodes PDF/DOC/DOCX attachments to inbox/decoded_attachments/, and upserts them into inbox_messages.
2. When a message carries an attachment, the sender is linked to a users row — created with the candidate role if new — through the inbox join table.
3. Any message with a stored attachment and no match result yet is enqueued onto Redis as an inbox.match_message task.
4. The worker extracts the résumé text, hands it plus the active job posts to the LangGraph agent, and writes suggested_job_post_ids, match_summary, match_reasoning and experience back onto the row.
5. Still inside the same task, the ATS engine auto-scores that CV against one job — the assigned post if there is one, otherwise the agent's top suggestion — and writes a row to candidates. See The ATS scoring engine.
6. New candidate accounts land inactive and are mailed a confirmation link; the link is what flips is_active.
7. A cron task sweeps Outlook read-status deltas back onto inbox_messages.message_read.
8. Recruiters read all of it through /inbox/all-applications and publish new roles with POST /job/post-job, which renders the ad copy and pushes it to Buffer.
9. The dashboard reads analytics/ (KPIs, hiring trend, funnel, recruiter and source performance), which aggregates over inbox, application_stage_transitions, offers, job_posts, hiring_costs and job_assignments.

Two different LLM passes, often confused. The matching agent answers "which of our open jobs is this CV for?" and writes onto inbox_messages. The ATS scoring engine answers "how well does this CV fit one chosen job, 0-100?" and writes to candidates. They run back-to-back in the same task but are separate codebases with separate prompts.

---

Tech stack

Concern	Choice
Web framework	FastAPI 0.136 + Uvicorn
ORM / models	SQLModel 0.0.38 on SQLAlchemy 2.0 (async, asyncpg)
Database	PostgreSQL, application objects in the app schema
Migrations	Alembic 1.18, driven by alembic_setup.py
Auth	PyJWT (HS256) access / refresh / reset tokens, bcrypt hashing
Task queue	Taskiq on Redis Streams, with a smart-retry + dead-letter middleware
LLM	OpenAI async client, orchestrated by LangGraph
PDF extraction	pypdf
HTTP client	httpx
Python	3.12 (see Dockerfile)

---

Directory layout

backend/
├── main.py                   # FastAPI app, lifespan, CORS, router mounting
├── db_setup.py               # Settings, async engine, sessions, init_db/lifespan
├── alembic_setup.py          # Alembic scaffolding, autogenerate, migrate-on-boot
├── llm_setup.py              # AsyncOpenAI client + llm_call helper
├── requirements.txt
├── Dockerfile                # image for the Taskiq worker / scheduler
├── alembic.ini               # generated by alembic_setup.py, not hand-written
├── migrations/               # generated env.py + versions/
│   └── manual/               # one-shot SQL (enum labels, RBAC seed, backfills)
├── LLM_CONTEXT_PROMPT.md     # house-style prompt to paste into an LLM before editing
│
├── users/                    # accounts, login, signup, RBAC enforcement
├── role/                     # roles, permission bundles, permission tags
├── forget_password/          # reset-code request → verify → new password
├── notifications/            # email-confirmation tokens and mail
├── inbox/                    # mailbox sync, attachments, applications
├── analytics/                # dashboard KPIs + charts (views only — no tables)
├── offer/                    # offers + offer_status_history
├── job/
│   ├── app.py                # routes for both sub-domains
│   ├── job_post/             # job ads + Buffer publishing
│   ├── candidate/            # CV reading, candidate profile, stage transitions model
│   ├── assignment/           # job_assignments + application_assignments
│   ├── cost/                 # hiring_costs
│   └── pipeline/             # stage-change service (single writer)
├── agent/                    # LangGraph CV → job-post matching agent
└── taskiq_management/        # broker, scheduler, DLQ middleware, smoke task

One dependency lives outside backend/: the bulk-ATS scoring engine at the repo root.

<repo root>/
├── app/                      # the bulk-ATS engine — imported as a library, never over HTTP
│   ├── models/scoring.py     #   ATSScore / CompletedCandidate / FailedCandidate
│   ├── services/pdf.py       #   extract_resume
│   ├── services/llm.py       #   OpenAIScorer (the Scorer protocol)
│   ├── services/scoring.py   #   score_batch, verify_matched_keywords
│   └── api/, main.py         #   its standalone FastAPI app — UNUSED by this backend
├── tests/                    # tests for app/ only; nothing covers the backend wrapper
└── CLAUDE.md                 # the engine's own spec

Install it once per environment, from backend/: pip install -e ..

There are no __init__.py files. The service is run from backend/, so imports are top-level (from users.app import router, from db_setup import get_session).

---

House style — what lives in which file

Every domain package follows the same six-file shape. This is enforced by convention, and LLM_CONTEXT_PROMPT.md is the canonical statement of it.

File	Owns	Must not do
app.py	Routes, inline request models, JSONResponse, dependency injection	Business rules, SQL
views.py	Business checks, calls models, raises HTTPException	Build login token envelopes
models.py	SQLModel table + @classmethod async def accessors	Import FastAPI, raise HTTPException
serializers.py	Hand-built dict builders (serialize_*)	Touch the DB or Depends
plugins.py	Pure helpers — hashing, JWT, HTTP calls to third parties	Import FastAPI
permissions.py	Bearer schemes and Depends aliases (auth domains only)	Hold route handlers

Additional rules that matter when you edit this code:

• Request bodies are Pydantic models declared inline in app.py, never in serializers.py.
• There are no Pydantic response models — responses are hand-built dicts.
• Route paths are verb-in-path (/users/create, /users/fetch), not REST-resource-only, and there is no /api/v1 prefix.
• Non-DB config is module-level load_dotenv() + os.getenv(...). Only database settings go through db_setup.Settings.

---

Domains

users/

Signup, login, refresh, CRUD, role assignment, and the RBAC machinery every other domain depends on. users/permissions.py defines the full PermissionTag vocabulary (13 modules × 8 actions = 104 tags) and the require_permission(...) dependency. A startup assertion (_assert_vocabulary_complete) fails loudly if the tag list ever drifts from PermissionModule × PermissionAction.

Signup always assigns the candidate role, creates the account inactive, and sends a confirmation email. Login rejects unconfirmed accounts.

role/

Three-level permission model: permission_tags (atomic module.action rows) → permissions (named bundles holding a JSONB array of tag ids) → roles (holding a JSONB array of bundle ids). Roles.resolve_tags() walks that chain and returns a flat tuple of tag names; dangling or inactive ids simply contribute nothing rather than erroring.

Eight system roles are seeded: system_administrator, hr_administrator, recruiter, hiring_manager, department_head, interviewer, ceo, candidate.

inbox/

The heart of the ingestion pipeline.

• views.py::Email talks to the external Email API, decodes attachments, upserts messages, and enqueues matching work.
• models.py holds Inbox_Messages (the mail rows plus all agent output columns), Inbox_Alerts, and Inbox — the join table linking a message to the candidate Users row it came from. _link_sender creates the candidate account on first contact, skipping noreply@-style senders.
• file_decoder.py turns Graph contentBytes into real PDF / DOCX / DOC files, validating magic bytes for each format and stripping path traversal from filenames.
• plugins.py resolves attachment paths (handling Windows paths written by the host API but read from a Linux worker), extracts résumé text, and calls the read-status sync endpoints.
• tasks.py / sync_tasks.py are the two Taskiq tasks.

job/

Two sub-domains behind one router:

• job_post/ — renders LinkedIn-shaped ad copy from a structured payload, resolves the Buffer channel (by explicit id, by platform alias, or by the configured default), creates the post over Buffer's GraphQL API, and records the mapped status. A queued post is recorded as scheduled, not published; only Buffer reporting sent promotes it.
• candidate/ — FileRead extracts text from an uploaded PDF (pypdf), and match_inbox_cv force-requeues an existing inbox message for matching. CandidateView reads the candidate profile through the inbox join and fills ai_score / recommendation from candidates. CandidateScoring is the ATS wrapper — see The ATS scoring engine. models.py also owns Activity, Feedback, Interviews, Notes, Candidates and ApplicationStageTransitions.
• pipeline/ — Pipeline.change_stage, the single writer of application_stage_transitions. Nothing else may move an application between stages.
• assignment/ — job_assignments and application_assignments, both temporal (valid_to IS NULL = current).
• cost/ — hiring_costs, the numerator of cost-per-hire.

analytics/

Read-only aggregation for the dashboard — views and serializers only, no tables of its own. Every window bound it builds is timezone-aware UTC, which is why every timestamp column it touches must be timestamptz.

offer/

offers plus offer_status_history, same temporal shape as the stage transitions.

notifications/ and forget_password/

Two parallel token flows, deliberately kept separate so each owns its own mail copy and env reads:

• Confirmation — a 32-byte url-safe secret, bcrypt-hashed in email_confirmation_tokens; the link carries <row_id>.<secret> because a bcrypt hash cannot be looked up. Replays (mail scanners, back button) are handled idempotently.
• Password reset — a short code mailed to the user, bcrypt-hashed in password_reset_codes, with a resend cooldown and a max-attempts cap. Verifying the code mints a type=reset JWT carrying the code row id (crid), which is the only thing that authorises the new-password call.

agent/

LangGraph state machine — see The matching agent.

taskiq_management/

Broker, scheduler, DLQ middleware, and a ping smoke task.

---

Data model

All tables live in the app schema (DB_DEFAULT_SCHEMA), with a shared naming convention for indexes, constraints and foreign keys. SQLModel.metadata is pointed at Base.metadata in db_setup.py so SQLModel and DeclarativeBase share one registry and Alembic sees everything.

Table	Key columns	Notes
users	id (uuid PK), email (unique), role_id → roles.id, password, is_active, is_deleted	Soft delete. role is selectin-loaded; lazy loads would raise MissingGreenlet under asyncio
roles	id, role_name (unique), permissions (JSONB int[]), is_system
permissions	id, name (unique), permission_tags (JSONB int[])	Named bundles
permission_tags	id, tag_name (unique), module, action	Unique on (module, action)
inbox_messages	id (uuid), message_id (upstream id, unique), full_email_response (JSONB), subject/body/from/to/cc/bcc, message_read, attachment, file_name, file_path, application_status, resume_text, experience, suggested_job_post_ids (JSONB), match_summary, match_reasoning, match_status, match_error, matched_at	One row per mail; agent output lands here
inbox	id, user_id → users.id, message_id → inbox_messages.id, alert_id	Join table linking a candidate to a message
inbox_alerts	id, alert_sender_name, alert_sender_email, is_read
job_posts	id (uuid), title, platform, channel_id, post_text, requirements/optional_skills (JSON), status, buffer_post_id, buffer_external_link, buffer_sent_at, buffer_error, created_by → users.id
password_reset_codes	id, email, code_hash, expires_at, attempts, is_used, verified_at
email_confirmation_tokens	id, user_id, email, token_hash, expires_at, is_used, confirmed_at

Tables added by the dashboard + ATS work

Nine tables landed together with analytics/, offer/, job/assignment/, job/cost/ and job/pipeline/. Owning module in brackets.

Table	Key columns	Notes
candidates (job/candidate)	id, job_id → job_posts.id, source (upload|inbox), inbox_message_id → inbox_messages.id, filename, file_path, content_sha256, candidate_name, job_title, current_company, years_experience, match_score, matched_keywords/missing_keywords (JSON), summary_critique, status, error_code, error_message, model, created_by	The ATS result table. Unique on (job_id, content_sha256) so re-scoring the same bytes against the same job updates in place. status is completed | failed; a failed row keeps match_score NULL and carries the error instead
application_stage_transitions (job/candidate)	id, inbox_id → inbox.id, from_stage, to_stage, valid_from, valid_to, changed_by, actor_kind, change_reason	Temporal history of inbox_messages.application_status. valid_to IS NULL = current stage; from_stage IS NULL = pipeline entry. Time-in-stage is a subtraction, not a window function. Single writer: job/pipeline/views.py::Pipeline.change_stage
job_assignments (job/assignment)	id, job_post_id, user_id, assignment_role, valid_from, valid_to	Who owns a requisition. Open rows (valid_to IS NULL) are what Recruiter Performance counts as open reqs
application_assignments (job/assignment)	id, inbox_id, user_id, assignment_role, valid_from, valid_to	Same temporal shape, per application
hiring_costs (job/cost)	id, job_post_id, cost_type, amount, currency, incurred_at, created_by	Numerator of the cost-per-hire KPI
offers (offer)	id, inbox_id, job_post_id, status, salary, start_date, expiry_date, sent_at, responded_at, closed_at	Feeds offers_sent / offers_accepted
offer_status_history (offer)	id, offer_id, from_status, to_status, valid_from, valid_to	Temporal history of offers.status
source_channels (inbox)	id, key (unique), label, is_active	The eleven BRD sourcing channels, seeded by migrations/manual/001
ats_results (inbox)	id, inbox_id, job_post_id, overall_score, band, is_current, superseded_by_id, model_name, computed_at	Declared but unused — no code reads or writes it. See Known gaps

inbox_messages also gained denormalised dashboard columns: ats_score, ats_band, recruiter_id, is_duplicate, source_channel_id, processing_state. ats_score and ats_band are read by serialize_application but never written by anything — the real score is joined from candidates at read time.

application_status is a str enum, extended by migrations/manual/001: PROCESS, PENDING, APPROVED, REJECTED, ONHOLD, CLOSED, SCREENING, ASSESSMENT, INTERVIEW, OFFER, HIRED.

match_status is free-form text written by the worker: processing, matched, skipped, no_text, failed, dlq.

Every timestamp column in the app schema is timestamptz. Model defaults are _now() = datetime.now(timezone.utc), never bare datetime.now(), which returns the writing host's local wall clock. This is load-bearing rather than stylistic: analytics/ builds its window bounds as aware UTC, and binding an aware datetime against a naive column makes asyncpg raise DataError: can't subtract offset-naive and offset-aware datetimes in its parameter encoder — the statement never reaches Postgres. A naive default written into an already-timestamptz column is worse, because it does not raise at all: asyncpg reads the local value as UTC and silently backdates the row.

---

API reference

Base URL: http://localhost:8000. Interactive docs at /docs.

Auth — users/app.py, notifications/app.py, forget_password/app.py

Method	Path	Guard	Purpose
POST	/users/signup	public	Create a candidate account (inactive) and mail a confirmation link
POST	/users/login	public	Email/username + password → token envelope
POST	/users/refresh	public	Refresh token → new token pair
GET	/users/me	any authenticated user	Current user with resolved permissions
POST	/users/confirm-email	public	Consume a confirmation token, activate the account
POST	/users/confirm-email/resend	public	Re-issue a confirmation link (cooldown enforced)
POST	/users/forget-password	public	Mail a reset code
POST	/users/forget-password/verify-code	public	Verify the code → type=reset JWT
POST	/users/forget-password/new-password	reset JWT	Set the new password

Users — users/app.py

Method	Path	Required tag
GET	/users/fetch	rbac_users.view
POST	/users/create	rbac_users.create
PUT	/users/update?record_id=	rbac_users.edit
PUT	/users/assign-role?record_id=	rbac_users.edit (+ rbac_users.manage inside the service)
PUT	/users/remove-role?record_id=	rbac_users.edit (+ rbac_users.manage)
DELETE	/users/delete?record_id=	rbac_users.delete

Roles and permissions — role/app.py

Method	Path	Required tag
GET	/roles/fetch	rbac_users.view
POST	/roles/create	rbac_users.create
PUT	/roles/update?record_id=	rbac_users.edit
DELETE	/roles/delete?record_id=	rbac_users.delete
GET	/permissions/fetch	rbac_users.view
POST	/permissions/create	rbac_users.manage
PUT	/permissions/update?record_id=	rbac_users.manage
GET	/permission-tags/fetch	rbac_users.view

Inbox — inbox/app.py

Method	Path	Guard	Purpose
GET	/email/fetch	upstream token only	Pull from the Email API, decode attachments, upsert, enqueue matching. test_on=true (default) returns raw payloads and skips the account-setup mails
GET	/inbox/fetch	none	Stored messages, with attachments inlined as base64
GET	/inbox/all-applications	inbox.view	The Applications tab. Filters: application_status, isread, search, record_id, top, skip
POST	/inbox/{record_id}/match	inbox.edit	Force a re-match of one message
POST	/inbox/{record_id}/read	inbox.edit	Mark read locally
GET	/inbox/{record_id}/read-status	inbox.edit	Re-pull read status from upstream for one message

Jobs and candidates — job/app.py

Method	Path	Required tag	Purpose
GET	/jobs/alias	public	Accepted platform shorthands (fb, ig, li, x, …)
POST	/job/post-job	job_board.create	Render the ad, create the Buffer post, persist the result
GET	/job/buffer/channels	job_board.view	Connected Buffer channels across all organizations
POST	/candidate/cv_upload	candidates.create	Upload a PDF; extract email, persist like an emailed CV, enqueue matching on the CV stream
POST	/candidate/inbox-match?inbox_message_id=	candidates.edit	Queue a forced re-match for a stored message
GET	/candidate/fetch?user_id=	candidates.view	Candidate profile via the inbox join, with ai_score / recommendation joined from candidates
GET	/job/fetch	job_board.view or candidates.view	List job posts. Either tag suffices — a recruiter scoring CVs needs a job to score against

POST /job/post-job takes mode ∈ addToQueue | shareNow | customScheduled; the customScheduled mode requires scheduler_date (and optionally scheduler_time), which the route combines into a UTC due_at.

ATS scoring — job/app.py

Method	Path	Required tag	Purpose
POST	/candidate/score	candidates.create	Multipart job_id + files[]; score uploaded PDFs, persist, return the leaderboard
POST	/candidate/score_inbox	candidates.create	JSON job_id + message_ids[] (inbox_messages PK uuids, not Graph ids); score decoded attachments
GET	/candidate/scored/fetch?job_id=	candidates.view	Persisted leaderboard; omit job_id for the whole pool
GET	/candidate/fetch_by_id?candidate_id=	candidates.view	One candidates row

Pipeline, assignments, costs — job/app.py

Method	Path	Required tag	Purpose
GET	/pipeline/transitions/fetch	pipeline.view	Stage history by transition_id or inbox_id
PATCH	/candidate/stage	pipeline.edit	Move an application to a stage. The only writer of application_stage_transitions — closes the open row, writes the new one, updates application_status. 400 if already at that stage, 422 on an invalid to_stage
GET	/job/assignments/fetch	jobs.view	Requisition assignments
POST	/job/assignments/create	jobs.edit	Assign a user to a requisition
GET	/candidate/assignments/fetch	candidates.view	Application assignments
POST	/candidate/assignments/create	candidates.edit	Assign a user to an application
GET	/job/costs/fetch	jobs.view	Hiring costs; filters job_post_id, from_date, to_date
POST	/job/costs/create	jobs.edit	Record a hiring cost
GET	/activity/fetch	candidates.view	Activity feed — activity_id, inbox_id, or top/skip for the global feed
POST	/activity/create	candidates.create	Write an activity row; links via inbox_id, message_id, or user_id

Analytics — analytics/app.py

All require analytics.view. Common query params: from_date, to_date, department, recruiter_id.

Method	Path	Purpose
GET	/analytics/kpis/fetch	The KPI cards, each with a prior-period comparison
GET	/analytics/hiring-trend/fetch?months=7	Applications vs hires by month
GET	/analytics/funnel/fetch	Candidate count per stage
GET	/analytics/recruiter-performance/fetch?top=5	Per recruiter: hires, open reqs, avg time-to-hire
GET	/analytics/source-performance/fetch	Applications per source channel

Recruiter Performance iterates users whose role is recruiter; with no such user the list is empty regardless of the rest of the data.

Offers — offer/app.py

Method	Path	Required tag
GET	/offers/fetch	offers.view
POST	/offers/create	offers.create
PATCH	/offers/update	offers.edit
POST	/offers/issue	offers.approve

---

Authentication and RBAC

Tokens. PyJWT, HS256, with a type claim that decode_token(..., expected_type=...) rejects on mismatch. Three types:

Type	Lifetime (default)	Extra claims
access	30 min	email, role_id
refresh	7 days	—
reset	10 min	crid (reset-code row id)

iat / exp are always timezone-aware UTC. Every token carries a jti.

Login response puts the OAuth2 fields at the root so Swagger's Authorize button can read them:

{
  "access_token": "...",
  "refresh_token": "...",
  "token_type": "bearer",
  "expires_in": 1800,
  "data": { "id": "...", "email": "...", "role_id": 3, "role_name": "recruiter", "...": "..." },
  "status_code": 200
}

Passwords use bcrypt directly rather than passlib — passlib 1.7.4 reads bcrypt.__about__.__version__, which bcrypt dropped in 4.1, and the failed probe makes it reject every password as over 72 bytes. Input is truncated to 72 bytes on a character boundary before hashing.

Authorization. get_current_user decodes the access token, loads the user by sub, rejects missing / deleted / inactive accounts, resolves the role's tags, and returns a serialized user dict with a permissions list. require_permission(*tags, require_all=True) is the dependency that guards routes; a user with no role assigned gets a 403 before any tag check runs.

Role assignment is additionally guarded in users/views.py: you cannot grant a role holding permissions you do not yourself hold.

---

Background jobs

Taskiq over Redis Streams, with a result backend and a Redis-backed schedule source.

Middleware order matters — DeadLetterMiddleware sits before SmartRetryMiddleware so a PermanentTaskError can set retry_on_error=False before retry logic runs.

Task	Trigger	What it does
inbox.match_message	enqueued by /email/fetch, /inbox/{id}/match, /candidate/inbox-match onto the inbox stream	Extract résumé text → run the agent → write match results → auto-score with the ATS
inbox.match_message (CV broker)	enqueued by /candidate/cv_upload onto the cv_upload stream	Same matcher as above; isolated so uploads never sit behind /email/fetch backlog
inbox.score_message	enqueued by PATCH /inbox/{id}/assign-job-post onto the inbox stream	ATS-score one message against one job. Idempotent — a completed (message, job) pair returns already_scored without paying for a second call
inbox.sync_read_status	cron, EMAIL_SYNC_CRON (default every minute)	Pull read-status deltas from the Email API and apply them
ping	manual	Framework smoke test

Auto-scoring never fails a match. match_inbox_message commits the agent result first, then scores inside its own try; a scoring exception is logged and swallowed. Likewise the enqueue in set_assigned_job_post is wrapped — a broker outage logs a warning and leaves the manual Score with ATS button as the fallback.

Retries. SmartRetryMiddleware with jitter and exponential delay, TASKIQ_MAX_RETRIES attempts, capped at TASKIQ_MAX_DELAY. Raising PermanentTaskError (missing record, no attachment, blank record_id) skips retries entirely.

Dead letter queue. Exhausted or permanently-failed tasks are written to the taskiq:dlq Redis stream as a JSON payload. For inbox.match_message specifically, the middleware also stamps match_status="dlq" on the row so the failure is visible in the UI rather than only in Redis.

The read-status latch. apply_read_status only ever applies false → true. Nothing pushes local reads back to Outlook, so upstream keeps reporting isRead=false; without the latch the every-minute sweep would un-read a mail the user just opened. The trade-off is that un-reading a mail in Outlook no longer propagates here. sync_read_status holds a Redis lock (inbox:sync_read_status:lock, 300s TTL) so overlapping cron ticks cannot double-run, and pages at most 10 rounds per tick.

---

The matching agent

A small LangGraph StateGraph over an AgentState TypedDict.

START → prepare → (conditional) → match_jobs → END
                └→ END        (when resume_text or job_posts is empty)

• prepare_context trims the subject and résumé text and normalizes the job posts down to the fields the model needs. Empty résumé text or no active posts short-circuits to status="skipped".
• match_jobs calls llm_call(prompt(), user_prompt(state), json_mode=True) and parses the response.

The parser (agent/decorators.py::parse_match_response) is deliberately strict: a suggested id must be present in the posts that were actually sent, must be a valid UUID, and duplicates are dropped. The model cannot invent a job post. reasoning returned as a list is joined into a string; non-string fields fall back to "".

The graph is compiled once per process (init_agent() from the FastAPI lifespan, or lazily on first use inside the worker) and the OpenAI client is created lazily and closed on shutdown.

Startup is fault-tolerant: if the broker or the LLM/agent fails to initialize, main.py logs a warning and the API still serves. Only the database is a hard requirement.

---

The ATS scoring engine

Is it linked to the database, or just an agentic flow?

Both, and the distinction matters. The engine itself is stateless and knows nothing about this database; the backend wraps it and owns all persistence.

• The engine — app/ at the repo root, not under backend/ — is the standalone bulk-ATS service (its own CLAUDE.md at the repo root is its spec). It takes a job-description string and résumé bytes and returns validated Pydantic objects. No SQLAlchemy, no session, no table.
• The backend imports it as a library, installed editable from the repo root: pip install -e .. → import app.* (see the tail of requirements.txt). It is not called over HTTP, and app/api/routes.py and app/main.py are unused here.
• job/candidate/views.py::CandidateScoring is the seam: it builds the JD from a JobPosts row, feeds the engine, and persists every result to candidates.

So the scoring is agentic, but the output is fully relational. What is not linked: ats_results and inbox_messages.ats_score / ats_band, which exist as columns but have no writer — see what is missing.

What the system gives the engine

Input	Built from	Where
job_description (str)	JobPosts columns only — title, employment_type, location, experience_min/max, description, requirements, optional_skills, in that fixed order	job/candidate/plugins.py::build_job_description
résumé bytes	uploaded UploadFile, or the decoded attachment at inbox_messages.file_path	CandidateScoring.score_uploads / score_inbox
scorer	OpenAIScorer over llm_setup's shared AsyncOpenAI client — one connection pool for the whole process, not a second one	plugins.py::get_scorer
concurrency	SCORING_CONCURRENCY	plugins.py::get_scoring_settings

build_job_description deliberately excludes post_text and salary, and is byte-stable per job: OpenAI prompt caching keys on an exact prefix match, so one volatile byte (an id, a timestamp) would stop the whole batch reusing the cached JD prefix.

Résumé text is run through normalize_spaced_text before scoring, so that keyword verification sees exactly the text the model saw. Designer-made CVs position every glyph individually and pypdf returns S K I L L S; the despace_line decorator rebuilds those.

What the engine gives back

ATSScore (app/models/scoring.py) — eight fields, all validated before they reach the DB:

Field	Type	Constraint
candidate_name	str | None	≤120 chars; null when the CV does not state it
job_title	str | None	≤120; most recent employment entry, verbatim
current_company	str | None	≤120
years_experience	int | None	0–60; a stated total wins, else computed from explicit dates, else null
match_score	int	0–100, required
matched_keywords	list[str]	≤30, deduplicated case-insensitively
missing_keywords	list[str]	≤30, JD-side wording
summary_critique	str	1–500 chars, one sentence

Results come back as a discriminated union — CompletedCandidate or FailedCandidate (filename, error_code, error_message) — so a partial batch cannot reach an invalid state.

matched_keywords are server-verified after parsing: verify_matched_keywords drops any keyword with no case-, separator- and plural-insensitive occurrence in the résumé text, because a matched keyword is an evidence pointer a recruiter reads as "this is in the CV".

Where the values land

Every field maps 1:1 onto candidates (CandidateScoring._score_and_persist):

ATSScore.candidate_name    -> candidates.candidate_name
ATSScore.job_title         -> candidates.job_title
ATSScore.current_company   -> candidates.current_company
ATSScore.years_experience  -> candidates.years_experience
ATSScore.match_score       -> candidates.match_score
ATSScore.matched_keywords  -> candidates.matched_keywords   (JSON)
ATSScore.missing_keywords  -> candidates.missing_keywords   (JSON)
ATSScore.summary_critique  -> candidates.summary_critique
                              candidates.status = "completed"

FailedCandidate.error_code/_message -> candidates.error_code/error_message
                                       candidates.status = "failed", match_score NULL

plus context the engine never sees: job_id, source (upload|inbox), inbox_message_id, filename (sanitised), file_path, content_sha256, created_by, and model (the OPENAI_MODEL that produced the score).

Results merge back by slot index, never by filename — inbox attachments routinely share a basename like resume.pdf. Per-file problems become persisted status="failed" rows rather than sinking the batch, which is a deliberate deviation from the standalone engine's HTTP API (that one rejects the whole request with 413/415).

Routing in code

Manual, from the UI:

POST /candidate/score          (multipart: job_id + files[])      job/app.py
POST /candidate/score_inbox    (json: job_id + message_ids[])     job/app.py
  -> CandidateScoring.score_uploads / .score_inbox                job/candidate/views.py
  -> _score_and_persist
       build_job_description(job)                                 job/candidate/plugins.py
       extract_resume(...) -> normalize_spaced_text(...)          app.services.pdf + plugins
       score_batch(resumes, job_description=, scorer=, concurrency=)   app.services.scoring
       Candidates.upsert_candidate(...) per slot                  job/candidate/models.py
  <- serialize_candidate[] sorted score desc, failures last

Automatic, two triggers, both idempotent:

(a) after every CV match
    inbox/tasks.py::match_inbox_message
      -> assigned_job_post_id, else suggested_job_post_ids[0]
      -> score_message_against_job(record_id, job_id)             inbox/tasks.py:25
         guard: a completed (message, job) row exists -> {"status": "already_scored"}
         attributes rows to job.created_by (no request user in a worker)
         scoring failure is caught and logged; the match result is already committed

(b) on job assignment
    PATCH /inbox/{record_id}/assign-job-post                      inbox/app.py
      -> Email.set_assigned_job_post                              inbox/views.py:178
      -> enqueue task "inbox.score_message" on the `inbox` queue
         broker down -> warning only; the manual button is the fallback

Read paths:

GET /candidate/scored/fetch?job_id=   leaderboard for one job, or the whole pool
GET /candidate/fetch_by_id?candidate_id=
GET /candidate/fetch?user_id=         talent-pool profile — CandidateView joins the
                                      candidates rows on inbox_message_id and fills
                                      ai_score / recommendation / scored_job_post_id

_recommendation bands the score to match the frontend: ≥82 Strong Match, ≥65 Potential Match, else Weak Match. Where a candidate has several scores, the one against the assigned job post wins, else the most recently updated.

All three write routes require candidates.create; read routes require candidates.view.

Configuration

The engine reads its own settings through app.core.config.get_settings(), from the same .env, so the shared names line up with what llm_setup uses:

Variable	Default	Used for
OPENAI_MODEL	gpt-5.4-mini	must support structured outputs
OPENAI_MAX_OUTPUT_TOKENS	4000	covers reasoning and visible tokens; too low truncates mid-JSON
OPENAI_EFFORT	low	omitted automatically for non-reasoning models
OPENAI_ENABLE_PROMPT_CACHE	true
SCORING_CONCURRENCY	5	semaphore bound in score_batch
MAX_RESUMES_PER_REQUEST	50	413 above this
MAX_PDF_SIZE_MB	10	per-file precheck
MAX_JD_CHARS	30000	422 if the rendered JD is larger
MAX_RESUME_CHARS	60000	truncation boundary

What is missing

• ats_results has no writer. The table and the AtsResults model exist with a full supersede chain (is_current, superseded_by_id), but nothing in backend/ references it beyond the class definition, and it holds 0 rows. Either wire it up as the score history (candidates currently overwrites in place on re-score, so history is lost) or drop it.
• inbox_messages.ats_score / ats_band are never written. serialize_application returns them on the Applications tab, so they are always null there, even for a candidate that has a completed score in candidates. Either denormalise on write in _score_and_persist, or have serialize_application join like CandidateView already does.
• Re-scoring destroys the previous result. upsert_candidate matches on (job_id, content_sha256) and updates in place, so there is no record that the score changed or which model produced the earlier one.
• DOC/DOCX CVs cannot be scored. They are decoded and stored, but score_inbox prechecks them to UNSUPPORTED_FILE_TYPE; only PDFs reach the engine.
• No job_id back-reference on the message. The auto-score picks suggested_job_post_ids[0] when nothing is assigned, but does not record which job it chose; you have to read candidates to find out.
• The engine's own test suite (repo-root tests/) does not cover the backend wrapper. Nothing tests build_job_description, the slot-merge, or the upsert.

---

External integrations

Service	Used by	Contract
Email API (a Microsoft Graph proxy)	inbox/	GET {EMAIL_URL}/emails, GET {EMAIL_URL}/emails/{id}, GET {EMAIL_URL}/sync/read-status, GET {EMAIL_URL}/sync/read-status/message/{id} — Bearer EMAIL_API_TOKEN
Teams Mail API	notifications/, forget_password/	multipart POST to TEAMS_MAIL_API_URL; success is HTTP 202, anything else raises
Buffer	job/job_post/	GraphQL against BUFFER_API_URL — createPost mutation, account { organizations } and channels queries
OpenAI	agent/, llm_setup.py	Chat Completions with response_format: json_object

Attachments are written to backend/inbox/decoded_attachments/. In Docker this directory is bind-mounted into the worker so both processes see the same files.

---

Configuration

Copy .env.example to .env and fill it in. .env is git-ignored; .env.example is not. db_setup.Settings reads backend/.env or the repo-root .env; every other module reads its own keys with os.getenv.

Database

Variable	Default	Notes
DB_USERNAME, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME	—	Required. DB_PORT must be an integer
DATABASE_URL	—	Full DSN; wins over the parts above
DB_SSLMODE	—	e.g. require on Azure; translated to asyncpg's ssl
DB_SCHEMAS	app	Comma-separated; created on startup
DB_DEFAULT_SCHEMA	app	Schema for models that declare none
DB_ECHO	false	SQL logging
DB_POOL_SIZE / DB_MAX_OVERFLOW / DB_POOL_RECYCLE	5 / 10 / 1800
DB_CONNECT_RETRIES	10	Startup wait-for-Postgres
DB_AUTO_MIGRATE	true	Run upgrade head on startup
DB_AUTOGENERATE	true	Write a revision when models drift
APP_NAME	hr-ats-portal	Postgres application_name

Auth

Variable	Default
JWT_SECRET_KEY	— (required)
JWT_ALGORITHM	HS256
JWT_ACCESS_TOKEN_EXPIRE_MINUTES	30
JWT_REFRESH_TOKEN_EXPIRE_DAYS	7
JWT_RESET_TOKEN_EXPIRE_MINUTES	10

Email ingestion

Variable	Default
EMAIL_URL, EMAIL_API_TOKEN	—
EMAIL_SYNC_FOLDER	inbox
EMAIL_SYNC_SINCE	—
EMAIL_SYNC_CRON	* * * * *
BACKEND_URL	http://localhost:8000 (used for the internal confirmation-resend call)
DEFAULT_CANDIDATE_PASSWORD	Utopia!@# — placeholder only; the account is inactive until confirmed

Mail out

Variable	Default
TEAMS_MAIL_API_URL, TEAMS_API_TOKEN	—
FRONTEND_URL	http://localhost:5173
CONFIRM_EMAIL_PATH	/auth/confirm-email
CONFIRM_TOKEN_TTL_SECONDS	86400
CONFIRM_TOKEN_RESEND_SECONDS	60
RESET_CODE_TTL_SECONDS	60
RESET_CODE_RESEND_SECONDS	30
RESET_CODE_MAX_ATTEMPTS	5

Buffer

Variable	Default
BUFFER_API	— (access token)
BUFFER_API_URL	https://api.buffer.com
BUFFER_CHANNEL_ID	— (fallback channel)

OpenAI

Variable	Default
OPENAI_API_KEY	—
OPENAI_MODEL	gpt-5.4-mini
OPENAI_TEMPERATURE	0 — leave blank to omit the parameter for models that reject it
OPENAI_MAX_OUTPUT_TOKENS	32768 (.env.example ships 4096)
OPENAI_TIMEOUT / OPENAI_MAX_RETRIES / OPENAI_CONNECT_RETRIES	60 / 3 / 3
OPENAI_BASE_URL, OPENAI_ORGANIZATION, OPENAI_PROJECT	— (set only for Azure or a gateway)

Taskiq / Redis

Variable	Default
REDIS_URL	redis://localhost:6379/0
TASKIQ_QUEUE_NAME	inbox
TASKIQ_CV_QUEUE_NAME	cv_upload
TASKIQ_CONSUMER_GROUP	taskiq
TASKIQ_MAX_RETRIES	3
TASKIQ_RETRY_DELAY	5
TASKIQ_MAX_DELAY	120
TASKIQ_IDLE_TIMEOUT_MS	600000
TASKIQ_DLQ_STREAM	taskiq:dlq
TASKIQ_WORKER_NAME	falls back to HOSTNAME
MANUAL_UPLOAD_TO_ADDRESS	manual-cv-upload@hr-ats.local — To address stamped on synthetic inbox rows so source resolves to Manual CV Upload
APP_VERSION	dev

---

Running locally

Prerequisites: Python 3.12, PostgreSQL, Redis.

cd backend

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -r requirements.txt

cp .env.example .env               # then fill it in

All commands must be run from backend/ — the import paths depend on it.

API:

uvicorn main:app --reload --port 8000

Startup connects to Postgres (retrying with backoff), creates the configured schemas, runs migrations to head, then starts the broker, the OpenAI client and the agent graph. Broker and LLM failures are logged and skipped; the API still comes up.

Worker (needs Redis):

taskiq worker taskiq_management.broker_setup:broker \
  inbox.tasks inbox.sync_tasks taskiq_management.tasks

CV-upload worker (isolated stream for manual uploads):

taskiq worker taskiq_management.cv_broker_setup:cv_broker inbox.cv_tasks

Scheduler (cron ticks for inbox.sync_read_status):

taskiq scheduler taskiq_management.broker_setup:scheduler inbox.sync_tasks

CV-upload scheduler (retries for the CV stream):

taskiq scheduler taskiq_management.cv_broker_setup:cv_scheduler inbox.cv_tasks

Docs: http://localhost:8000/docs

---

Database migrations

alembic_setup.py wraps Alembic so the plain alembic CLI and the app's own migrate-on-startup share one configuration. It scaffolds alembic.ini, migrations/env.py and script.py.mako on first use and never overwrites them. Model modules are discovered automatically — every <package>/models.py under backend/ is imported before the metadata is diffed.

python alembic_setup.py migrate            # upgrade to head, then autogenerate any drift
python alembic_setup.py revision -m "add x" # write a revision if the models have drifted
python alembic_setup.py upgrade -r head
python alembic_setup.py downgrade -r -1
python alembic_setup.py current
python alembic_setup.py head

Alembic autogenerate does not detect new PostgreSQL enum labels. Permission-tag rows and analytics role bundles are also seeded out-of-band. Those live in migrations/manual/ and must be run by hand in psql (autocommit for ADD VALUE):

# After `python alembic_setup.py upgrade` has created the new tables/columns:
psql "$DATABASE_URL" -f migrations/manual/001_dashboard_rbac_and_enum.sql

001_dashboard_rbac_and_enum.sql extends candidate_application_status, seeds all 104 permission_tags, creates the analytics_dashboard bundle and attaches it to the system roles that need the dashboard, seeds the eleven BRD source_channels, and backfills source_channel_id / stage-transition / requisition-status rows.

<<<<<<< HEAD

Run it with the session timezone set to UTC. The file writes NOW() into columns of both kinds. A client whose session timezone is not UTC stores a shifted wall clock in any naive column and a correct instant in the timestamptz ones, which is how the current dev data ended up with source_channels rows seven hours off from the application_stage_transitions rows written by the same transaction. The database's own default (pg_settings.reset_val) is UTC; it is the client that overrides it.

PGTZ=UTC psql "$DATABASE_URL" -f migrations/manual/001_dashboard_rbac_and_enum.sql

migrations/versions/*.py is effectively git-ignored. .gitignore line 56 carries the pattern **_**_**.py, which matches every generated revision filename (20260812_1035-b3f1c2d4e5a6_inbox_timestamps_tz_aware.py and friends). Only the four revisions committed before that rule landed are tracked — 14 of the 18 on disk are not, so a fresh clone cannot reach head. Combined with DB_AUTOGENERATE=true, each developer's instance invents its own revision ids for the same schema change and the histories diverge. Fix the pattern and commit the missing revisions before anyone else clones this branch. =======

>>>>>>> c283ac0e50dbe497671957e7fb064edb75f5988b

Migrations run under a Postgres advisory lock, so several workers booting at once cannot
migrate concurrently. Empty revisions are suppressed. Alembic's own `alembic_version` table is
excluded from autogenerate, as is anything outside the configured schemas.

The module is named `alembic_setup` rather than `alembic` because `backend/` is on `sys.path`
and a module called `alembic.py` would shadow the installed package.

---

## Docker

The repo-root `docker-compose.yml` runs Redis plus the four Taskiq processes (inbox
worker/scheduler and CV-upload worker/scheduler); the API itself is expected to run on the
host (the compose file points the containers at `host.docker.internal` for the database).

```bash
docker compose up -d          # from the repo root
docker compose logs -f taskiq-worker
docker compose logs -f taskiq-cv-worker

backend/Dockerfile builds a python:3.12-slim image whose default command is the Taskiq worker. backend/inbox/decoded_attachments is bind-mounted so the worker can read the attachments the API wrote.

---

Response conventions

Every handler wraps its body in the same try/except:

try:
    service=Email(session=session)
    data=await service.some_method(...)
    return JSONResponse(content={"data":data,"status_code":200})
except HTTPException:
    raise
except Exception as e:
    raise HTTPException(status_code=500,detail=str(e))

Shape	Response
List	{"data": [...], "total": <n>, "status_code": 200}
Single record	{"data": {...}, "total": 1, "status_code": 200}
Login / refresh	OAuth2 fields at the root, user under data
Error	FastAPI's {"detail": "..."} with the real status code

Serializers always str() UUIDs, .isoformat() datetimes, and never emit password.

---

Known gaps and gotchas

• DB_PORT must be set. db_setup.Settings evaluates int(os.getenv("DB_PORT")) at class definition time, so a missing value raises TypeError on import rather than a friendly config error.
• CORS is fully open (allow_origins=["*"] with credentials). Fine for development, needs tightening before production.
• /email/fetch and /inbox/fetch carry no permission guard. /email/fetch authenticates only against the upstream Email API token.
• .doc / .docx résumés are decoded and stored but not parsed. extract_resume_text handles PDFs only and reports no PDF attachment to extract for the rest.
• serialize_application returns null for ats_score / ats_band. The columns now exist on inbox_messages and the serializer reads them, but nothing ever writes them — the ATS persists to candidates instead. The Applications tab therefore shows no score even for candidates that have one. processing is still derived from message_read alone, so it is only ever "Read" or "Unread"; Imported/Processed/Rejected need processing_state to be written.
• ats_results is a dead table — declared, migrated, 0 rows, no reader and no writer. See The ATS scoring engine.
• Recruiter Performance is empty until a user holds the recruiter role. The query starts from Users JOIN Roles WHERE role_name = 'recruiter', so with no such user the widget renders empty no matter how much other data exists. Its hires column additionally needs inbox_messages.recruiter_id, for which there is no endpoint — the column is only ever set from created_by while a candidate is being created.
• application_stage_transitions rows created by migrations/manual/001 are timestamp- skewed if the file was run under a non-UTC psql session — the backfill writes the naive inbox.created_at into a timestamptz column. On the current dev database they sit 12 hours off, which skews the hires series of the hiring-trend chart and every time-to-hire average.
• The frontend's chart error state is misleading. Dashboard.jsx appends "This widget needs the analytics.view permission" to every error, including a 500, so a server fault reads as a permissions problem.
• Inbox.get_candidate_profile filters on cls.user.role_id, which is a relationship attribute rather than a joined column; the candidate-profile query needs a join before it behaves as intended.
• Attachment paths may be Windows absolutes written by the host API but read by a Linux worker. resolve_attachment_path normalizes separators and falls back to the basename under the mounted decoded_attachments directory.
• Read status is a one-way latch — see Background jobs.
• There is no test suite in backend/ at present.

---

Editing this codebase

Before changing anything here, read LLM_CONTEXT_PROMPT.md. It states the house style in full and is the reference used to keep new code indistinguishable from users/ and inbox/. The short version: mirror the neighbouring file, keep the layer duties intact, add no new layers, and do not reformat code you did not otherwise need to touch.

Adding an endpoint, in order:

1. Model accessor in models.py (if it touches the DB).
2. Service method in views.py.
3. serialize_* in serializers.py if the shape is new.
4. Route in app.py with the standard try/except and JSONResponse.
5. CurrentUser or Depends(require_permission(...)) if the route is protected.