feat: carry controlled SBOM source refs

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118

src/activity_core/context_resolvers/sbom_nexus.py

@@ -19,6 +19,11 @@ first), plus fleet counts for the evidence report.
               "sbom_age_days": int,
               "has_sbom": bool,
               "checkout_available": bool | None,
+              "source_ref": {
+                  "kind": "forgejo-archive-v1",
+                  "repository": str,
+                  "revision": str,
+              } | None,
           },
           ...
       ],
@@ -43,6 +48,7 @@ Config: SBOM_NEXUS_URL env var (default: http://127.0.0.1:8010).
 from __future__ import annotations
 
 import os
+import re
 from collections.abc import Callable
 from typing import Any
 from urllib.parse import quote
@@ -50,7 +56,10 @@ from uuid import NAMESPACE_URL, uuid5
 
 import httpx
 
-from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver
+from activity_core.context_resolvers.base import (
+    CONTEXT_RESOLVER_REGISTRY,
+    ContextResolver,
+)
 
 _DEFAULT_SBOM_NEXUS_URL = "http://127.0.0.1:8010"
 _TIMEOUT_SECONDS = 15.0
@@ -62,6 +71,13 @@ _MAX_CATCH_UP_LIMIT = 25
 # Mirrors state_hub._NEVER_SCANNED_AGE_DAYS: a repo that was never scanned
 # sorts ahead of every real age without needing a null-aware comparison.
 _NEVER_SCANNED_AGE_DAYS = 9999
+_TERMINAL_SKIP_REASONS = {
+    "no-checkout",
+    "no-manifest",
+    "ingest-error",
+    "source-unavailable",
+    "source-rejected",
+}
 
 
 def _base_url() -> str:
@@ -135,12 +151,24 @@ def _normalise_entry(raw: Any) -> dict[str, Any] | None:
     if not isinstance(checkout_available, bool):
         checkout_available = None
 
+    source_ref = raw.get("source_ref")
+    if not isinstance(source_ref, dict) or (
+        source_ref.get("kind") != "forgejo-archive-v1"
+        or not isinstance(source_ref.get("repository"), str)
+        or not isinstance(source_ref.get("revision"), str)
+        or re.fullmatch(r"[0-9a-f]{40}", source_ref["revision"]) is None
+    ):
+        source_ref = None
+    else:
+        source_ref = dict(source_ref)
+
     return {
         "repo_slug": repo_slug,
         "last_sbom_at": last_sbom_at,
         "sbom_age_days": max(0, age_days),
         "has_sbom": has_sbom,
         "checkout_available": checkout_available,
+        "source_ref": source_ref,
     }
 
 
@@ -206,25 +234,27 @@ def _skip(
     if (
         not isinstance(raw, dict)
         or raw.get("status") != "skipped"
-        or raw.get("reason") not in {"no-checkout", "no-manifest", "ingest-error"}
+        or raw.get("reason") not in _TERMINAL_SKIP_REASONS
     ):
         raise RuntimeError(f"sbom-nexus skip returned an invalid outcome for {repo_slug}")
     return raw
 
 
-def _ingest(repo_slug: str, *, operation_id: str) -> dict[str, Any]:
+def _ingest(
+    repo_slug: str,
+    *,
+    operation_id: str,
+    source_ref: dict[str, Any] | None = None,
+) -> dict[str, Any]:
     raw = _post_json(
         f"/sbom/{quote(repo_slug, safe='')}/ingest",
+        {"source_ref": source_ref} if source_ref else None,
         idempotency_key=_operation_key(operation_id, repo_slug),
     )
     valid = isinstance(raw, dict) and raw.get("status") in {"ingested", "skipped"}
     if not valid:
         raise RuntimeError(f"sbom-nexus ingest returned an invalid outcome for {repo_slug}")
-    if raw.get("status") == "skipped" and raw.get("reason") not in {
-        "no-checkout",
-        "no-manifest",
-        "ingest-error",
-    }:
+    if raw.get("status") == "skipped" and raw.get("reason") not in _TERMINAL_SKIP_REASONS:
         raise RuntimeError(f"sbom-nexus ingest returned an invalid skip for {repo_slug}")
     return raw
 
@@ -282,14 +312,19 @@ def apply_bounded_ingest(
             continue
         if on_progress:
             on_progress(list(outcomes_by_slug.values()))
-        if repo.get("checkout_available") is False:
+        source_ref = repo.get("source_ref")
+        if repo.get("checkout_available") is False and not source_ref:
             outcome = _skip(
                 repo_slug,
                 "no-checkout",
                 operation_id=operation_id,
             )
         else:
-            outcome = _ingest(repo_slug, operation_id=operation_id)
+            outcome = _ingest(
+                repo_slug,
+                operation_id=operation_id,
+                source_ref=source_ref if isinstance(source_ref, dict) else None,
+            )
         outcomes_by_slug[repo_slug] = _compact_outcome(repo_slug, outcome)
         if on_progress:
             on_progress(list(outcomes_by_slug.values()))