Add provenance scanner for diagnostics links

backend/app/features/catalog/api.py

@@ -20,6 +20,7 @@
 from app.features.catalog.models import (
     Artifact,
     Case,
+    DiagnosticProvenanceState,
     Execution,
     ExternalLink,
     MetadataChange,
@@ -32,7 +33,9 @@
     CaseSummaryOut,
     CaseUpdate,
     CatalogOverviewOut,
+    DiagnosticProvenanceStateOut,
     DiagnosticsLinkRequest,
+    DiagnosticsScannerLinkRequest,
     ExecutionCreate,
     ExecutionExternalLinkOut,
     ExecutionFilterOptionsOut,
@@ -748,6 +751,115 @@ def link_case_diagnostics(
     )
 
 
+@diagnostics_router.get(
+    "/scanner-state", response_model=DiagnosticProvenanceStateOut | None
+)
+def get_diagnostics_scanner_state(
+    machine: str,
+    archive_relative_case_path: str,
+    db: Session = Depends(get_database_session),
+    user: User = Depends(current_active_user),
+) -> DiagnosticProvenanceStateOut | None:
+    """Return successful scanner state for one machine/archive case path."""
+    _require_diagnostics_scanner_role(user)
+    resolved_machine = resolve_machine_by_name(db, machine)
+
+    if resolved_machine is None:
+        raise HTTPException(status_code=404, detail="Unknown machine.")
+
+    state = (
+        db.query(DiagnosticProvenanceState)
+        .filter(DiagnosticProvenanceState.machine_name == resolved_machine.name)
+        .filter(
+            DiagnosticProvenanceState.archive_relative_case_path
+            == archive_relative_case_path
+        )
+        .one_or_none()
+    )
+
+    return DiagnosticProvenanceStateOut.model_validate(state) if state else None
+
+
+@diagnostics_router.post("/scanner/link", status_code=status.HTTP_204_NO_CONTENT)
+def link_scanner_diagnostics(
+    payload: DiagnosticsScannerLinkRequest,
+    db: Session = Depends(get_database_session),
+    user: User = Depends(current_active_user),
+) -> None:
+    """Atomically upsert one scanner-managed case diagnostic link and state."""
+    _require_diagnostics_scanner_role(user)
+
+    if len(payload.diagnostics) != 1:
+        raise HTTPException(
+            status_code=422, detail="Scanner payload requires one diagnostic."
+        )
+
+    if _unsafe_archive_relative_path(payload.provenance.archive_relative_case_path):
+        raise HTTPException(
+            status_code=422, detail="Invalid archive-relative case path."
+        )
+
+    machine = resolve_machine_by_name(db, payload.machine)
+    if machine is None:
+        raise HTTPException(status_code=404, detail="No matching case found.")
+
+    case_id = _resolve_case_id_for_diagnostics_link(
+        db=db,
+        case_name=payload.case_name,
+        machine_name=payload.machine,
+        hpc_username=payload.hpc_username,
+    )
+    diagnostic = payload.diagnostics[0]
+    now = datetime.now(timezone.utc)
+
+    with transaction(db):
+        link_id = db.execute(
+            pg_insert(ExternalLink)
+            .values(
+                case_id=case_id,
+                kind=ExternalLinkKind.DIAGNOSTIC,
+                url=str(diagnostic.url),
+                label=diagnostic.name,
+                created_at=now,
+                updated_at=now,
+            )
+            .on_conflict_do_update(
+                index_elements=[
+                    ExternalLink.case_id,
+                    ExternalLink.kind,
+                    ExternalLink.url,
+                ],
+                index_where=ExternalLink.case_id.is_not(None),
+                set_={"label": diagnostic.name, "updated_at": now},
+            )
+            .returning(ExternalLink.id)
+        ).scalar_one()
+        db.execute(
+            pg_insert(DiagnosticProvenanceState)
+            .values(
+                link_id=link_id,
+                machine_name=machine.name,
+                archive_relative_case_path=payload.provenance.archive_relative_case_path,
+                settings_filename=payload.provenance.settings_filename,
+                provenance_timestamp=payload.provenance.provenance_timestamp,
+                fingerprint=payload.provenance.fingerprint,
+                linked_url=str(diagnostic.url),
+                submitted_at=now,
+            )
+            .on_conflict_do_update(
+                constraint="uq_diagnostic_provenance_states_machine_path",
+                set_={
+                    "link_id": link_id,
+                    "settings_filename": payload.provenance.settings_filename,
+                    "provenance_timestamp": payload.provenance.provenance_timestamp,
+                    "fingerprint": payload.provenance.fingerprint,
+                    "linked_url": str(diagnostic.url),
+                    "submitted_at": now,
+                },
+            )
+        )
+
+
 @execution_router.get(
     "",
     response_model=ExecutionPageOut,
@@ -1219,6 +1331,20 @@ def _resolve_case_id_for_diagnostics_link(
     return match[0]
 
 
+def _require_diagnostics_scanner_role(user: User) -> None:
+    if user.role not in (UserRole.ADMIN, UserRole.SERVICE_ACCOUNT):
+        raise HTTPException(
+            status_code=403,
+            detail="Scanner access requires an administrator or service account.",
+        )
+
+
+def _unsafe_archive_relative_path(value: str) -> bool:
+    return value.startswith("/") or any(
+        part in {"", ".", ".."} for part in value.split("/")
+    )
+
+
 def _upsert_case_diagnostic_links(
     *,
     db: Session,

backend/app/features/catalog/models.py

@@ -322,3 +322,48 @@ class ExternalLink(Base, IDMixin, TimestampMixin):
         foreign_keys=[case_id],
         passive_deletes=True,
     )
+    diagnostic_provenance_state: Mapped[DiagnosticProvenanceState | None] = (
+        relationship(
+            back_populates="link",
+            cascade="all, delete-orphan",
+            passive_deletes=True,
+            uselist=False,
+        )
+    )
+
+
+class DiagnosticProvenanceState(Base, IDMixin):
+    """Successful scanner submission state for one published diagnostics link."""
+
+    __tablename__ = "diagnostic_provenance_states"
+    __table_args__ = (
+        UniqueConstraint(
+            "machine_name",
+            "archive_relative_case_path",
+            name="uq_diagnostic_provenance_states_machine_path",
+        ),
+        UniqueConstraint("link_id", name="uq_diagnostic_provenance_states_link_id"),
+    )
+
+    link_id: Mapped[UUID] = mapped_column(
+        PG_UUID(as_uuid=True),
+        ForeignKey("external_links.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    machine_name: Mapped[str] = mapped_column(String(200), nullable=False)
+    archive_relative_case_path: Mapped[str] = mapped_column(Text, nullable=False)
+    settings_filename: Mapped[str] = mapped_column(String(255), nullable=False)
+    provenance_timestamp: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False
+    )
+    fingerprint: Mapped[str] = mapped_column(String(128), nullable=False)
+    linked_url: Mapped[str] = mapped_column(String(1000), nullable=False)
+    submitted_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False
+    )
+
+    link: Mapped[ExternalLink] = relationship(
+        back_populates="diagnostic_provenance_state",
+        foreign_keys=[link_id],
+        passive_deletes=True,
+    )

backend/app/features/catalog/schemas.py

@@ -152,6 +152,33 @@ class DiagnosticsLinkRequest(CamelInBaseModel):
     ]
 
 
+class DiagnosticProvenanceMetadata(CamelInBaseModel):
+    """Immutable provenance identity supplied by the diagnostics scanner."""
+
+    archive_relative_case_path: Annotated[
+        str, Field(..., min_length=1, max_length=1000)
+    ]
+    settings_filename: Annotated[str, Field(..., min_length=1, max_length=255)]
+    provenance_timestamp: datetime
+    fingerprint: Annotated[str, Field(..., min_length=1, max_length=128)]
+
+
+class DiagnosticsScannerLinkRequest(DiagnosticsLinkRequest):
+    """Scanner-only diagnostics link request with successful provenance state."""
+
+    provenance: DiagnosticProvenanceMetadata
+
+
+class DiagnosticProvenanceStateOut(CamelOutBaseModel):
+    machine_name: str
+    archive_relative_case_path: str
+    settings_filename: str
+    provenance_timestamp: datetime
+    fingerprint: str
+    linked_url: str
+    submitted_at: datetime
+
+
 class ArtifactCreate(CamelInBaseModel):
     """Schema for creating a new Artifact."""
 

backend/app/scripts/README.md

@@ -18,10 +18,14 @@ scripts/
 │   ├── archive_ingestor_core.py
 │   ├── archive_layout.py
 │   ├── archive_workflow.py
+│   ├── diagnostics_archives.py
+│   ├── diagnostics_link_scanner.py
 │   ├── hpc_upload_archive_ingestor.py
 │   ├── nersc_archive_ingestor.py
 │   ├── sites/
-│       └── nersc.sh
+│   │   ├── lcrc-diagnostics-scanner.sh
+│   │   ├── nersc-diagnostics-scanner.sh
+│   │   └── nersc.sh
 │   └── v3_data/
 │       ├── __init__.py
 │       ├── lcrc-v3.env.example
@@ -209,6 +213,35 @@ the documented Chrysalis archive root, and records uploads under machine
 `ARCHIVE_YEAR_END` variables remain supported. `SCAN_MODE`,
 `ARCHIVE_YEAR_START`, and `MACHINE_NAME` are ignored because source site and
 scan scope are fixed.
+## Diagnostics Provenance Scanner
+
+Scans newest paired zppy provenance from the reviewed static registry and creates
+case-scoped diagnostic links. It never reads Mache configuration at runtime.
+
+Run through the NERSC wrapper:
+
+```bash
+SIMBOARD_API_TOKEN=<service-account-token> \
+MACHINE_NAME=perlmutter \
+DRY_RUN=true \
+backend/app/scripts/ingestion/sites/nersc-diagnostics-scanner.sh
+```
+
+Use `sites/lcrc-diagnostics-scanner.sh` at LCRC with
+`MACHINE_NAME=chrysalis`. `MACHINE_NAME` is required for every diagnostics
+scanner invocation; wrappers do not assign a machine default. A non-dry run
+also requires an API base URL and service-account token. Roots and public URLs
+come only from `diagnostics_archives.py`.
+
+Start with `DRY_RUN=true`; it needs no API URL or token. Inspect logs, then
+schedule with `DRY_RUN=false`, which requires both API URL and service token.
+The scanner emits structured events for startup configuration, discovery,
+candidate selection, state lookups, retry outcomes, and completion; credentials
+are never logged. Dry runs also emit one candidate event per discovered link.
+Scanner account needs read/traverse access to `production/` and `development/`,
+provenance settings, and published output. Failed or not-ready candidates retry
+next run. Refresh registry entries from Mache `[web_portal]` cfg data only in a
+reviewed change; never add archive-path environment overrides.
 
 ## HPC Upload Archive Ingestor
 

backend/app/scripts/ingestion/diagnostics_archives.py

@@ -0,0 +1,35 @@
+"""Reviewed diagnostics archive locations; never populated at scanner runtime."""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class DiagnosticsArchive:
+    root: str
+    public_base_url: str
+
+
+# Refresh from Mache [web_portal] configuration in a reviewed change when sites move.
+# Source: https://github.com/E3SM-Project/mache/tree/main/mache/machines
+DIAGNOSTICS_ARCHIVES_BY_MACHINE: dict[str, DiagnosticsArchive] = {
+    "perlmutter": DiagnosticsArchive(
+        root="/global/cfs/cdirs/e3sm/www/diagnostics_archive",
+        public_base_url="https://portal.nersc.gov/cfs/e3sm/diagnostics_archive",
+    ),
+    "pm": DiagnosticsArchive(
+        root="/global/cfs/cdirs/e3sm/www/diagnostics_archive",
+        public_base_url="https://portal.nersc.gov/cfs/e3sm/diagnostics_archive",
+    ),
+    "pm-cpu": DiagnosticsArchive(
+        root="/global/cfs/cdirs/e3sm/www/diagnostics_archive",
+        public_base_url="https://portal.nersc.gov/cfs/e3sm/diagnostics_archive",
+    ),
+    "pm-gpu": DiagnosticsArchive(
+        root="/global/cfs/cdirs/e3sm/www/diagnostics_archive",
+        public_base_url="https://portal.nersc.gov/cfs/e3sm/diagnostics_archive",
+    ),
+    "chrysalis": DiagnosticsArchive(
+        root="/lcrc/group/e3sm/public_html/diagnostic_output/diagnostics_archive",
+        public_base_url="https://web.lcrc.anl.gov/public/e3sm/diagnostic_output/diagnostics_archive",
+    ),
+}