feat: add controlled source ingestion and replay

Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118
2026-08-22 23:57:37 +02:00 · 2026-08-22 23:57:37 +02:00 · 879012c776
commit 879012c776
parent b95fba9a9f
16 changed files with 1156 additions and 154 deletions
--- a/README.md
+++ b/README.md
@ -31,5 +31,13 @@ development environments; mounted secret files are preferred for production.
 - `POST /sbom/{repo_slug}/ingest`
 - State Hub-compatible `/sbom/` snapshot, entry, repository, and licence routes

+Authoritative automation may ingest a controlled `forgejo-archive-v1`
+`source_ref` pinned to a full commit SHA. Nexus constructs the Forgejo URL,
+streams and safely extracts the archive within configured limits, records the
+archive and manifest provenance, and removes the transient directory. It does
+not accept arbitrary source URLs. `Idempotency-Key` is durably enforced on
+repository ingest and skip operations; a matching retry replays the original
+snapshot outcome and conflicting key reuse returns HTTP 409.
+
 See [docs/state-hub-sbom-extraction-review.md](docs/state-hub-sbom-extraction-review.md)
 for the extraction inventory and cutover dispositions.
--- a/WORK-RECORDS.md
+++ b/WORK-RECORDS.md
@ -9,8 +9,13 @@
 | Kind | ID | Status | Lane | Source |
 | --- | --- | --- | --- | --- |
 | workplan | SBOM-WP-0001 | finished | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md |
+| workplan | SBOM-WP-0003 | active | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md |
 | task | SBOM-WP-0001-T01 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md |
 | task | SBOM-WP-0001-T02 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md |
 | task | SBOM-WP-0001-T03 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md |
 | task | SBOM-WP-0001-T04 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md |
 | task | SBOM-WP-0001-T05 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md |
+| task | SBOM-WP-0003-T01 | done | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md |
+| task | SBOM-WP-0003-T02 | done | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md |
+| task | SBOM-WP-0003-T03 | done | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md |
+| task | SBOM-WP-0003-T04 | wait | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md |
--- a/docs/operator-guide.md
+++ b/docs/operator-guide.md
@ -42,8 +42,48 @@ curl -X POST http://127.0.0.1:8010/sbom/example/ingest
 ```

 The ingest response is terminal: `ingested`, or `skipped` with one of
-`no-checkout`, `no-manifest`, or `ingest-error`. A skip advances queue fairness
-but does not advance `last_success_at`.
+`no-checkout`, `no-manifest`, `ingest-error`, `source-unavailable`, or
+`source-rejected`. A skip advances queue fairness but does not advance
+`last_success_at`.
+
+## Controlled Forgejo source
+
+Production automation uses the contract selected by `CUST-WP-0064`: Repo
+Manager projects a canonical Coulomb repository and full commit SHA, Activity
+Core freezes that `source_ref` with its bounded target set, and Nexus alone
+fetches and scans the source. Example projection:
+
+```json
+{
+  "source_ref": {
+    "kind": "forgejo-archive-v1",
+    "repository": "coulomb/example",
+    "revision": "0123456789abcdef0123456789abcdef01234567",
+    "observed_ref": "refs/heads/main",
+    "observed_at": "2026-08-22T20:00:00Z"
+  }
+}
+```
+
+Enable only after the package has migration `0002`, bounded ephemeral storage,
+and Forgejo/DNS egress:
+
+```sh
+export SBOM_NEXUS_CONTROLLED_SOURCE_ENABLED=true
+export SBOM_NEXUS_FORGEJO_BASE_URL=http://forgejo-gitea-http.forgejo.svc.cluster.local:3000
+export SBOM_NEXUS_SOURCE_TMP=/var/run/sbom-sources
+```
+
+The runtime defaults are one concurrent scan, 120 seconds each for fetch and
+scan, 100 MiB compressed, 512 MiB expanded, and 100,000 archive members. The
+source endpoint accepts no arbitrary URL and requires no credential for public
+Coulomb repositories. Never substitute a Forgejo administrator token.
+
+Automated ingest and skip requests send the same value in `Idempotency-Key`
+and `X-Activity-Core-Operation-ID`. Nexus persists that operation identity in
+the snapshot transaction. A retry with the same request replays the original
+terminal outcome; reuse against a different route, repository, reason, or
+source reference returns HTTP 409.

 ## Inspect catch-up

--- a/migrations/versions/0002_controlled_source_and_operation_receipts.py
+++ b/migrations/versions/0002_controlled_source_and_operation_receipts.py
@ -0,0 +1,47 @@
+"""Controlled source provenance and durable operation receipts.
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-08-22
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0002"
+down_revision = "0001"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("repositories", sa.Column("source_ref_json", sa.JSON(), nullable=True))
+    op.add_column("snapshots", sa.Column("source_provenance_json", sa.JSON(), nullable=True))
+    op.create_table(
+        "operation_receipts",
+        sa.Column("operation_key", sa.String(length=200), nullable=False),
+        sa.Column("request_fingerprint", sa.String(length=64), nullable=False),
+        sa.Column("snapshot_id", sa.String(length=36), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["snapshot_id"],
+            ["snapshots.id"],
+            name=op.f("fk_operation_receipts_snapshot_id_snapshots"),
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("operation_key", name=op.f("pk_operation_receipts")),
+    )
+    op.create_index(
+        "ix_operation_receipts_snapshot",
+        "operation_receipts",
+        ["snapshot_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_operation_receipts_snapshot", table_name="operation_receipts")
+    op.drop_table("operation_receipts")
+    op.drop_column("snapshots", "source_provenance_json")
+    op.drop_column("repositories", "source_ref_json")
--- a/src/sbom_nexus/api.py
+++ b/src/sbom_nexus/api.py
@ -1,23 +1,43 @@
 """FastAPI product and State Hub compatibility surface."""
 from __future__ import annotations

+import hashlib
+import json
 import os
+import subprocess
+import sys
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Literal

-from fastapi import FastAPI, HTTPException, Query, Request
+from fastapi import FastAPI, Header, HTTPException, Query, Request
 from pydantic import BaseModel, Field

 from sbom_nexus.config import database_target
 from sbom_nexus.scanner import VALID_ECOSYSTEMS, scan_repository
-from sbom_nexus.storage import Store
+from sbom_nexus.source_fetch import (
+    ControlledSourceDisabled,
+    SourceRejected,
+    SourceUnavailable,
+    controlled_source_enabled,
+    fetch_controlled_source,
+)
+from sbom_nexus.storage import OperationKeyConflict, Store

 DEFAULT_STALE_DAYS = int(os.getenv("SBOM_NEXUS_STALE_DAYS", "30"))


+class SourceRef(BaseModel):
+    kind: Literal["forgejo-archive-v1"]
+    repository: str = Field(min_length=1, max_length=400)
+    revision: str = Field(pattern=r"^[0-9a-f]{40}$")
+    observed_ref: str | None = Field(default=None, max_length=300)
+    observed_at: datetime | None = None
+
+
 class RepositoryUpsert(BaseModel):
    checkout_path: str | None = None
+    source_ref: SourceRef | None = None
    active: bool = True
    last_sbom_at: datetime | None = None
    last_attempt_at: datetime | None = None
@ -50,10 +70,20 @@ class HistoricalImport(BaseModel):


 class SkipRequest(BaseModel):
-    reason: Literal["no-checkout", "no-manifest", "ingest-error"]
+    reason: Literal[
+        "no-checkout",
+        "no-manifest",
+        "ingest-error",
+        "source-unavailable",
+        "source-rejected",
+    ]
    detail: str | None = None


+class IngestRequest(BaseModel):
+    source_ref: SourceRef | None = None
+
+
 def _store(request: Request) -> Store:
    return request.app.state.store

@ -81,6 +111,96 @@ def _auto_create(store: Store) -> bool:
    return store.dialect == "sqlite"


+def _operation_key(idempotency_key: str | None, operation_id: str | None) -> str | None:
+    if idempotency_key and operation_id and idempotency_key != operation_id:
+        raise HTTPException(status_code=400, detail="operation identity headers disagree")
+    key = idempotency_key or operation_id
+    if key and len(key) > 200:
+        raise HTTPException(status_code=400, detail="operation identity is too long")
+    return key
+
+
+def _request_fingerprint(route: str, repo_slug: str, body: dict[str, Any]) -> str:
+    encoded = json.dumps(
+        {"route": route, "repo_slug": repo_slug, "body": body},
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode()
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _outcome(repo_slug: str, snapshot: dict[str, Any]) -> dict[str, Any]:
+    status = snapshot["status"]
+    if status == "ingested":
+        return {
+            "repo_slug": repo_slug,
+            "snapshot_id": snapshot["id"],
+            "status": "ingested",
+            "entry_count": snapshot["entry_count"],
+            "snapshot_at": snapshot["snapshot_at"],
+            "source_revision": snapshot["source_revision"],
+        }
+    result = {
+        "repo_slug": repo_slug,
+        "snapshot_id": snapshot["id"],
+        "status": "skipped",
+        "reason": status,
+        "entry_count": snapshot["entry_count"],
+        "snapshot_at": snapshot["snapshot_at"],
+    }
+    if status == "ingest-error":
+        result["errors"] = snapshot["errors"]
+    return result
+
+
+def _replay_or_none(
+    store: Store,
+    repo_slug: str,
+    operation_key: str | None,
+    fingerprint: str,
+) -> dict[str, Any] | None:
+    try:
+        snapshot = store.replay_operation(operation_key, fingerprint)
+    except OperationKeyConflict as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    return _outcome(repo_slug, snapshot) if snapshot else None
+
+
+def _scan_controlled_source(
+    source_root: Path, repo_slug: str, revision: str
+) -> dict[str, Any]:
+    timeout = int(os.getenv("SBOM_NEXUS_SOURCE_SCAN_TIMEOUT_SECONDS", "120"))
+    try:
+        completed = subprocess.run(
+            [
+                sys.executable,
+                "-m",
+                "sbom_nexus.cli",
+                "scan",
+                str(source_root),
+                "--slug",
+                repo_slug,
+                "--source-revision",
+                revision,
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise SourceRejected("source scan exceeded its time limit") from exc
+    if completed.returncode not in {0, 1}:
+        raise SourceRejected("source scanner process failed")
+    try:
+        result = json.loads(completed.stdout)
+    except json.JSONDecodeError as exc:
+        raise SourceRejected("source scanner returned malformed output") from exc
+    if not isinstance(result, dict) or result.get("source_revision") != revision:
+        raise SourceRejected("source scanner returned invalid provenance")
+    return result
+
+
 def create_app(database_path: str | Path | None = None) -> FastAPI:
    application = FastAPI(
        title="SBOM Nexus",
@ -107,6 +227,11 @@ def create_app(database_path: str | Path | None = None) -> FastAPI:
        return _store(request).upsert_repository(
            repo_slug,
            checkout_path=body.checkout_path,
+            source_ref=(
+                body.source_ref.model_dump(mode="json", exclude_none=True)
+                if body.source_ref
+                else None
+            ),
            active=body.active,
            last_attempt_at=body.last_attempt_at or legacy_time,
            last_success_at=body.last_success_at or legacy_time,
@ -210,63 +335,140 @@ def create_app(database_path: str | Path | None = None) -> FastAPI:
        )

    @application.post("/sbom/{repo_slug}/ingest")
-    def ingest_repository(repo_slug: str, request: Request) -> dict[str, Any]:
+    def ingest_repository(
+        repo_slug: str,
+        request: Request,
+        body: IngestRequest | None = None,
+        idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
+        operation_id: str | None = Header(
+            default=None, alias="X-Activity-Core-Operation-ID"
+        ),
+    ) -> dict[str, Any]:
        store = _store(request)
        repo = store.get_repository(repo_slug)
        if repo is None:
            raise _not_found(repo_slug)
-        checkout_path = repo["checkout_path"]
-        if not checkout_path or not Path(checkout_path).is_dir():
-            return _record_skip(store, repo_slug, "no-checkout")
+        key = _operation_key(idempotency_key, operation_id)
+        selected_ref = body.source_ref if body and body.source_ref else None
+        source_ref = (
+            selected_ref.model_dump(mode="json", exclude_none=True)
+            if selected_ref
+            else repo.get("source_ref")
+        )
+        fingerprint = _request_fingerprint(
+            "ingest", repo_slug, {"source_ref": source_ref}
+        )
+        replay = _replay_or_none(store, repo_slug, key, fingerprint)
+        if replay:
+            return replay

-        scan = scan_repository(Path(checkout_path), slug=repo_slug)
+        if source_ref:
+            try:
+                with fetch_controlled_source(repo_slug, source_ref) as controlled:
+                    scan = _scan_controlled_source(
+                        controlled.root,
+                        repo_slug,
+                        source_ref["revision"],
+                    )
+                    provenance = controlled.provenance
+            except ControlledSourceDisabled as exc:
+                raise HTTPException(status_code=503, detail=str(exc)) from exc
+            except SourceUnavailable as exc:
+                return _record_skip(
+                    store,
+                    repo_slug,
+                    "source-unavailable",
+                    str(exc),
+                    operation_key=key,
+                    request_fingerprint=fingerprint,
+                    source_revision=source_ref["revision"],
+                    source_provenance=source_ref,
+                )
+            except SourceRejected as exc:
+                return _record_skip(
+                    store,
+                    repo_slug,
+                    "source-rejected",
+                    str(exc),
+                    operation_key=key,
+                    request_fingerprint=fingerprint,
+                    source_revision=source_ref["revision"],
+                    source_provenance=source_ref,
+                )
+            source = "forgejo-archive-v1"
+        else:
+            provenance = None
+            checkout_path = repo["checkout_path"]
+            if not checkout_path or not Path(checkout_path).is_dir():
+                reason = "source-unavailable" if controlled_source_enabled() else "no-checkout"
+                return _record_skip(
+                    store,
+                    repo_slug,
+                    reason,
+                    operation_key=key,
+                    request_fingerprint=fingerprint,
+                )
+            scan = scan_repository(Path(checkout_path), slug=repo_slug)
+            source = "repository-scan"
        if not scan["sources"] and not scan["errors"]:
-            return _record_skip(store, repo_slug, "no-manifest")
+            return _record_skip(
+                store,
+                repo_slug,
+                "no-manifest",
+                operation_key=key,
+                request_fingerprint=fingerprint,
+                source_revision=scan["source_revision"],
+                source_provenance=provenance,
+            )
        if scan["errors"]:
+            status = "ingest-error"
+        else:
+            status = "ingested"
+        try:
            snapshot, _created = store.record_snapshot(
                repo_slug,
                entries=scan["entries"],
-                status="ingest-error",
-                source="ingest-error",
+                status=status,
+                source=source if status == "ingested" else "ingest-error",
                source_revision=scan["source_revision"],
+                source_provenance=provenance,
                sources=scan["sources"],
                errors=scan["errors"],
+                operation_key=key,
+                request_fingerprint=fingerprint if key else None,
            )
-            return {
-                "repo_slug": repo_slug,
-                "snapshot_id": snapshot["id"],
-                "status": "skipped",
-                "reason": "ingest-error",
-                "entry_count": snapshot["entry_count"],
-                "errors": scan["errors"],
-                "snapshot_at": snapshot["snapshot_at"],
-            }
-
-        snapshot, _created = store.record_snapshot(
-            repo_slug,
-            entries=scan["entries"],
-            status="ingested",
-            source="repository-scan",
-            source_revision=scan["source_revision"],
-            sources=scan["sources"],
-        )
-        return {
-            "repo_slug": repo_slug,
-            "snapshot_id": snapshot["id"],
-            "status": "ingested",
-            "entry_count": snapshot["entry_count"],
-            "snapshot_at": snapshot["snapshot_at"],
-            "source_revision": snapshot["source_revision"],
-        }
+        except OperationKeyConflict as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        return _outcome(repo_slug, snapshot)

    @application.post("/sbom/{repo_slug}/skip")
    def skip_repository(
-        repo_slug: str, body: SkipRequest, request: Request
+        repo_slug: str,
+        body: SkipRequest,
+        request: Request,
+        idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
+        operation_id: str | None = Header(
+            default=None, alias="X-Activity-Core-Operation-ID"
+        ),
    ) -> dict[str, Any]:
        store = _store(request)
        if store.get_repository(repo_slug) is None:
            raise _not_found(repo_slug)
-        return _record_skip(store, repo_slug, body.reason, body.detail)
+        key = _operation_key(idempotency_key, operation_id)
+        fingerprint = _request_fingerprint(
+            "skip", repo_slug, {"reason": body.reason, "detail": body.detail}
+        )
+        replay = _replay_or_none(store, repo_slug, key, fingerprint)
+        if replay:
+            return replay
+        return _record_skip(
+            store,
+            repo_slug,
+            body.reason,
+            body.detail,
+            operation_key=key,
+            request_fingerprint=fingerprint,
+        )

    @application.get("/sbom/{repo_slug}")
    def get_repo_sbom(repo_slug: str, request: Request) -> dict[str, Any]:
@ -279,21 +481,29 @@ def create_app(database_path: str | Path | None = None) -> FastAPI:


 def _record_skip(
-    store: Store, repo_slug: str, reason: str, detail: str | None = None
+    store: Store,
+    repo_slug: str,
+    reason: str,
+    detail: str | None = None,
+    *,
+    operation_key: str | None = None,
+    request_fingerprint: str | None = None,
+    source_revision: str | None = None,
+    source_provenance: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
    errors = [{"reason": reason, "detail": detail}] if detail else []
-    snapshot, _created = store.record_snapshot(
-        repo_slug,
-        entries=[],
-        status=reason,
-        source=reason,
-        errors=errors,
-    )
-    return {
-        "repo_slug": repo_slug,
-        "snapshot_id": snapshot["id"],
-        "status": "skipped",
-        "reason": reason,
-        "entry_count": 0,
-        "snapshot_at": snapshot["snapshot_at"],
-    }
+    try:
+        snapshot, _created = store.record_snapshot(
+            repo_slug,
+            entries=[],
+            status=reason,
+            source=reason,
+            source_revision=source_revision,
+            source_provenance=source_provenance,
+            errors=errors,
+            operation_key=operation_key,
+            request_fingerprint=request_fingerprint if operation_key else None,
+        )
+    except OperationKeyConflict as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    return _outcome(repo_slug, snapshot)
--- a/src/sbom_nexus/cli.py
+++ b/src/sbom_nexus/cli.py
@ -27,7 +27,9 @@ def _serve(args: argparse.Namespace) -> None:


 def _scan(args: argparse.Namespace) -> None:
-    result = scan_repository(Path(args.path), slug=args.slug)
+    result = scan_repository(
+        Path(args.path), slug=args.slug, source_revision=args.source_revision
+    )
    rendered = json.dumps(result, indent=2, sort_keys=True)
    if args.output:
        output = Path(args.output)
@ -55,6 +57,7 @@ def build_parser() -> argparse.ArgumentParser:
    scan = subparsers.add_parser("scan", help="Derive a snapshot from repository sources")
    scan.add_argument("path", nargs="?", default=".")
    scan.add_argument("--slug")
+    scan.add_argument("--source-revision")
    scan.add_argument("--output")
    scan.add_argument("--force", action="store_true")
    scan.set_defaults(handler=_scan)
--- a/src/sbom_nexus/database.py
+++ b/src/sbom_nexus/database.py
@ -37,6 +37,7 @@ repositories = Table(
    Column("id", String(36), primary_key=True),
    Column("slug", String(200), nullable=False, unique=True),
    Column("checkout_path", Text, nullable=True),
+    Column("source_ref_json", JSON, nullable=True),
    Column("active", Boolean, nullable=False),
    Column("last_attempt_at", DateTime(timezone=True), nullable=True),
    Column("last_success_at", DateTime(timezone=True), nullable=True),
@ -61,12 +62,27 @@ snapshots = Table(
    Column("status", String(50), nullable=False),
    Column("entry_count", Integer, nullable=False),
    Column("source_revision", String(200), nullable=True),
+    Column("source_provenance_json", JSON, nullable=True),
    Column("sources_json", JSON, nullable=False),
    Column("errors_json", JSON, nullable=False),
    Column("legacy_id", String(100), nullable=True, unique=True),
    Column("created_at", DateTime(timezone=True), nullable=False),
 )

+operation_receipts = Table(
+    "operation_receipts",
+    metadata,
+    Column("operation_key", String(200), primary_key=True),
+    Column("request_fingerprint", String(64), nullable=False),
+    Column(
+        "snapshot_id",
+        String(36),
+        ForeignKey("snapshots.id", ondelete="RESTRICT"),
+        nullable=False,
+    ),
+    Column("created_at", DateTime(timezone=True), nullable=False),
+)
+
 entries = Table(
    "entries",
    metadata,
@ -97,6 +113,7 @@ Index("ix_snapshots_repo_time", snapshots.c.repo_id, snapshots.c.snapshot_at)
 Index("ix_entries_snapshot", entries.c.snapshot_id)
 Index("ix_entries_repo", entries.c.repo_id)
 Index("ix_entries_license", entries.c.license_spdx)
+Index("ix_operation_receipts_snapshot", operation_receipts.c.snapshot_id)


 def database_url(value: str | Path) -> str:
--- a/src/sbom_nexus/projection.py
+++ b/src/sbom_nexus/projection.py
@ -2,6 +2,7 @@

 from __future__ import annotations

+import re
 import socket
 import urllib.parse
 from collections import Counter
@ -16,6 +17,38 @@ def _checkout_path(repo: dict[str, Any], host_id: str | None) -> str | None:
    return host_paths.get(selected_host) or repo.get("local_path")


+def _forgejo_repository(remote_url: str | None) -> str | None:
+    if not remote_url:
+        return None
+    patterns = (
+        r"^forgejo-remote:(coulomb/[a-z0-9._-]+?)(?:\.git)?$",
+        r"^https://forgejo\.coulomb\.social/(coulomb/[a-z0-9._-]+?)(?:\.git)?$",
+        r"^ssh://git@forgejo\.coulomb\.social(?::\d+)?/(coulomb/[a-z0-9._-]+?)(?:\.git)?$",
+    )
+    for pattern in patterns:
+        match = re.fullmatch(pattern, remote_url)
+        if match:
+            return match.group(1)
+    return None
+
+
+def _source_ref(repo: dict[str, Any]) -> dict[str, Any] | None:
+    repository = _forgejo_repository(repo.get("remote_url"))
+    revision = repo.get("git_fingerprint")
+    if repository != f"coulomb/{repo['slug']}":
+        return None
+    if not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision):
+        return None
+    observed_at = repo.get("last_state_synced_at") or repo.get("updated_at")
+    return {
+        "kind": "forgejo-archive-v1",
+        "repository": repository,
+        "revision": revision,
+        "observed_ref": "repo-manager:git_fingerprint",
+        **({"observed_at": observed_at} if observed_at else {}),
+    }
+
+
 def sync_repository_projections(
    source_url: str,
    target_url: str,
@ -38,13 +71,16 @@ def sync_repository_projections(
            }
        active = repo.get("status", "active") == "active"
        checkout_path = _checkout_path(repo, host_id)
+        source_ref = _source_ref(repo)
        projections[slug] = {
            "slug": slug,
            "active": active,
            "checkout_path": checkout_path,
+            "source_ref": source_ref,
        }
        statuses["active" if active else "inactive"] += 1
        statuses["with_checkout_path" if checkout_path else "without_checkout_path"] += 1
+        statuses["with_source_ref" if source_ref else "without_source_ref"] += 1

    if dry_run:
        return {
@ -65,6 +101,7 @@ def sync_repository_projections(
            body={
                "active": projection["active"],
                "checkout_path": projection["checkout_path"],
+                "source_ref": projection["source_ref"],
            },
        )

@ -78,7 +115,7 @@ def sync_repository_projections(
        actual = target_repositories[slug]
        differences = {
            field: {"source": expected[field], "target": actual.get(field)}
-            for field in ("active", "checkout_path")
+            for field in ("active", "checkout_path", "source_ref")
            if expected[field] != actual.get(field)
        }
        if differences:
--- a/src/sbom_nexus/scanner.py
+++ b/src/sbom_nexus/scanner.py
@ -325,7 +325,12 @@ def _utc_now_text() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


-def scan_repository(repo_root: Path, *, slug: str | None = None) -> dict[str, Any]:
+def scan_repository(
+    repo_root: Path,
+    *,
+    slug: str | None = None,
+    source_revision: str | None = None,
+) -> dict[str, Any]:
    repo_root = repo_root.resolve()
    sources: list[dict[str, Any]] = []
    entries: list[Entry] = []
@ -353,7 +358,7 @@ def scan_repository(repo_root: Path, *, slug: str | None = None) -> dict[str, An
        "ok": not errors,
        "repo_slug": slug or repo_root.name,
        "repo_path": str(repo_root),
-        "source_revision": _head_sha(repo_root),
+        "source_revision": source_revision or _head_sha(repo_root),
        "generated_at": _utc_now_text(),
        "authority": "detected lockfiles and reviewed sbom-tools.yaml",
        "sources": sources,
--- a/src/sbom_nexus/source_fetch.py
+++ b/src/sbom_nexus/source_fetch.py
@ -0,0 +1,234 @@
+"""Bounded retrieval and extraction for controlled Forgejo source archives."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import shutil
+import tarfile
+import tempfile
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections.abc import Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+SOURCE_KIND = "forgejo-archive-v1"
+REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
+SLUG_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,199}$")
+_SOURCE_SCAN_SLOT = threading.BoundedSemaphore(value=1)
+
+
+class SourceUnavailable(RuntimeError):
+    """The selected immutable source cannot currently be retrieved."""
+
+
+class SourceRejected(RuntimeError):
+    """The source reference or archive violates the controlled-input policy."""
+
+
+class ControlledSourceDisabled(RuntimeError):
+    """The controlled-source feature flag is disabled."""
+
+
+@dataclass(frozen=True)
+class ControlledSource:
+    root: Path
+    provenance: dict[str, Any]
+
+
+def controlled_source_enabled() -> bool:
+    return os.getenv("SBOM_NEXUS_CONTROLLED_SOURCE_ENABLED", "false").lower() in {
+        "1",
+        "true",
+        "yes",
+    }
+
+
+def _positive_int(name: str, default: int) -> int:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise RuntimeError(f"{name} must be an integer") from exc
+    if value < 1:
+        raise RuntimeError(f"{name} must be positive")
+    return value
+
+
+def validate_source_ref(repo_slug: str, source_ref: dict[str, Any]) -> dict[str, Any]:
+    if not SLUG_PATTERN.fullmatch(repo_slug):
+        raise SourceRejected("invalid repository slug")
+    kind = source_ref.get("kind")
+    repository = source_ref.get("repository")
+    revision = source_ref.get("revision")
+    if kind != SOURCE_KIND:
+        raise SourceRejected("unsupported source kind")
+    if repository != f"coulomb/{repo_slug}":
+        raise SourceRejected("source repository does not match repository slug")
+    if not isinstance(revision, str) or not REVISION_PATTERN.fullmatch(revision):
+        raise SourceRejected("source revision must be a full lowercase Git SHA")
+    return {
+        "kind": SOURCE_KIND,
+        "repository": repository,
+        "revision": revision,
+        **(
+            {"observed_ref": str(source_ref["observed_ref"])}
+            if source_ref.get("observed_ref")
+            else {}
+        ),
+        **(
+            {"observed_at": str(source_ref["observed_at"])}
+            if source_ref.get("observed_at")
+            else {}
+        ),
+    }
+
+
+class _SameHostRedirect(urllib.request.HTTPRedirectHandler):
+    def __init__(self, allowed_scheme: str, allowed_netloc: str) -> None:
+        super().__init__()
+        self.allowed_scheme = allowed_scheme
+        self.allowed_netloc = allowed_netloc
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
+        target = urllib.parse.urlsplit(newurl)
+        if (
+            target.scheme != self.allowed_scheme
+            or target.netloc != self.allowed_netloc
+        ):
+            raise SourceRejected("source redirect left the configured Forgejo host")
+        return super().redirect_request(req, fp, code, msg, headers, newurl)
+
+
+def _download(url: str, destination: Path) -> tuple[str, int]:
+    max_bytes = _positive_int("SBOM_NEXUS_SOURCE_MAX_ARCHIVE_BYTES", 100 * 1024 * 1024)
+    timeout = _positive_int("SBOM_NEXUS_SOURCE_FETCH_TIMEOUT_SECONDS", 120)
+    parsed_url = urllib.parse.urlsplit(url)
+    if not parsed_url.hostname:
+        raise SourceRejected("configured Forgejo base URL has no host")
+    request = urllib.request.Request(url, headers={"User-Agent": "sbom-nexus/controlled-source-v1"})
+    opener = urllib.request.build_opener(
+        _SameHostRedirect(parsed_url.scheme, parsed_url.netloc)
+    )
+    digest = hashlib.sha256()
+    size = 0
+    started = time.monotonic()
+    try:
+        with opener.open(request, timeout=timeout) as response, destination.open("wb") as out:
+            while chunk := response.read(1024 * 1024):
+                if time.monotonic() - started > timeout:
+                    raise SourceRejected("source fetch exceeded its time limit")
+                size += len(chunk)
+                if size > max_bytes:
+                    raise SourceRejected("source archive exceeded its compressed size limit")
+                digest.update(chunk)
+                out.write(chunk)
+    except SourceRejected:
+        raise
+    except (OSError, urllib.error.HTTPError, urllib.error.URLError) as exc:
+        raise SourceUnavailable("pinned Forgejo source archive is unavailable") from exc
+    return digest.hexdigest(), size
+
+
+def _safe_member_path(name: str) -> PurePosixPath:
+    path = PurePosixPath(name)
+    if not name or path.is_absolute() or ".." in path.parts:
+        raise SourceRejected("source archive contains an unsafe path")
+    return path
+
+
+def _extract(archive: Path, destination: Path) -> Path:
+    max_members = _positive_int("SBOM_NEXUS_SOURCE_MAX_MEMBERS", 100_000)
+    max_expanded = _positive_int("SBOM_NEXUS_SOURCE_MAX_EXPANDED_BYTES", 512 * 1024 * 1024)
+    try:
+        bundle = tarfile.open(archive, mode="r:gz")
+    except (OSError, tarfile.TarError) as exc:
+        raise SourceRejected("source archive is not a valid gzip-compressed tar") from exc
+    with bundle:
+        members = bundle.getmembers()
+        if not members or len(members) > max_members:
+            raise SourceRejected("source archive member count is outside the allowed range")
+        expanded = 0
+        roots: set[str] = set()
+        checked: list[tuple[tarfile.TarInfo, PurePosixPath]] = []
+        for member in members:
+            path = _safe_member_path(member.name)
+            roots.add(path.parts[0])
+            if not (member.isdir() or member.isfile()):
+                raise SourceRejected("source archive contains a link or special file")
+            if member.isfile():
+                expanded += member.size
+                if expanded > max_expanded:
+                    raise SourceRejected("source archive exceeded its expanded size limit")
+            checked.append((member, path))
+        if len(roots) != 1:
+            raise SourceRejected("source archive must have exactly one top-level directory")
+
+        for member, relative in checked:
+            target = destination.joinpath(*relative.parts)
+            if member.isdir():
+                target.mkdir(parents=True, exist_ok=True)
+                continue
+            target.parent.mkdir(parents=True, exist_ok=True)
+            source = bundle.extractfile(member)
+            if source is None:
+                raise SourceRejected("source archive member could not be read")
+            with source, target.open("wb") as out:
+                shutil.copyfileobj(source, out, length=1024 * 1024)
+    return destination / next(iter(roots))
+
+
+@contextmanager
+def fetch_controlled_source(
+    repo_slug: str, source_ref: dict[str, Any]
+) -> Iterator[ControlledSource]:
+    if not controlled_source_enabled():
+        raise ControlledSourceDisabled("controlled source ingestion is disabled")
+    validated = validate_source_ref(repo_slug, source_ref)
+    base = os.getenv(
+        "SBOM_NEXUS_FORGEJO_BASE_URL",
+        "https://forgejo.coulomb.social",
+    ).rstrip("/")
+    parsed = urllib.parse.urlsplit(base)
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+        raise SourceRejected("configured Forgejo base URL is invalid")
+    repository = validated["repository"]
+    revision = validated["revision"]
+    url = f"{base}/{repository}/archive/{revision}.tar.gz"
+    temp_parent = Path(os.getenv("SBOM_NEXUS_SOURCE_TMP", "/tmp/sbom-nexus-sources"))
+    temp_parent.mkdir(parents=True, exist_ok=True)
+    wait_seconds = _positive_int("SBOM_NEXUS_SOURCE_SLOT_TIMEOUT_SECONDS", 240)
+    if not _SOURCE_SCAN_SLOT.acquire(timeout=wait_seconds):
+        raise SourceRejected("controlled source scan slot wait exceeded its time limit")
+    try:
+        with tempfile.TemporaryDirectory(prefix="scan-", dir=temp_parent) as raw_temp:
+            temp = Path(raw_temp)
+            archive = temp / "source.tar.gz"
+            extracted = temp / "extracted"
+            extracted.mkdir()
+            archive_sha256, archive_bytes = _download(url, archive)
+            try:
+                root = _extract(archive, extracted)
+            except SourceRejected:
+                raise
+            except OSError as exc:
+                raise SourceRejected("source archive could not be safely extracted") from exc
+            yield ControlledSource(
+                root=root,
+                provenance={
+                    **validated,
+                    "archive_sha256": archive_sha256,
+                    "archive_bytes": archive_bytes,
+                },
+            )
+    finally:
+        _SOURCE_SCAN_SLOT.release()
--- a/src/sbom_nexus/storage.py
+++ b/src/sbom_nexus/storage.py
@ -2,6 +2,7 @@

 from __future__ import annotations

+import time
 import uuid
 from collections import defaultdict
 from datetime import UTC, datetime, timedelta
@ -10,10 +11,12 @@ from typing import Any

 from sqlalchemy import func, insert, select, update
 from sqlalchemy.engine import RowMapping
+from sqlalchemy.exc import IntegrityError

 from sbom_nexus.database import (
    create_database_engine,
    metadata,
+    operation_receipts,
    repositories,
    snapshots,
 )
@ -23,15 +26,17 @@ from sbom_nexus.scanner import is_copyleft
 SUCCESS_STATUSES = frozenset({"ingested", "imported"})


+class OperationKeyConflict(ValueError):
+    """An operation key was reused for a different mutation request."""
+
+
 def utc_now() -> datetime:
    return datetime.now(UTC)


 def parse_datetime(value: datetime | str) -> datetime:
    parsed = (
-        datetime.fromisoformat(value.replace("Z", "+00:00"))
-        if isinstance(value, str)
-        else value
+        datetime.fromisoformat(value.replace("Z", "+00:00")) if isinstance(value, str) else value
    )
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
@ -64,6 +69,7 @@ class Store:
        slug: str,
        *,
        checkout_path: str | None = None,
+        source_ref: dict[str, Any] | None = None,
        active: bool = True,
        last_attempt_at: datetime | str | None = None,
        last_success_at: datetime | str | None = None,
@ -72,15 +78,18 @@ class Store:
        attempt = parse_datetime(last_attempt_at) if last_attempt_at else None
        success = parse_datetime(last_success_at) if last_success_at else None
        with self.engine.begin() as connection:
-            existing = connection.execute(
-                select(repositories).where(repositories.c.slug == slug)
-            ).mappings().one_or_none()
+            existing = (
+                connection.execute(select(repositories).where(repositories.c.slug == slug))
+                .mappings()
+                .one_or_none()
+            )
            if existing:
                connection.execute(
                    update(repositories)
                    .where(repositories.c.slug == slug)
                    .values(
                        checkout_path=checkout_path,
+                        source_ref_json=source_ref,
                        active=active,
                        last_attempt_at=attempt or existing["last_attempt_at"],
                        last_success_at=success or existing["last_success_at"],
@ -93,6 +102,7 @@ class Store:
                        id=str(uuid.uuid4()),
                        slug=slug,
                        checkout_path=checkout_path,
+                        source_ref_json=source_ref,
                        active=active,
                        last_attempt_at=attempt,
                        last_success_at=success,
@ -102,25 +112,61 @@ class Store:
                        updated_at=now,
                    )
                )
-            row = connection.execute(
-                select(repositories).where(repositories.c.slug == slug)
-            ).mappings().one()
+            row = (
+                connection.execute(select(repositories).where(repositories.c.slug == slug))
+                .mappings()
+                .one()
+            )
        return self._repository_dict(row)

    def get_repository(self, slug: str) -> dict[str, Any] | None:
        with self.engine.connect() as connection:
-            row = connection.execute(
-                select(repositories).where(repositories.c.slug == slug)
-            ).mappings().one_or_none()
+            row = (
+                connection.execute(select(repositories).where(repositories.c.slug == slug))
+                .mappings()
+                .one_or_none()
+            )
        return self._repository_dict(row) if row else None

    def list_repositories(self) -> list[dict[str, Any]]:
        with self.engine.connect() as connection:
-            rows = connection.execute(
-                select(repositories).order_by(repositories.c.slug)
-            ).mappings().all()
+            rows = (
+                connection.execute(select(repositories).order_by(repositories.c.slug))
+                .mappings()
+                .all()
+            )
        return [self._repository_dict(row) for row in rows]

+    def replay_operation(
+        self, operation_key: str | None, request_fingerprint: str | None
+    ) -> dict[str, Any] | None:
+        if operation_key is None:
+            return None
+        if request_fingerprint is None:
+            raise ValueError("request_fingerprint is required with operation_key")
+        with self.engine.connect() as connection:
+            receipt = (
+                connection.execute(
+                    select(operation_receipts).where(
+                        operation_receipts.c.operation_key == operation_key
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if receipt is None:
+                return None
+            if receipt["request_fingerprint"] != request_fingerprint:
+                raise OperationKeyConflict("operation key was already used for a different request")
+            snapshot = (
+                connection.execute(
+                    select(snapshots).where(snapshots.c.id == receipt["snapshot_id"])
+                )
+                .mappings()
+                .one()
+            )
+        return self._snapshot_dict(snapshot)
+
    def record_snapshot(
        self,
        repo_slug: str,
@ -130,83 +176,174 @@ class Store:
        source: str,
        snapshot_at: datetime | str | None = None,
        source_revision: str | None = None,
+        source_provenance: dict[str, Any] | None = None,
        sources: list[dict[str, Any]] | None = None,
        errors: list[dict[str, Any]] | None = None,
        snapshot_id: str | None = None,
        legacy_id: str | None = None,
+        operation_key: str | None = None,
+        request_fingerprint: str | None = None,
    ) -> tuple[dict[str, Any], bool]:
+        if bool(operation_key) != bool(request_fingerprint):
+            raise ValueError("operation_key and request_fingerprint must be supplied together")
        timestamp = parse_datetime(snapshot_at or utc_now())
        created_at = utc_now()
-        with self.engine.begin() as connection:
-            repo = connection.execute(
-                select(repositories).where(repositories.c.slug == repo_slug)
-            ).mappings().one_or_none()
-            if repo is None:
-                raise KeyError(repo_slug)
-            if legacy_id:
-                existing = connection.execute(
-                    select(snapshots).where(snapshots.c.legacy_id == legacy_id)
-                ).mappings().one_or_none()
-                if existing:
-                    return self._snapshot_dict(existing), False
-
-            new_snapshot_id = snapshot_id or str(uuid.uuid4())
-            connection.execute(
-                insert(snapshots).values(
-                    id=new_snapshot_id,
-                    repo_id=repo["id"],
-                    snapshot_at=timestamp,
-                    source=source,
+        try:
+            with self.engine.begin() as connection:
+                return self._record_snapshot_transaction(
+                    connection,
+                    repo_slug=repo_slug,
+                    entries=entries,
                    status=status,
-                    entry_count=len(entries),
+                    source=source,
+                    timestamp=timestamp,
+                    created_at=created_at,
                    source_revision=source_revision,
-                    sources_json=sources or [],
-                    errors_json=errors or [],
+                    source_provenance=source_provenance,
+                    sources=sources,
+                    errors=errors,
+                    snapshot_id=snapshot_id,
                    legacy_id=legacy_id,
+                    operation_key=operation_key,
+                    request_fingerprint=request_fingerprint,
+                )
+        except IntegrityError:
+            if not operation_key or not request_fingerprint:
+                raise
+            for _attempt in range(20):
+                replay = self.replay_operation(operation_key, request_fingerprint)
+                if replay is not None:
+                    return replay, False
+                time.sleep(0.05)
+            raise
+
+    def _record_snapshot_transaction(
+        self,
+        connection,
+        *,
+        repo_slug: str,
+        entries: list[dict[str, Any]],
+        status: str,
+        source: str,
+        timestamp: datetime,
+        created_at: datetime,
+        source_revision: str | None,
+        source_provenance: dict[str, Any] | None,
+        sources: list[dict[str, Any]] | None,
+        errors: list[dict[str, Any]] | None,
+        snapshot_id: str | None,
+        legacy_id: str | None,
+        operation_key: str | None,
+        request_fingerprint: str | None,
+    ) -> tuple[dict[str, Any], bool]:
+        repo = (
+            connection.execute(select(repositories).where(repositories.c.slug == repo_slug))
+            .mappings()
+            .one_or_none()
+        )
+        if repo is None:
+            raise KeyError(repo_slug)
+        if operation_key:
+            receipt = (
+                connection.execute(
+                    select(operation_receipts).where(
+                        operation_receipts.c.operation_key == operation_key
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if receipt:
+                if receipt["request_fingerprint"] != request_fingerprint:
+                    raise OperationKeyConflict(
+                        "operation key was already used for a different request"
+                    )
+                existing = (
+                    connection.execute(
+                        select(snapshots).where(snapshots.c.id == receipt["snapshot_id"])
+                    )
+                    .mappings()
+                    .one()
+                )
+                return self._snapshot_dict(existing), False
+        if legacy_id:
+            existing = (
+                connection.execute(select(snapshots).where(snapshots.c.legacy_id == legacy_id))
+                .mappings()
+                .one_or_none()
+            )
+            if existing:
+                return self._snapshot_dict(existing), False
+
+        new_snapshot_id = snapshot_id or str(uuid.uuid4())
+        connection.execute(
+            insert(snapshots).values(
+                id=new_snapshot_id,
+                repo_id=repo["id"],
+                snapshot_at=timestamp,
+                source=source,
+                status=status,
+                entry_count=len(entries),
+                source_revision=source_revision,
+                source_provenance_json=source_provenance,
+                sources_json=sources or [],
+                errors_json=errors or [],
+                legacy_id=legacy_id,
+                created_at=created_at,
+            )
+        )
+        if entries:
+            connection.execute(
+                insert(entry_table),
+                [
+                    {
+                        "id": str(uuid.uuid4()),
+                        "repo_id": repo["id"],
+                        "snapshot_id": new_snapshot_id,
+                        "package_name": entry["package_name"],
+                        "package_version": entry.get("package_version"),
+                        "ecosystem": entry["ecosystem"],
+                        "license_spdx": entry.get("license_spdx"),
+                        "is_direct": bool(entry.get("is_direct", True)),
+                        "is_dev": bool(entry.get("is_dev", False)),
+                        "source_path": entry.get("source_path"),
+                        "created_at": created_at,
+                    }
+                    for entry in entries
+                ],
+            )
+
+        previous_attempt = repo["last_attempt_at"]
+        is_latest = previous_attempt is None or timestamp >= parse_datetime(previous_attempt)
+        if is_latest:
+            last_success = repo["last_success_at"]
+            if status in SUCCESS_STATUSES:
+                last_success = timestamp
+            connection.execute(
+                update(repositories)
+                .where(repositories.c.id == repo["id"])
+                .values(
+                    last_attempt_at=timestamp,
+                    last_success_at=last_success,
+                    last_status=status,
+                    last_source=source,
+                    updated_at=created_at,
+                )
+            )
+        if operation_key:
+            connection.execute(
+                insert(operation_receipts).values(
+                    operation_key=operation_key,
+                    request_fingerprint=request_fingerprint,
+                    snapshot_id=new_snapshot_id,
                    created_at=created_at,
                )
            )
-            if entries:
-                connection.execute(
-                    insert(entry_table),
-                    [
-                        {
-                            "id": str(uuid.uuid4()),
-                            "repo_id": repo["id"],
-                            "snapshot_id": new_snapshot_id,
-                            "package_name": entry["package_name"],
-                            "package_version": entry.get("package_version"),
-                            "ecosystem": entry["ecosystem"],
-                            "license_spdx": entry.get("license_spdx"),
-                            "is_direct": bool(entry.get("is_direct", True)),
-                            "is_dev": bool(entry.get("is_dev", False)),
-                            "source_path": entry.get("source_path"),
-                            "created_at": created_at,
-                        }
-                        for entry in entries
-                    ],
-                )
-
-            previous_attempt = repo["last_attempt_at"]
-            is_latest = previous_attempt is None or timestamp >= parse_datetime(previous_attempt)
-            if is_latest:
-                last_success = repo["last_success_at"]
-                if status in SUCCESS_STATUSES:
-                    last_success = timestamp
-                connection.execute(
-                    update(repositories)
-                    .where(repositories.c.id == repo["id"])
-                    .values(
-                        last_attempt_at=timestamp,
-                        last_success_at=last_success,
-                        last_status=status,
-                        last_source=source,
-                        updated_at=created_at,
-                    )
-                )
-            row = connection.execute(
-                select(snapshots).where(snapshots.c.id == new_snapshot_id)
-            ).mappings().one()
+        row = (
+            connection.execute(select(snapshots).where(snapshots.c.id == new_snapshot_id))
+            .mappings()
+            .one()
+        )
        return self._snapshot_dict(row), True

    def list_snapshots(self, repo_slug: str | None = None) -> list[dict[str, Any]]:
@ -231,11 +368,15 @@ class Store:
            row = connection.execute(statement).mappings().one_or_none()
            if row is None:
                return None
-            entry_rows = connection.execute(
-                select(entry_table)
-                .where(entry_table.c.snapshot_id == snapshot_id)
-                .order_by(entry_table.c.package_name)
-            ).mappings().all()
+            entry_rows = (
+                connection.execute(
+                    select(entry_table)
+                    .where(entry_table.c.snapshot_id == snapshot_id)
+                    .order_by(entry_table.c.package_name)
+                )
+                .mappings()
+                .all()
+            )
        result = self._snapshot_dict(row)
        result["entries"] = []
        for entry in entry_rows:
@ -299,9 +440,7 @@ class Store:

    def licence_report(self) -> dict[str, Any]:
        current_entries = self.latest_entries()
-        groups: dict[str | None, dict[str, Any]] = defaultdict(
-            lambda: {"count": 0, "repos": set()}
-        )
+        groups: dict[str | None, dict[str, Any]] = defaultdict(lambda: {"count": 0, "repos": set()})
        risks: list[dict[str, Any]] = []
        for entry in current_entries:
            license_id = entry.get("license_spdx")
@ -317,9 +456,7 @@ class Store:
                        "source_path": entry.get("source_path"),
                    }
                )
-        sorted_groups = sorted(
-            groups.items(), key=lambda item: (-item[1]["count"], item[0] or "")
-        )
+        sorted_groups = sorted(groups.items(), key=lambda item: (-item[1]["count"], item[0] or ""))
        return {
            "groups": [
                {
@ -350,8 +487,7 @@ class Store:
        stale = [
            repo
            for repo in active_repositories
-            if repo["last_attempt_at"] is None
-            or parse_datetime(repo["last_attempt_at"]) < cutoff
+            if repo["last_attempt_at"] is None or parse_datetime(repo["last_attempt_at"]) < cutoff
        ]
        stale.sort(
            key=lambda repo: (
@ -390,6 +526,7 @@ class Store:
            "id": row["id"],
            "slug": row["slug"],
            "checkout_path": row["checkout_path"],
+            "source_ref": row["source_ref_json"],
            "active": bool(row["active"]),
            "last_attempt_at": datetime_text(row["last_attempt_at"]),
            "last_success_at": datetime_text(row["last_success_at"]),
@ -409,6 +546,7 @@ class Store:
            "status": row["status"],
            "entry_count": row["entry_count"],
            "source_revision": row["source_revision"],
+            "source_provenance": row["source_provenance_json"],
            "sources": row["sources_json"] or [],
            "errors": row["errors_json"] or [],
            "legacy_id": row["legacy_id"],
@ -444,6 +582,7 @@ class Store:
        attempt = repo["last_attempt_at"]
        age_days = 9999 if attempt is None else max(0, (now - parse_datetime(attempt)).days)
        checkout_path = repo["checkout_path"]
+        source_ref = repo.get("source_ref")
        return {
            "repo_slug": repo["slug"],
            "last_sbom_at": attempt,
@ -451,6 +590,10 @@ class Store:
            "last_success_at": repo["last_success_at"],
            "sbom_age_days": age_days,
            "has_sbom": repo["last_success_at"] is not None,
-            "checkout_available": bool(checkout_path and Path(checkout_path).is_dir()),
+            "checkout_available": bool(
+                source_ref or (checkout_path and Path(checkout_path).is_dir())
+            ),
+            "source_available": source_ref is not None,
+            "source_ref": source_ref,
            "last_status": repo["last_status"],
        }
--- a/tests/test_api.py
+++ b/tests/test_api.py
@ -1,11 +1,14 @@
 from __future__ import annotations

+from contextlib import contextmanager
 from datetime import UTC, datetime, timedelta
 from pathlib import Path

 from fastapi.testclient import TestClient

+from sbom_nexus import api
 from sbom_nexus.api import create_app
+from sbom_nexus.source_fetch import ControlledSource


 def client_for(tmp_path: Path) -> TestClient:
@ -19,6 +22,7 @@ def register(
    checkout_path: str | None = None,
    last_sbom_at: datetime | None = None,
    active: bool = True,
+    source_ref: dict | None = None,
 ) -> None:
    response = client.put(
        f"/repositories/{slug}",
@ -26,6 +30,7 @@ def register(
            "checkout_path": checkout_path,
            "last_sbom_at": last_sbom_at.isoformat() if last_sbom_at else None,
            "active": active,
+            "source_ref": source_ref,
        },
    )
    assert response.status_code == 200
@ -135,6 +140,96 @@ def test_checkout_scan_records_provenance_and_success(tmp_path: Path) -> None:
    assert detail["entries"][0]["snapshot_at"] == detail["snapshot_at"]


+def test_ingest_operation_key_replays_without_a_second_snapshot(tmp_path: Path) -> None:
+    repo = tmp_path / "idempotent"
+    repo.mkdir()
+    (repo / "requirements.txt").write_text("fastapi==0.136.1\n", encoding="utf-8")
+    client = client_for(tmp_path)
+    register(client, "idempotent", checkout_path=str(repo))
+    headers = {
+        "Idempotency-Key": "activity-operation-1",
+        "X-Activity-Core-Operation-ID": "activity-operation-1",
+    }
+
+    first = client.post("/sbom/idempotent/ingest", headers=headers)
+    second = client.post("/sbom/idempotent/ingest", headers=headers)
+
+    assert first.status_code == 200
+    assert second.json() == first.json()
+    assert len(client.get("/sbom/snapshots/?repo_slug=idempotent").json()) == 1
+    conflict = client.post(
+        "/sbom/idempotent/skip",
+        headers=headers,
+        json={"reason": "no-checkout"},
+    )
+    assert conflict.status_code == 409
+
+
+def test_skip_operation_key_replays_original_outcome(tmp_path: Path) -> None:
+    client = client_for(tmp_path)
+    register(client, "skip-replay")
+    headers = {"Idempotency-Key": "skip-operation-1"}
+
+    first = client.post(
+        "/sbom/skip-replay/skip",
+        headers=headers,
+        json={"reason": "source-unavailable", "detail": "not projected"},
+    )
+    second = client.post(
+        "/sbom/skip-replay/skip",
+        headers=headers,
+        json={"reason": "source-unavailable", "detail": "not projected"},
+    )
+
+    assert first.status_code == 200
+    assert second.json() == first.json()
+    assert len(client.get("/sbom/snapshots/?repo_slug=skip-replay").json()) == 1
+
+
+def test_controlled_source_records_explicit_revision_and_archive_provenance(
+    tmp_path: Path, monkeypatch
+) -> None:
+    extracted = tmp_path / "extracted"
+    extracted.mkdir()
+    (extracted / "requirements.txt").write_text(
+        "fastapi==0.136.1\n", encoding="utf-8"
+    )
+    revision = "a" * 40
+    source_ref = {
+        "kind": "forgejo-archive-v1",
+        "repository": "coulomb/controlled",
+        "revision": revision,
+        "observed_ref": "refs/heads/main",
+        "observed_at": "2026-08-22T20:00:00Z",
+    }
+
+    @contextmanager
+    def fake_fetch(repo_slug: str, selected: dict):
+        assert repo_slug == "controlled"
+        assert selected["revision"] == revision
+        yield ControlledSource(
+            root=extracted,
+            provenance={**selected, "archive_sha256": "b" * 64, "archive_bytes": 42},
+        )
+
+    monkeypatch.setattr(api, "fetch_controlled_source", fake_fetch)
+    client = client_for(tmp_path)
+    register(client, "controlled", source_ref=source_ref)
+
+    result = client.post(
+        "/sbom/controlled/ingest",
+        json={"source_ref": source_ref},
+        headers={"Idempotency-Key": "controlled-operation-1"},
+    )
+
+    assert result.status_code == 200
+    assert result.json()["source_revision"] == revision
+    detail = client.get(f"/sbom/snapshots/{result.json()['snapshot_id']}").json()
+    assert detail["source"] == "forgejo-archive-v1"
+    assert detail["source_provenance"]["archive_sha256"] == "b" * 64
+    assert detail["sources"][0]["path"] == "requirements.txt"
+
+
 def test_historical_import_is_idempotent(tmp_path: Path) -> None:
    client = client_for(tmp_path)
    payload = {
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@ -23,9 +23,16 @@ def test_initial_migration_upgrades_and_downgrades(tmp_path: Path) -> None:
    assert set(inspect(engine).get_table_names()) == {
        "alembic_version",
        "entries",
+        "operation_receipts",
        "repositories",
        "snapshots",
    }
+    assert "source_ref_json" in {
+        column["name"] for column in inspect(engine).get_columns("repositories")
+    }
+    assert "source_provenance_json" in {
+        column["name"] for column in inspect(engine).get_columns("snapshots")
+    }
    assert {index["name"] for index in inspect(engine).get_indexes("entries")} == {
        "ix_entries_license",
        "ix_entries_repo",
--- a/tests/test_projection.py
+++ b/tests/test_projection.py
@ -12,6 +12,9 @@ def test_projection_sync_is_projection_only_and_reconciles(monkeypatch) -> None:
            "status": "active",
            "local_path": "/fallback/active",
            "host_paths": {"build-host": "/repos/active"},
+            "remote_url": "forgejo-remote:coulomb/active-repo.git",
+            "git_fingerprint": "a" * 40,
+            "last_state_synced_at": "2026-08-22T20:00:00Z",
        },
        {
            "slug": "retired-repo",
@ -51,10 +54,13 @@ def test_projection_sync_is_projection_only_and_reconciles(monkeypatch) -> None:
    assert result["counts"] == {
        "active": 1,
        "with_checkout_path": 2,
+        "with_source_ref": 1,
        "inactive": 1,
+        "without_source_ref": 1,
    }
    assert result["reconciliation"]["matched_repo_count"] == 2
    assert target["active-repo"]["checkout_path"] == "/repos/active"
+    assert target["active-repo"]["source_ref"]["revision"] == "a" * 40
    assert target["retired-repo"]["active"] is False
    assert all("/sbom/" not in path for _, _, path in calls)

--- a/tests/test_source_fetch.py
+++ b/tests/test_source_fetch.py
@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import io
+import tarfile
+from pathlib import Path
+
+import pytest
+
+from sbom_nexus.source_fetch import SourceRejected, _extract, validate_source_ref
+
+
+def _archive(path: Path, members: dict[str, bytes]) -> None:
+    with tarfile.open(path, "w:gz") as bundle:
+        for name, content in members.items():
+            info = tarfile.TarInfo(name)
+            info.size = len(content)
+            bundle.addfile(info, io.BytesIO(content))
+
+
+def test_validate_source_ref_binds_repository_to_slug() -> None:
+    revision = "a" * 40
+    assert validate_source_ref(
+        "demo",
+        {
+            "kind": "forgejo-archive-v1",
+            "repository": "coulomb/demo",
+            "revision": revision,
+        },
+    )["revision"] == revision
+    with pytest.raises(SourceRejected, match="does not match"):
+        validate_source_ref(
+            "demo",
+            {
+                "kind": "forgejo-archive-v1",
+                "repository": "coulomb/other",
+                "revision": revision,
+            },
+        )
+
+
+def test_extract_rejects_path_traversal(tmp_path: Path) -> None:
+    archive = tmp_path / "unsafe.tar.gz"
+    _archive(archive, {"repo/../../escaped": b"nope"})
+
+    with pytest.raises(SourceRejected, match="unsafe path"):
+        _extract(archive, tmp_path / "out")
+
+
+def test_extract_accepts_one_regular_root(tmp_path: Path) -> None:
+    archive = tmp_path / "safe.tar.gz"
+    _archive(archive, {"repo/requirements.txt": b"fastapi==0.136.1\n"})
+    destination = tmp_path / "out"
+    destination.mkdir()
+
+    root = _extract(archive, destination)
+
+    assert root == destination / "repo"
+    assert (root / "requirements.txt").read_text() == "fastapi==0.136.1\n"
--- a/workplans/SBOM-WP-0003-controlled-source-and-replay.md
+++ b/workplans/SBOM-WP-0003-controlled-source-and-replay.md
@ -0,0 +1,87 @@
+---
+id: SBOM-WP-0003
+type: workplan
+title: "Controlled Forgejo source ingestion and durable operation replay"
+domain: infotech
+repo: sbom-nexus
+status: active
+owner: codex
+topic_slug: infotech
+created: "2026-08-22"
+updated: "2026-08-22"
+quality_dor: DoR-Ok
+quality_dor_at: "2026-08-22"
+quality_dor_by: codex
+quality_dor_note: "CUST-WP-0064 selected a full-SHA public Forgejo archive contract with bounded extraction, explicit provenance, owner handoffs, idempotency, failure semantics, acceptance evidence, and rollback."
+parent_workplan: CUST-WP-0064
+related:
+  - CUST-IN-0013
+  - ACTIVITY-WP-0033
+  - RMGR-WP-0011
+---
+
+# Controlled Forgejo source ingestion and durable operation replay
+
+## Implement durable operation receipts
+
+```task
+id: SBOM-WP-0003-T01
+status: done
+priority: high
+```
+
+Enforce supplied `Idempotency-Key` / `X-Activity-Core-Operation-ID` values on
+repository ingest and skip. Persist a request fingerprint and snapshot link in
+the same transaction, replay the original terminal outcome, and reject key
+reuse for a different operation.
+
+Completed with migration `0002`, transactional operation receipts, early
+replay before source work, request-conflict HTTP 409 behavior, and ingest/skip
+tests proving one snapshot across duplicate requests.
+
+## Add controlled full-SHA source ingestion
+
+```task
+id: SBOM-WP-0003-T02
+status: done
+priority: high
+```
+
+Consume the `forgejo-archive-v1` source reference selected in
+`the-custodian/docs/sbom-controlled-scan-input-contract-v1.md`. Validate the
+identity, stream and safely extract within fixed limits, pass the explicit
+revision into the scanner, persist archive provenance, and always clean up.
+
+Completed with strict Coulomb identity/full-SHA validation, same-host fetches,
+streaming compressed limits, safe regular-file-only extraction, one scan slot,
+subprocess scan timeout, explicit revision override, archive provenance, and
+temporary-directory cleanup. A real Forgejo archive scan produced 33 entries
+from one manifest with zero errors.
+
+## Extend repository projection and outcomes
+
+```task
+id: SBOM-WP-0003-T03
+status: done
+priority: high
+```
+
+Store and return source references in repository/catch-up projections. Add
+terminal `source-unavailable` and `source-rejected` outcomes without changing
+oldest-N ranking or success-time semantics.
+
+Completed in the repository projection, API model, storage schema, catch-up
+response, and additive skip handling. Legacy checkout scanning remains
+available for local/operator compatibility while the production flag is dark.
+
+## Prove package integration and production behavior
+
+```task
+id: SBOM-WP-0003-T04
+status: wait
+priority: high
+```
+
+Coordinate the schema migration, ephemeral volume, Forgejo-only egress, and
+feature flag with `rapp-sbom-nexus`; then pass unit/integration tests and the
+attended plus scheduled production proof owned by CUST-WP-0064.