fix(gate): anchor current state on the project's main line, not the newest scan

apps/backend/api/v1/policy_gate.py

@@ -79,6 +79,7 @@
     post_pr_comment,
 )
 from services.scan_resolution import SnapshotScanNotFound, resolve_snapshot_scan_id
+from services.scan_service import normalize_ref
 from services.upgrade_recommendation import (
     FindingSignal,
     priority_rank,
@@ -233,7 +234,7 @@ def _build_response_body(result: GateResult) -> GateResultResponse:
 @router.get(
     "/projects/{project_id}/gate-result",
     response_model=GateResultResponse,
-    summary="Evaluate the build-gate verdict for the project's latest succeeded scan",
+    summary="Evaluate the build-gate verdict for the project's current-state scan",
 )
 async def get_gate_result_endpoint(
     request: Request,
@@ -248,6 +249,18 @@ async def get_gate_result_endpoint(
             "404. Omit for the default latest-succeeded behaviour (the CI contract)."
         ),
     ),
+    ref: str | None = Query(
+        default=None,
+        max_length=255,
+        description=(
+            "Optional branch anchor: evaluate against the newest succeeded scan "
+            "of this normalized git ref (``main``, ``pr-12``) instead of the "
+            "project's main line. A CI job should pass its own ref so its verdict "
+            "cannot be decided by a scan of a different branch. Exact: a ref with "
+            "no succeeded scan yields the no-signal pass, never another branch's "
+            "findings. Ignored when ``scan_id`` is given, which is more specific."
+        ),
+    ),
     session: AsyncSession = Depends(get_db),
     actor: CurrentUser = Depends(_principal_from_jwt_or_api_key),
 ) -> Response:
@@ -261,7 +274,9 @@ async def get_gate_result_endpoint(
     # pinned scan_id (cross-project / non-succeeded / nonexistent) → existence-
     # hide 404. ``None`` → latest succeeded (unchanged CI default).
     try:
-        resolved_scan_id = await resolve_snapshot_scan_id(session, project_id, scan_id)
+        resolved_scan_id = await resolve_snapshot_scan_id(
+            session, project_id, scan_id, ref=normalize_ref(ref)
+        )
     except SnapshotScanNotFound:
         return problem_response(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -270,7 +285,12 @@ async def get_gate_result_endpoint(
             instance=request.url.path,
         )
 
-    gate_result = await evaluate_gate(session, project_id, scan_id=resolved_scan_id)
+    # Pass the ref as well: `resolved_scan_id` is None both when nothing was
+    # pinned and when the named branch has no succeeded scan, and only the ref
+    # tells evaluate_gate which of the two it is.
+    gate_result = await evaluate_gate(
+        session, project_id, scan_id=resolved_scan_id, ref=normalize_ref(ref)
+    )
     body = _build_response_body(gate_result)
     return Response(
         content=body.model_dump_json(),

apps/backend/services/policy_gate.py

@@ -629,6 +629,7 @@ async def evaluate_gate(
     project_id: uuid.UUID,
     *,
     scan_id: uuid.UUID | None = None,
+    ref: str | None = None,
 ) -> GateResult:
     """Compute the build-gate verdict for ``project_id``.
 
@@ -643,9 +644,16 @@ async def evaluate_gate(
     :func:`services.scan_resolution.resolve_snapshot_scan_id` before calling).
     When ``None`` (the default, used by CI's "latest verdict" contract) the
     latest succeeded scan is resolved here exactly as before.
+
+    ``ref`` scopes that internal resolution to one branch, and must be passed
+    whenever the caller resolved with a ref of its own. The caller's ``None``
+    means "not pinned", but a ref that has never had a succeeded scan also
+    resolves to ``None`` — without this parameter the two collapse and the gate
+    would answer a branch-scoped request with the main line's findings, which is
+    the exact confusion the ref anchor exists to prevent.
     """
     if scan_id is None:
-        scan_id = await _latest_succeeded_scan_id(session, project_id)
+        scan_id = await _latest_succeeded_scan_id(session, project_id, ref=ref)
     evaluated_at = datetime.now(tz=UTC)
 
     # Read the EPSS threshold once per evaluation (None when the gate is

apps/backend/services/scan_resolution.py

@@ -38,14 +38,38 @@
 
 import uuid
 
-from sqlalchemy import String, cast, select
+from sqlalchemy import String, cast, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql import Select
 from sqlalchemy.sql.elements import ColumnElement
 
 from models import Project, Scan
 
 
+def _project_main_line_ref() -> ColumnElement[str]:
+    """SQL for the ref a project's *current state* should be read from.
+
+    ``projects.default_branch`` is the declared main line, but it is unset on
+    most rows (the create form does not ask for it, and the source pipeline
+    clones the remote's own HEAD rather than passing ``--branch``). Falling back
+    to ``main`` makes the common repo work without configuration; a project
+    whose main line is ``master`` or ``trunk`` simply matches nothing and keeps
+    the old project-wide behaviour, so the guess can only help.
+    """
+    return func.coalesce(func.nullif(func.btrim(Project.default_branch), ""), "main")
+
+
+def _on_main_line() -> ColumnElement[bool]:
+    """SQL predicate: this scan targeted the project's main line.
+
+    ``IS NOT DISTINCT FROM`` rather than ``=`` because ``Scan.ref`` is NULL for
+    ad-hoc scans, and ``NULL = 'main'`` is NULL — which sorts FIRST under
+    ``ORDER BY ... DESC`` in Postgres and would make ref-less scans outrank the
+    main line exactly when we are trying to prefer it.
+    """
+    return Scan.ref.is_not_distinct_from(_project_main_line_ref())
+
+
 class SnapshotScanNotFound(Exception):
     """Raised when an explicit ``scan_id`` snapshot anchor is not resolvable.
 
@@ -68,8 +92,26 @@ class SnapshotScanNotFound(Exception):
 async def latest_succeeded_scan_id(
     session: AsyncSession,
     project_id: uuid.UUID,
+    ref: str | None = None,
 ) -> uuid.UUID | None:
-    """Return the ID of the project's most recent ``status='succeeded'`` scan, or None.
+    """Return the ID of the project's current-state ``status='succeeded'`` scan.
+
+    "Current state" prefers the project's MAIN LINE. Ordering by
+    :func:`_on_main_line` first and recency second means a project that scans
+    several branches resolves to its main line's newest snapshot, while a
+    project with no main-line scan falls through to the newest snapshot of any
+    branch — the pre-existing behaviour, unchanged.
+
+    Without that preference the anchor was purely "newest succeeded scan of this
+    project", so a project wired to CI on both ``main`` and ``release/1.x``
+    flipped its Overview, badges, and — because the build gate resolves through
+    here too — its CI verdict, to whichever branch happened to finish last.
+    ``main``'s pipeline could be blocked by a release branch's critical CVE.
+
+    Pass *ref* to anchor on a specific branch instead. That is exact: a branch
+    with no succeeded scan returns ``None`` (the caller's "empty 200" path)
+    rather than silently falling back to another branch's findings, because a
+    caller that named a branch wants that branch or nothing.
 
     We deliberately do NOT use ``Project.latest_scan_id`` here: that pointer
     reflects the last *attempted* scan, so a successful scan whose last attempt
@@ -84,11 +126,14 @@ async def latest_succeeded_scan_id(
     """
     stmt = (
         select(Scan.id)
+        .join(Project, Project.id == Scan.project_id)
         .where(Scan.project_id == project_id)
         .where(cast(Scan.status, String) == "succeeded")
-        .order_by(Scan.created_at.desc(), Scan.id.desc())
+        .order_by(_on_main_line().desc(), Scan.created_at.desc(), Scan.id.desc())
         .limit(1)
     )
+    if ref is not None:
+        stmt = stmt.where(Scan.ref == ref)
     result = await session.execute(stmt)
     return result.scalar_one_or_none()
 
@@ -128,7 +173,15 @@ def latest_succeeded_scan_select(
         .join(Project, Project.id == Scan.project_id)
         .distinct(Scan.project_id)
         .where(cast(Scan.status, String) == "succeeded")
-        .order_by(Scan.project_id, Scan.created_at.desc(), Scan.id.desc())
+        .order_by(
+            Scan.project_id,
+            # Same main-line preference as the single-project resolver — the
+            # two must agree or the inventory would list a component the owning
+            # project's Components tab does not show.
+            _on_main_line().desc(),
+            Scan.created_at.desc(),
+            Scan.id.desc(),
+        )
     )
     if project_filter is not None:
         stmt = stmt.where(project_filter)
@@ -139,6 +192,7 @@ async def resolve_snapshot_scan_id(
     session: AsyncSession,
     project_id: uuid.UUID,
     scan_id: uuid.UUID | None,
+    ref: str | None = None,
 ) -> uuid.UUID | None:
     """Resolve which scan a detail-read surface should anchor on (feature #28).
 
@@ -170,7 +224,7 @@ async def resolve_snapshot_scan_id(
     invoking the resolver (same contract as :func:`latest_succeeded_scan_id`).
     """
     if scan_id is None:
-        return await latest_succeeded_scan_id(session, project_id)
+        return await latest_succeeded_scan_id(session, project_id, ref=ref)
 
     # Validate ownership AND succeeded status in one statement. We deliberately
     # do NOT split "wrong project" from "not succeeded": both collapse to the

apps/backend/tests/integration/test_release_snapshots_api.py

@@ -173,6 +173,7 @@ async def _seed_succeeded_scan(
     n_high: int = 0,
     forbidden_license: bool = False,
     release: str | None = None,
+    ref: str | None = None,
     status: str = "succeeded",
 ) -> uuid.UUID:
     """Create a scan (default succeeded) with components + findings keyed to it.
@@ -193,6 +194,7 @@ async def _seed_succeeded_scan(
             progress_percent=100 if status == "succeeded" else 0,
             scan_metadata=metadata,
             created_at=created_at,
+            ref=ref,
         )
         session.add(scan)
         await session.commit()
@@ -262,11 +264,16 @@ async def _seed_team_with_user(client: AsyncClient, *, role: str = "developer"):
     return team, user
 
 
-async def _seed_empty_project(client: AsyncClient, *, team_id: uuid.UUID) -> uuid.UUID:
+async def _seed_empty_project(
+    client: AsyncClient, *, team_id: uuid.UUID, default_branch: str | None = None
+) -> uuid.UUID:
     factory = await _factory(client)
     async with factory() as session:
         team = (await session.execute(select(Team).where(Team.id == team_id))).scalar_one()
         project = await make_project(session, team=team)
+        if default_branch is not None:
+            project.default_branch = default_branch
+            await session.commit()
         return project.id
 
 
@@ -637,6 +644,153 @@ async def test_omitting_scan_id_returns_latest_succeeded(client) -> None:
     assert response.json()["last_succeeded_scan_at"] is not None
 
 
+# ---------------------------------------------------------------------------
+# Current-state anchor follows the project's main line
+# ---------------------------------------------------------------------------
+
+
+async def _seed_two_branch_project(client: AsyncClient, *, default_branch: str | None):
+    """Main line scanned first (2 critical), another branch scanned LAST (1 high).
+
+    The release branch finishing last is the whole point: under a purely
+    recency-based anchor it would decide the project's current state.
+    """
+    team, user = await _seed_team_with_user(client)
+    project_id = await _seed_empty_project(
+        client, team_id=team.id, default_branch=default_branch
+    )
+    base = datetime(2026, 5, 20, tzinfo=UTC)
+    main_scan = await _seed_succeeded_scan(
+        client, project_id=project_id, created_at=base, n_critical=2, ref="main"
+    )
+    release_scan = await _seed_succeeded_scan(
+        client,
+        project_id=project_id,
+        created_at=base + timedelta(days=2),
+        n_high=1,
+        ref="release/1.x",
+    )
+    return user, project_id, main_scan, release_scan
+
+
+async def test_anchor_prefers_main_line_over_a_newer_other_branch(client) -> None:
+    user, project_id, _main_scan, _release_scan = await _seed_two_branch_project(
+        client, default_branch="main"
+    )
+    headers = _bearer_for(user)
+
+    overview = await client.get(f"/v1/projects/{project_id}/overview", headers=headers)
+    assert overview.status_code == 200, overview.text
+    # main's 2 critical, not release/1.x's 1 high, even though the latter is newer.
+    assert overview.json()["severity_distribution"]["critical"] == 2
+    assert overview.json()["severity_distribution"]["high"] == 0
+
+
+async def test_gate_verdict_is_not_decided_by_another_branch(client) -> None:
+    # The defect that motivated this: main's CI asks for its verdict and gets
+    # the release branch's, because that branch scanned more recently.
+    user, project_id, main_scan, release_scan = await _seed_two_branch_project(
+        client, default_branch="main"
+    )
+    headers = _bearer_for(user)
+
+    default = await client.get(f"/v1/projects/{project_id}/gate-result", headers=headers)
+    assert default.status_code == 200, default.text
+    assert default.json()["scan_id"] == str(main_scan)
+    assert default.json()["critical_cve_count"] == 2
+
+    # A CI job on the release branch names its own ref and gets its own verdict.
+    pinned = await client.get(
+        f"/v1/projects/{project_id}/gate-result",
+        headers=headers,
+        params={"ref": "release/1.x"},
+    )
+    assert pinned.status_code == 200, pinned.text
+    assert pinned.json()["scan_id"] == str(release_scan)
+    assert pinned.json()["critical_cve_count"] == 0
+
+
+async def test_gate_ref_accepts_a_fully_qualified_ref(client) -> None:
+    # CI passes $GITHUB_REF; the endpoint must normalize it the same way the
+    # scan-create path did, or the branch would never match its own scans.
+    user, project_id, main_scan, _release = await _seed_two_branch_project(
+        client, default_branch="release/1.x"
+    )
+    headers = _bearer_for(user)
+
+    response = await client.get(
+        f"/v1/projects/{project_id}/gate-result",
+        headers=headers,
+        params={"ref": "refs/heads/main"},
+    )
+    assert response.status_code == 200, response.text
+    assert response.json()["scan_id"] == str(main_scan)
+
+
+async def test_anchor_falls_back_when_no_scan_is_on_the_main_line(client) -> None:
+    # A project whose main line is 'trunk' matches nothing, so the guess must
+    # degrade to the pre-existing "newest succeeded scan" rule, not to nothing.
+    user, project_id, _main, release_scan = await _seed_two_branch_project(
+        client, default_branch="trunk"
+    )
+    headers = _bearer_for(user)
+
+    response = await client.get(f"/v1/projects/{project_id}/gate-result", headers=headers)
+    assert response.status_code == 200, response.text
+    assert response.json()["scan_id"] == str(release_scan)
+
+
+async def test_anchor_defaults_to_main_when_default_branch_is_unset(client) -> None:
+    # default_branch is NULL on most projects (the create form never asks), so
+    # the fallback to 'main' is what makes the fix reach them at all.
+    user, project_id, main_scan, _release = await _seed_two_branch_project(
+        client, default_branch=None
+    )
+    headers = _bearer_for(user)
+
+    response = await client.get(f"/v1/projects/{project_id}/gate-result", headers=headers)
+    assert response.status_code == 200, response.text
+    assert response.json()["scan_id"] == str(main_scan)
+
+
+async def test_refless_scans_are_unaffected_by_the_main_line_preference(client) -> None:
+    # Ad-hoc scans carry ref=NULL. NULL must not be treated as matching 'main'
+    # (SQL NULL sorts first under DESC), so these projects keep pure recency.
+    team, user = await _seed_team_with_user(client)
+    project_id = await _seed_empty_project(client, team_id=team.id)
+    headers = _bearer_for(user)
+    base = datetime(2026, 5, 20, tzinfo=UTC)
+    await _seed_succeeded_scan(client, project_id=project_id, created_at=base, n_critical=2)
+    newest = await _seed_succeeded_scan(
+        client, project_id=project_id, created_at=base + timedelta(days=1), n_high=1
+    )
+
+    response = await client.get(f"/v1/projects/{project_id}/gate-result", headers=headers)
+    assert response.status_code == 200, response.text
+    assert response.json()["scan_id"] == str(newest)
+
+
+async def test_gate_ref_with_no_succeeded_scan_does_not_borrow_another_branch(
+    client,
+) -> None:
+    user, project_id, _main, _release = await _seed_two_branch_project(
+        client, default_branch="main"
+    )
+    headers = _bearer_for(user)
+
+    response = await client.get(
+        f"/v1/projects/{project_id}/gate-result",
+        headers=headers,
+        params={"ref": "feature/nope"},
+    )
+    assert response.status_code == 200, response.text
+    # No signal for that branch → the documented no-scan pass, and crucially
+    # NOT main's two criticals.
+    assert response.json()["scan_id"] is None
+    assert response.json()["gate"] == "pass"
+    assert response.json()["critical_cve_count"] == 0
+
+
 # ---------------------------------------------------------------------------
 # IDOR + invalid-pin guards
 # ---------------------------------------------------------------------------