ATS scores land on inbox tables + ats_results history; manual SQL auto-applies on boot

- inbox_messages.ats_score/ats_band written on every completed inbox score;
  inbox.ats_id always points at the current ats_results row
- ats_results now holds the supersede-chained history for BOTH inbox and
  upload scores (new candidate_id link, inbox_id nullable for uploads)
- migrations/manual/*.sql apply automatically at startup, tracked once per
  database in manual_migrations - developers just pull and boot
- 002_backfill_inbox_ats.sql backfills pre-existing scores
- Add Candidate modal: Matching-style role picker above the CV dropzone,
  and the CV is scored via /candidate/score after creation

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/README.md

@@ -82,9 +82,7 @@ See [Routing in code](#routing-in-code).
 
 | 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`) |
+| **Re-scoring overwrites the `candidates` row** | `upsert_candidate` updates in place on (`job_id`, `content_sha256`); the score history survives in `ats_results`, which every completed score (inbox and upload) appends to |
 | **`.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
@@ -192,7 +190,7 @@ backend/
 ├── 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)
+│   └── manual/               # one-shot SQL (enum labels, RBAC seed, backfills) — auto-applied at startup
 ├── LLM_CONTEXT_PROMPT.md     # house-style prompt to paste into an LLM before editing
 │
 ├── users/                    # accounts, login, signup, RBAC enforcement
@@ -355,7 +353,7 @@ indexes, constraints and foreign keys. `SQLModel.metadata` is pointed at `Base.m
 | `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` | `id`, `user_id` → `users.id`, `message_id` → `inbox_messages.id`, `alert_id`, `ats_id` → `ats_results.id` | Join table linking a candidate to a message. `ats_id` always points at the CURRENT `ats_results` row, repointed on every completed inbox score |
 | `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` | |
@@ -376,12 +374,13 @@ Nine tables landed together with `analytics/`, `offer/`, `job/assignment/`, `job
 | `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](#known-gaps-and-gotchas) |
+| `ats_results` *(inbox)* | `id`, `inbox_id` (nullable), `candidate_id` → `candidates.id`, `job_post_id`, `overall_score`, `band`, `is_current`, `superseded_by_id`, `model_name`, `computed_at` | Score history for EVERY completed score. Inbox scores chain per application (`inbox_id`, via `_sync_inbox_ats`); upload scores chain per `candidates` row (`inbox_id` NULL, via `_sync_upload_ats`). The previous current row is superseded (`is_current=false`, `superseded_by_id`); `migrations/manual/002` backfills pre-existing scores |
 
 `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.
+`recruiter_id`, `is_duplicate`, `source_channel_id`, `processing_state`. `ats_score` and
+`ats_band` are written by `CandidateScoring._sync_inbox_ats` on every completed inbox-sourced
+score (assigned-job score wins; bands: ≥82 Strong Match, ≥65 Potential Match, else Weak Match)
+and read by `serialize_application` on the Applications tab.
 
 `application_status` is a `str` enum, extended by `migrations/manual/001`: `PROCESS`,
 `PENDING`, `APPROVED`, `REJECTED`, `ONHOLD`, `CLOSED`, `SCREENING`, `ASSESSMENT`, `INTERVIEW`,
@@ -647,9 +646,10 @@ this database; the backend wraps it and owns all persistence.
 - `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-is-missing).
+So the scoring is agentic, but the output is fully relational. Inbox-sourced scores are
+additionally denormalised by `_sync_inbox_ats` onto `inbox_messages.ats_score` / `ats_band`
+and appended to `ats_results` as a supersede-chained history — see
+[where the values land](#where-the-values-land).
 
 ### What the system gives the engine
 
@@ -718,6 +718,26 @@ basename like `resume.pdf`. Per-file problems become persisted `status="failed"`
 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).
 
+**Every completed score also lands in `ats_results`.** Upload-sourced scores go through
+`_sync_upload_ats`: `inbox_id` stays NULL (there is no inbox application), the row links via
+`candidate_id`, and re-scoring the same bytes supersedes the previous current row — the chain
+is stable because `upsert_candidate` keeps the same `candidates.id` for the same job+file.
+
+**Inbox scores additionally land on the inbox tables** (`_sync_inbox_ats`, called once per
+message with its best completed score of the batch):
+
+- `inbox_messages.ats_score` / `ats_band` — the denormalised columns the Applications tab
+  reads. A score against the *assigned* job always wins them; a score against any other job
+  only lands while no completed assigned-job score exists (mirroring `_recommendation`).
+- `ats_results` — one history row per scoring event, `is_current=true`; the previous current
+  row flips to `is_current=false` with `superseded_by_id` pointing at its successor, and
+  `inbox.ats_id` is repointed at the new row so the current score is one direct id join away.
+  Requires the `inbox` join row (the sender must be linked to a `users` account); without it
+  only the denormalised columns are written.
+- A sync failure is rolled back and logged, never propagated — the `candidates` row is the
+  primary outcome and is already committed. Pre-existing scores are backfilled by
+  `migrations/manual/002_backfill_inbox_ats.sql`.
+
 ### Routing in code
 
 **Manual, from the UI:**
@@ -787,17 +807,9 @@ The engine reads its own settings through `app.core.config.get_settings()`, from
 
 ### 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.
+- **`candidates` itself keeps only the latest result.** `upsert_candidate` matches on
+  (`job_id`, `content_sha256`) and updates in place; the full score history lives in
+  `ats_results`, which both `_sync_inbox_ats` and `_sync_upload_ats` append to.
 - **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
@@ -990,11 +1002,17 @@ 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`):
+`migrations/manual/` and **apply themselves at startup**: after upgrade + autogenerate,
+`alembic_setup.run_manual_sql()` executes every `migrations/manual/*.sql` in filename order,
+once per database, tracked in the `manual_migrations` table (filename PK, `applied_at`) and
+serialised under the same advisory lock as the boot migration. Pulling the repo and booting
+the API is enough — no psql session needed. The files stay idempotent regardless, so a
+database where one was already run by hand simply absorbs one harmless re-run while it gets
+recorded.
 
 ```bash
-# After `python alembic_setup.py upgrade` has created the new tables/columns:
-psql "$DATABASE_URL" -f migrations/manual/001_dashboard_rbac_and_enum.sql
+# Equivalent manual run, only if ever needed:
+PGTZ=UTC 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
@@ -1002,15 +1020,12 @@ psql "$DATABASE_URL" -f migrations/manual/001_dashboard_rbac_and_enum.sql
 system roles that need the dashboard, seeds the eleven BRD `source_channels`, and
 backfills `source_channel_id` / stage-transition / requisition-status rows.
 
-> **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.
-> ```bash
-> PGTZ=UTC psql "$DATABASE_URL" -f migrations/manual/001_dashboard_rbac_and_enum.sql
-> ```
+> **Timezone.** The files write `NOW()` into columns of both kinds. The startup runner is
+> safe here: asyncpg leaves the server's UTC default alone. The trap is manual psql runs —
+> psql adopts the client OS timezone, storing 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. If you must run one by hand, set `PGTZ=UTC` as above.
 
 > **`migrations/versions/*.py` is effectively git-ignored.** `.gitignore` line 56 carries the
 > pattern `**_**_**.py`, which matches every generated revision filename

backend/alembic_setup.py

@@ -171,6 +171,7 @@ def config(connection: Connection | None = None) -> Config:
 
 
 VERSION_TABLE = "alembic_version"
+MANUAL_TABLE = "manual_migrations"
 
 
 def _include_object(obj: Any, name: str, type_: str, reflected: bool, compare_to: Any) -> bool:
@@ -178,7 +179,7 @@ def _include_object(obj: Any, name: str, type_: str, reflected: bool, compare_to
     s = get_settings()
     if type_ != "table":
         return True
-    if name == VERSION_TABLE:  # Alembic's own bookkeeping; never ours to alter
+    if name in (VERSION_TABLE, MANUAL_TABLE):  # migration bookkeeping; never ours to alter
         return False
     return not s.db_schemas or (obj.schema or s.db_default_schema) in s.db_schemas
 
@@ -258,6 +259,42 @@ async def autogenerate(message: str = "auto") -> str | None:
     return after if after != before else None
 
 
+async def run_manual_sql() -> None:
+    """Apply `migrations/manual/*.sql` once per database, in filename order.
+
+    This is what lets a developer just pull and boot: the one-shot data
+    migrations (enum labels, RBAC seed, backfills) apply themselves instead of
+    requiring a psql session. The files are written idempotent, but a tracking
+    table pins each to a single application per database; a file already run by
+    hand before this runner existed re-runs once (harmlessly) to get recorded.
+
+    Runs on the raw asyncpg connection: the files are multi-statement psql
+    batches, which the prepared-statement path cannot execute.
+    """
+    files = sorted(p for p in (MIGRATIONS / "manual").glob("*.sql") if p.is_file())
+    if not files:
+        return
+    schema = get_settings().db_default_schema
+    table = f'"{schema}".{MANUAL_TABLE}' if schema else MANUAL_TABLE
+    async with get_engine().connect() as conn:
+        raw = await conn.get_raw_connection()
+        driver = raw.driver_connection
+        await driver.execute(
+            f"CREATE TABLE IF NOT EXISTS {table} ("
+            " filename text PRIMARY KEY,"
+            " applied_at timestamptz NOT NULL DEFAULT now())"
+        )
+        applied = {r["filename"] for r in await driver.fetch(f"SELECT filename FROM {table}")}
+        for path in files:
+            if path.name in applied:
+                continue
+            # utf-8-sig: Windows editors save SQL with a BOM, which would
+            # otherwise reach Postgres glued onto the first statement.
+            await driver.execute(path.read_text(encoding="utf-8-sig"))
+            await driver.execute(f"INSERT INTO {table} (filename) VALUES ($1)", path.name)
+            logger.info("applied manual migration %s", path.name)
+
+
 @asynccontextmanager
 async def _lock() -> AsyncIterator[None]:
     """Advisory lock, so only one worker migrates when several boot at once."""
@@ -271,12 +308,13 @@ async def _lock() -> AsyncIterator[None]:
 
 
 async def migrate(*, autogen: bool | None = None, message: str = "auto") -> None:
-    """Apply pending revisions, then any fresh model drift, under the lock."""
+    """Apply pending revisions, fresh model drift, then manual SQL, under the lock."""
     should_autogen = get_settings().db_autogenerate if autogen is None else autogen
     async with _lock():
         await upgrade()
         if should_autogen and await autogenerate(message):
             await upgrade()
+        await run_manual_sql()
     logger.info("database at revision %s", await current())
 
 

backend/inbox/models.py

@@ -56,6 +56,12 @@ class Inbox(SQLModel, table=True):
     favorite: Optional[bool] = Field(default=False)
     rating: Optional[float] = Field(default=0.0)
 
+    # The CURRENT ats_results row for this application. Written together with the
+    # supersede chain in CandidateScoring._sync_inbox_ats: every completed inbox
+    # score inserts an ats_results row and repoints this at it, so the score is
+    # one direct id join away instead of a filter on ats_results.is_current.
+    ats_id: uuid.UUID | None = Field(default=None, foreign_key="ats_results.id")
+
     # selectin on one-to-many: joined would repeat the inbox row per child
     interviews: List["Interviews"] = Relationship(
         back_populates="inbox",
@@ -601,7 +607,10 @@ class AtsResults(SQLModel, table=True):
     __tablename__ = "ats_results"
 
     id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
-    inbox_id: int = Field(index=True, foreign_key="inbox.id")
+    # Inbox applications link here; NULL for upload-sourced scores, which have no
+    # inbox row. candidate_id links every score to the candidates row it scored.
+    inbox_id: int | None = Field(default=None, index=True, foreign_key="inbox.id")
+    candidate_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="candidates.id")
     job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id")
     overall_score: float = Field(default=0.0)
     band: str = Field(default="")
@@ -632,6 +641,18 @@ class AtsResults(SQLModel, table=True):
         )
         return result.scalars().first()
 
+    @classmethod
+    async def get_current_for_candidate(cls, session: AsyncSession, candidate_id):
+        """Current row for an upload-sourced score, chained per candidates row —
+        stable across re-scores because upsert_candidate keeps the same id for
+        the same (job, content_sha256)."""
+        result = await session.execute(
+            select(cls)
+            .where(cls.candidate_id == candidate_id, cls.is_current == True)  # noqa: E712
+            .order_by(cls.computed_at.desc())
+        )
+        return result.scalars().first()
+
     @classmethod
     async def insert_result(cls, session: AsyncSession, fields: dict):
         row = cls(**fields)

backend/job/candidate/views.py

@@ -12,7 +12,7 @@ from app.core.errors import ATSError,ErrorCode
 from app.models.scoring import CompletedCandidate
 from app.services.pdf import extract_resume,sanitize_filename
 from app.services.scoring import score_batch
-from inbox.models import Inbox_Messages,Inbox
+from inbox.models import Inbox_Messages,Inbox,AtsResults
 from job.candidate.models import Candidates
 from job.candidate.plugins import (
     FILE_NOT_FOUND,
@@ -407,6 +407,35 @@ class CandidateScoring:
             fields={**results_by_slot[slot],**common}
             rows.append(await Candidates.upsert_candidate(self.session,fields))
 
+        # Every completed score lands in ats_results. Inbox scores additionally
+        # denormalise onto inbox_messages / inbox (one sync per message: a
+        # multi-attachment mail keeps its best completed score); upload scores
+        # chain per candidates row instead — they have no inbox application.
+        if source_kind=="inbox":
+            best={}
+            for row in rows:
+                if row.status=="completed" and row.inbox_message_id:
+                    cur=best.get(row.inbox_message_id)
+                    if cur is None or (row.match_score or 0)>(cur.match_score or 0):
+                        best[row.inbox_message_id]=row
+            for message_id,row in best.items():
+                try:
+                    await self._sync_inbox_ats(message_id,job,row)
+                except Exception:
+                    # The candidates row is the primary outcome and is already
+                    # committed; a denorm failure must not fail the scoring call.
+                    await self.session.rollback()
+                    logger.exception("inbox ATS denorm failed for message %s",message_id)
+        else:
+            for row in rows:
+                if row.status!="completed":
+                    continue
+                try:
+                    await self._sync_upload_ats(job,row)
+                except Exception:
+                    await self.session.rollback()
+                    logger.exception("upload ATS history failed for candidate %s",row.id)
+
         # Leaderboard order: completed by score desc, failures last, stable.
         rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0))
         return [serialize_candidate(row) for row in rows]
@@ -437,6 +466,92 @@ class CandidateScoring:
             "summary_critique":None,
         }
 
+    async def _sync_inbox_ats(self,message_id,job,row):
+        """Land a completed score on the inbox tables (README "What is missing").
+
+        inbox_messages.ats_score / ats_band are what serialize_application renders
+        on the Applications tab; ats_results keeps the per-application history that
+        candidates' in-place upsert cannot. Precedence mirrors _recommendation: a
+        score against the assigned job always wins the denormalised columns, any
+        other job's score only lands while no completed assigned-job score exists.
+        The history row is appended regardless — it records the scoring event.
+        """
+        msg=await Inbox_Messages.get_inbox_message_by_id(self.session,message_id)
+        if msg is None:
+            return
+        band=CandidateView._recommendation(row.match_score) or ""
+
+        denorm=True
+        assigned=msg.assigned_job_post_id
+        if assigned and str(assigned)!=str(job.id):
+            outranked=await self.session.execute(
+                select(Candidates).where(
+                    Candidates.inbox_message_id==msg.id,
+                    Candidates.job_id==assigned,
+                    Candidates.status=="completed",
+                )
+            )
+            denorm=outranked.scalars().first() is None
+        if denorm:
+            msg.ats_score=float(row.match_score)
+            msg.ats_band=band
+            self.session.add(msg)
+
+        # ats_results hangs off the inbox JOIN row (int PK), which only exists once
+        # the sender is linked to a users account; without it there is no history row.
+        link=await Inbox.get_inbox_by_message_id(self.session,message_id)
+        if link is not None:
+            prev=await AtsResults.get_current_for_inbox(self.session,link.id)
+            entry=AtsResults(
+                inbox_id=link.id,
+                candidate_id=row.id,
+                job_post_id=job.id,
+                overall_score=float(row.match_score),
+                band=band,
+                model_name=row.model,
+                is_current=True,
+            )
+            self.session.add(entry)
+            # Flush the INSERT before touching prev/link: with no relationship()
+            # edge the unit of work emits the UPDATEs first, and both FKs
+            # (superseded_by_id, inbox.ats_id) reject a pointer to a row that is
+            # not inserted yet.
+            await self.session.flush()
+            if prev is not None:
+                prev.is_current=False
+                prev.superseded_by_id=entry.id
+                self.session.add(prev)
+            # inbox.ats_id always points at the CURRENT score row.
+            link.ats_id=entry.id
+            link.updated_at=datetime.now(timezone.utc)
+            self.session.add(link)
+        await self.session.commit()
+
+    async def _sync_upload_ats(self,job,row):
+        """History row for an upload-sourced score — no inbox application exists,
+        so inbox_id stays NULL and the supersede chain runs per candidates row
+        (stable across re-scores: upsert keeps the id for the same job+bytes)."""
+        band=CandidateView._recommendation(row.match_score) or ""
+        prev=await AtsResults.get_current_for_candidate(self.session,row.id)
+        entry=AtsResults(
+            inbox_id=None,
+            candidate_id=row.id,
+            job_post_id=job.id,
+            overall_score=float(row.match_score),
+            band=band,
+            model_name=row.model,
+            is_current=True,
+        )
+        self.session.add(entry)
+        # Same flush-before-pointing rule as the inbox path: the FK on
+        # superseded_by_id must see the new row inserted first.
+        await self.session.flush()
+        if prev is not None:
+            prev.is_current=False
+            prev.superseded_by_id=entry.id
+            self.session.add(prev)
+        await self.session.commit()
+
 
 class CandidateView:
     def __init__(self,session:AsyncSession):

backend/migrations/manual/002_backfill_inbox_ats.sql

@@ -0,0 +1,143 @@
+-- 002_backfill_inbox_ats.sql
+-- Manual one-shot: backfill the inbox-side ATS columns that gained a writer in
+-- job/candidate/views.py::CandidateScoring._sync_inbox_ats. Before that change,
+-- inbox_messages.ats_score / ats_band were never written and ats_results held
+-- 0 rows, so messages scored earlier show null on the Applications tab.
+--
+-- Idempotent: the UPDATE only touches rows still missing a score, and the
+-- INSERT skips any application that already has an ats_results row.
+-- Applied automatically at startup by alembic_setup.run_manual_sql() once the
+-- schema is at head; recorded in manual_migrations. Safe to re-run by hand.
+
+-- Winning score per message, mirroring CandidateView._recommendation precedence:
+-- the completed score against the ASSIGNED job wins, else the newest completed.
+WITH ranked AS (
+    SELECT
+        c.inbox_message_id,
+        c.job_id,
+        c.match_score,
+        c.model,
+        c.updated_at,
+        ROW_NUMBER() OVER (
+            PARTITION BY c.inbox_message_id
+            ORDER BY (c.job_id = m.assigned_job_post_id) DESC NULLS LAST,
+                     c.updated_at DESC
+        ) AS rn
+    FROM app.candidates c
+    JOIN app.inbox_messages m ON m.id = c.inbox_message_id
+    WHERE c.status = 'completed'
+      AND c.inbox_message_id IS NOT NULL
+)
+UPDATE app.inbox_messages m
+SET ats_score = r.match_score,
+    ats_band = CASE
+        WHEN r.match_score >= 82 THEN 'Strong Match'
+        WHEN r.match_score >= 65 THEN 'Potential Match'
+        ELSE 'Weak Match'
+    END
+FROM ranked r
+WHERE r.rn = 1
+  AND m.id = r.inbox_message_id
+  AND m.ats_score IS NULL;
+
+-- One current ats_results row per already-scored application. computed_at takes
+-- the candidates row's timestamp so the history reflects when the score happened.
+-- inbox can hold several join rows per message; DISTINCT ON keeps the newest.
+WITH ranked AS (
+    SELECT
+        c.inbox_message_id,
+        c.job_id,
+        c.match_score,
+        c.model,
+        c.updated_at,
+        ROW_NUMBER() OVER (
+            PARTITION BY c.inbox_message_id
+            ORDER BY (c.job_id = m.assigned_job_post_id) DESC NULLS LAST,
+                     c.updated_at DESC
+        ) AS rn
+    FROM app.candidates c
+    JOIN app.inbox_messages m ON m.id = c.inbox_message_id
+    WHERE c.status = 'completed'
+      AND c.inbox_message_id IS NOT NULL
+),
+links AS (
+    SELECT DISTINCT ON (message_id) message_id, id AS inbox_id
+    FROM app.inbox
+    ORDER BY message_id, created_at DESC
+)
+INSERT INTO app.ats_results
+    (id, inbox_id, job_post_id, overall_score, band, is_current,
+     superseded_by_id, model_name, computed_at, created_at)
+SELECT
+    gen_random_uuid(),
+    l.inbox_id,
+    r.job_id,
+    r.match_score,
+    CASE
+        WHEN r.match_score >= 82 THEN 'Strong Match'
+        WHEN r.match_score >= 65 THEN 'Potential Match'
+        ELSE 'Weak Match'
+    END,
+    true,
+    NULL,
+    r.model,
+    r.updated_at,
+    NOW()
+FROM ranked r
+JOIN links l ON l.message_id = r.inbox_message_id
+WHERE r.rn = 1
+  AND NOT EXISTS (
+      SELECT 1 FROM app.ats_results a WHERE a.inbox_id = l.inbox_id
+  );
+
+-- inbox.ats_id -> the CURRENT ats_results row for that application, so the
+-- score is one direct id join away. Newest current row wins if several exist.
+UPDATE app.inbox i
+SET ats_id = a.id,
+    updated_at = NOW()
+FROM (
+    SELECT DISTINCT ON (inbox_id) inbox_id, id
+    FROM app.ats_results
+    WHERE is_current AND inbox_id IS NOT NULL
+    ORDER BY inbox_id, computed_at DESC
+) a
+WHERE a.inbox_id = i.id
+  AND i.ats_id IS DISTINCT FROM a.id;
+
+-- Upload-sourced scores get history rows too: inbox_id stays NULL (no inbox
+-- application exists) and the row links via candidate_id instead.
+INSERT INTO app.ats_results
+    (id, inbox_id, candidate_id, job_post_id, overall_score, band, is_current,
+     superseded_by_id, model_name, computed_at, created_at)
+SELECT
+    gen_random_uuid(),
+    NULL,
+    c.id,
+    c.job_id,
+    c.match_score,
+    CASE
+        WHEN c.match_score >= 82 THEN 'Strong Match'
+        WHEN c.match_score >= 65 THEN 'Potential Match'
+        ELSE 'Weak Match'
+    END,
+    true,
+    NULL,
+    c.model,
+    c.updated_at,
+    NOW()
+FROM app.candidates c
+WHERE c.status = 'completed'
+  AND c.source = 'upload'
+  AND NOT EXISTS (
+      SELECT 1 FROM app.ats_results a WHERE a.candidate_id = c.id
+  );
+
+-- Older inbox history rows predate candidate_id; link the current ones back to
+-- the candidates row that produced them.
+UPDATE app.ats_results a
+SET candidate_id = c.id
+FROM app.inbox i
+JOIN app.candidates c ON c.inbox_message_id = i.message_id AND c.status = 'completed'
+WHERE a.inbox_id = i.id
+  AND a.candidate_id IS NULL
+  AND c.job_id = a.job_post_id;

frontend/src/screens/Candidates.jsx

@@ -5,7 +5,8 @@
    toCandidateView mapper. Facets, columns and actions that had no backing
    column (stage, recruiter, notice period, favourites…) are gone rather than
    rendered as placeholders — the Inbox screen set that precedent. Adding
-   candidates happens through CV Import (real scoring), not a manual form.
+   candidates happens through CV Import or the Add Candidate modal below —
+   both run the CV through the same persisted ATS scoring pipeline.
    ============================================================ */
 
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
@@ -201,6 +202,26 @@ export default function Candidates() {
     setAtsFor(c)
   }
 
+  /* After a manual add, the CV goes through the same persisted scoring pipeline
+     CV Import and the profile ATS match use (POST /candidate/score): the score
+     lands in the scored `candidates` table, and re-uploading the same bytes
+     against the same job updates that row rather than duplicating it. The
+     mutation lives here, not in AddCandidate, because the modal closes on save
+     and an unmounted component's mutation callbacks never fire. */
+  const scoreCv = useMutation({
+    mutationFn: ({ jobPostId, file }) => candidatesApi.scoreUploads(jobPostId, [file]),
+    onSuccess: (res) => {
+      const row = Array.isArray(res?.data) ? res.data[0] : null
+      if (row?.status === 'completed') {
+        toast(`CV scored ${row.match_score}/100 against the applied job — saved to the pool`, 'success')
+      } else {
+        toast(`CV could not be scored${row?.error_code ? ` — ${row.error_code}` : ''}`, 'warning')
+      }
+      qc.invalidateQueries({ queryKey: qk.candidates.all() })
+    },
+    onError: (err) => toast(friendlyAuthError(err, 'Candidate saved, but CV scoring failed'), 'error'),
+  })
+
   return (
     <div className="page">
       <div className="page-head">
@@ -377,9 +398,10 @@ export default function Candidates() {
       {adding && (
         <AddCandidate
           onClose={() => setAdding(false)}
-          onSave={() => {
+          onSave={({ jobPostId, file } = {}) => {
             setAdding(false)
             toast('Candidate added to pipeline', 'success')
+            if (jobPostId && file) scoreCv.mutate({ jobPostId, file })
           }}
           onInvalid={() => toast('Please fix the highlighted fields', 'error')}
         />
@@ -477,6 +499,62 @@ export function AtsMatch({ candidate: c, jobTitle, onClose, onProfile }) {
   )
 }
 
+/**
+ * One selectable role, drawn the way Job Matching draws its suggested roles
+ * (Matching.jsx::JobCard) minus the AI-rank tag and resume highlighting — the
+ * CV is parsed server-side after submit, so there is no extracted text to
+ * light requirement chips against yet.
+ */
+function RoleCard({ post, selected, onSelect, disabled }) {
+  const meta = [
+    post?.employment_type,
+    post?.location,
+    post?.experience_min != null || post?.experience_max != null
+      ? `${post?.experience_min ?? '?'}–${post?.experience_max ?? '?'} yrs`
+      : null,
+  ].filter(Boolean).join(' · ')
+
+  return (
+    <div
+      role="radio"
+      aria-checked={selected}
+      tabIndex={0}
+      className="list-row"
+      onClick={() => !disabled && onSelect(String(post.id))}
+      onKeyDown={(e) => {
+        if (disabled) return
+        if (e.key === 'Enter' || e.key === ' ') {
+          e.preventDefault()
+          onSelect(String(post.id))
+        }
+      }}
+      style={{
+        cursor: disabled ? 'default' : 'pointer',
+        borderColor: selected ? 'var(--primary)' : undefined,
+        boxShadow: selected ? 'var(--ring)' : undefined,
+        marginBottom: 8,
+        alignItems: 'flex-start',
+      }}
+    >
+      <div className="lr-main" style={{ minWidth: 0 }}>
+        <div className="flex items-center gap-8" style={{ marginBottom: 4, flexWrap: 'wrap' }}>
+          <div className="lr-title">{post.title}</div>
+          <Badge>{post.status || 'draft'}</Badge>
+          {selected && <Icon name="check-circle" />}
+        </div>
+        {meta && <div className="cell-sub">{meta}</div>}
+        {(post.requirements || []).length > 0 && (
+          <div className="k-tags" style={{ marginTop: 8 }}>
+            {(post.requirements || []).slice(0, 8).map((req) => (
+              <span key={req} className="tag">{req}</span>
+            ))}
+          </div>
+        )}
+      </div>
+    </div>
+  )
+}
+
 /**
  * Add Candidate — the only writer on this screen that reaches the server.
  *
@@ -492,6 +570,11 @@ export function AtsMatch({ candidate: c, jobTitle, onClose, onProfile }) {
  * Manual rows do not pass through `inbox`, so /candidate/fetch may not surface
  * them immediately; the save still invalidates the candidates query so the
  * live-backed screens refetch and pick the row up once an application links it.
+ *
+ * On success the CV and job go back to the parent (onSave), which scores the
+ * file against that job via POST /candidate/score. The Matching-style role
+ * cards above the dropzone are that "what to score against" choice — which is
+ * why the picker sits with the CV rather than among the identity fields.
  */
 function AddCandidate({ onClose, onSave, onInvalid }) {
   const { toast } = useToast()
@@ -520,14 +603,28 @@ function AddCandidate({ onClose, onSave, onInvalid }) {
   // field from an effect would either loop or need a ref to guard it.
   const jobPostId = form.values.job || (posts[0] ? String(posts[0].id) : '')
 
+  const [roleSearch, setRoleSearch] = useState('')
+  // Client-side filter: the posts are already fetched, and a PickRoleModal-style
+  // server search would refetch on every keystroke for the same rows.
+  const visiblePosts = useMemo(() => {
+    const needle = roleSearch.trim().toLowerCase()
+    if (!needle) return posts
+    return posts.filter((p) => (
+      (p.title || '').toLowerCase().includes(needle)
+      || (p.location || '').toLowerCase().includes(needle)
+    ))
+  }, [posts, roleSearch])
+
   const create = useMutation({
     mutationFn: (vars) => candidatesApi.createManual(vars),
     onError: (err) => toast(friendlyAuthError(err, 'Could not add the candidate.'), 'error'),
-    onSuccess: () => {
+    onSuccess: (_res, vars) => {
       // The new user_id lands in /candidate/fetch's join the moment an
       // application exists for them, so let the live-backed screens refetch.
       qc.invalidateQueries({ queryKey: qk.candidates.all() })
-      onSave()
+      // Hand the file and job back so the parent can score the CV — from the
+      // mutate vars, not local state, so a mid-flight field edit cannot skew it.
+      onSave({ jobPostId: vars.jobPostId, file: vars.file })
     },
   })
 
@@ -610,24 +707,6 @@ function AddCandidate({ onClose, onSave, onInvalid }) {
             <FieldError>{form.errors.email}</FieldError>
           </div>
           <div className="form-field"><label>Phone</label><input {...field('phone')} placeholder="+1 (555) 000-0000" /></div>
-          <div className="form-field">
-            <label>Applied Job <span className="req">*</span></label>
-            <select
-              value={jobPostId}
-              onChange={(e) => form.setField('job', e.target.value)}
-              className={form.errors.job ? 'err' : ''}
-              disabled={postsQuery.isPending || !posts.length}
-            >
-              {postsQuery.isPending && <option value="">Loading job posts…</option>}
-              {!postsQuery.isPending && !posts.length && (
-                <option value="">{postsQuery.isError ? 'Could not load job posts' : 'No active job posts'}</option>
-              )}
-              {posts.map((p) => (
-                <option key={p.id} value={String(p.id)}>{p.title}</option>
-              ))}
-            </select>
-            <FieldError>{form.errors.job}</FieldError>
-          </div>
           <div className="form-field"><label>Experience (years)</label><input type="number" {...field('experience')} /></div>
           <div className="form-field"><label>Current Company</label><input {...field('company')} placeholder="Acme Inc." /></div>
           <div className="form-field"><label>Current Position</label><input {...field('position')} placeholder="Senior Merchandiser" /></div>
@@ -658,7 +737,57 @@ function AddCandidate({ onClose, onSave, onInvalid }) {
           </div>
         </div>
 
-        {/* .req is scoped to `.form-field label .req`, so tint it here. */}
+        {/* Role selection sits directly above the CV because it is what the CV
+            gets scored against — same card UI as Job Matching's role list.
+            (.req is scoped to `.form-field label .req`, so tint it here.) */}
+        <div className="form-section-title">
+          Applied Job <span className="req" style={{ color: 'var(--danger)' }}>*</span>
+        </div>
+        {posts.length > 0 && (
+          <div className="toolbar-search" style={{ maxWidth: 'none', marginBottom: 10 }}>
+            <Icon name="search" />
+            <input
+              value={roleSearch}
+              onChange={(e) => setRoleSearch(e.target.value)}
+              placeholder="Search title or location…"
+            />
+          </div>
+        )}
+        {postsQuery.isPending && (
+          <EmptyState icon="briefcase" title="Loading roles…">Fetching open job posts.</EmptyState>
+        )}
+        {postsQuery.isError && (
+          <EmptyState icon="alert" title="Couldn’t load roles">
+            {friendlyAuthError(postsQuery.error, 'Request failed')}
+          </EmptyState>
+        )}
+        {postsQuery.isSuccess && posts.length === 0 && (
+          <EmptyState icon="briefcase" title="No active job posts">
+            The candidate is saved without a role link — create a job post first to score CVs.
+          </EmptyState>
+        )}
+        {visiblePosts.length > 0 && (
+          <div
+            role="radiogroup"
+            aria-label="Applied job"
+            style={{ maxHeight: 264, overflowY: 'auto', paddingRight: 2 }}
+          >
+            {visiblePosts.map((p) => (
+              <RoleCard
+                key={p.id}
+                post={p}
+                selected={String(p.id) === jobPostId}
+                onSelect={(id) => form.setField('job', id)}
+                disabled={create.isPending}
+              />
+            ))}
+          </div>
+        )}
+        {postsQuery.isSuccess && posts.length > 0 && visiblePosts.length === 0 && (
+          <EmptyState icon="briefcase" title="No roles found">Try a different search.</EmptyState>
+        )}
+        <FieldError>{form.errors.job}</FieldError>
+
         <div className="form-section-title">
           CV / Resume <span className="req" style={{ color: 'var(--danger)' }}>*</span>
         </div>