Implement bounded SBOM Nexus catch-up

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834

activity-definitions/daily-sbom-catchup.md

@@ -16,6 +16,7 @@ context_sources:
     required: true
     params:
       limit: 3          # catch_up_limit — operator knob, not a nexus constant
+      apply: true       # declared bounded side-effect; manual runs require confirmation
     bind_to: context.catchup
 # One ranked call returns only the N repos that have lacked a current SBOM the
 # longest (never-scanned first) plus fleet counts. There is deliberately no
@@ -24,10 +25,9 @@ context_sources:
 
 # Daily SBOM Catch-up
 
-> **Disabled until CUST-WP-0062-T03 lands.** The `sbom-nexus` catch-up API does
-> not exist yet; the resolver contract is exercised against a test double in
-> `tests/test_sbom_nexus_context_resolver.py`. Do not enable this schedule
-> against the old per-repo `/sbom/{slug}` walk (CUST-WP-0062-T05).
+> **Disabled until production enable evidence is ready.** The `sbom-nexus`
+> ranked and terminal ingest/skip APIs are now available. Keep this definition
+> off until cluster reachability and a bounded manual fire are proven.
 
 Replaces `weekly-sbom-staleness` / `flag-stale-sbom` (ACTIVITY-WP-0030).
 The weekly check reported the backlog — 111 / 111 repos stale on 2026-08-18,
@@ -79,18 +79,20 @@ The progress event names the repos selected, updated, and skipped with a
 reason (`no-checkout`, `no-manifest`, `ingest-error`), plus the fleet counters
 so `never_count` can be watched declining day over day.
 
-## Not yet implemented (ACTIVITY-WP-0030-T02)
+## Bounded side-effect (ACTIVITY-WP-0030-T02)
 
-The bounded ingest side-effect — calling sbom-nexus ingest for each of the N
-selected repos using the registered checkout — waits on CUST-WP-0062-T03. Until
-then this definition resolves and reports only. When T02 lands, a recorded skip
-must advance the queue position so an impossible repo does not permanently head
-it.
+`params.apply: true` calls sbom-nexus for each of the already-truncated N
+targets. A target with an explicitly unavailable checkout is recorded through
+`POST /sbom/{slug}/skip` as `no-checkout`; other targets use the terminal ingest
+route, which returns `ingested`, `no-manifest`, or `ingest-error`. Transport and
+contract failures are recorded as `ingest-error`. The resulting `updated` and
+`skipped` arrays are part of the run context and progress report. There is no
+task or issue emission.
 
 ## Enable checklist
 
 1. CUST-WP-0062-T02/T03 done: `sbom-nexus` stood up, `GET /sbom/catch-up`
-   returns oldest-N in one call.
+   returns oldest-N in one call. **Done 2026-08-22.**
 2. `SBOM_NEXUS_URL` reachable from the railiance01 worker.
 3. ACTIVITY-WP-0030-T02 ingest side-effect implemented and dry-run proven.
 4. `weekly-sbom-staleness` confirmed off in source **and** production

src/activity_core/context_resolvers/sbom_nexus.py

@@ -36,9 +36,10 @@ Until CUST-WP-0062-T03 lands there is no live endpoint — the query is exercise
 against a test double (``tests/test_sbom_nexus_context_resolver.py``) and the
 daily definition stays ``enabled: false``.
 
-This adapter is read-only. Ingest of the selected repos is a declared bounded
-side-effect owned by ACTIVITY-WP-0030-T02 and is deliberately not implemented
-here.
+With ``params.apply: true`` the adapter performs the declared T02 side-effect:
+each selected repository receives exactly one terminal ingest or skip outcome.
+The ranked response is truncated before any write, so the number of processed
+repositories can never exceed ``limit``. The default remains read-only.
 
 Config: SBOM_NEXUS_URL env var (default: http://127.0.0.1:8010).
 """
@@ -47,6 +48,7 @@ from __future__ import annotations
 
 import os
 from typing import Any
+from urllib.parse import quote
 
 import httpx
 
@@ -84,6 +86,14 @@ def _fetch_json(path: str, params: dict[str, Any] | None = None) -> Any:
         return response.json()
 
 
+def _post_json(path: str, payload: dict[str, Any] | None = None) -> Any:
+    url = f"{_base_url()}{path}"
+    with httpx.Client(timeout=_TIMEOUT_SECONDS) as client:
+        response = client.post(url, json=payload)
+        response.raise_for_status()
+        return response.json()
+
+
 def _int_or(value: Any, default: int) -> int:
     try:
         return int(value)
@@ -147,7 +157,7 @@ def _catch_up(params: dict[str, Any]) -> dict[str, Any]:
     never_count = _int_or(payload.get("never_count"), 0)
     stale_count = _int_or(payload.get("stale_count"), len(repos))
 
-    return {
+    result = {
         "repos": repos,
         "selected_count": len(repos),
         "stale_count": stale_count,
@@ -155,6 +165,68 @@ def _catch_up(params: dict[str, Any]) -> dict[str, Any]:
         "total_count": total_count,
         "limit": limit,
     }
+    if params.get("apply") is True:
+        result.update(_apply_bounded_ingest(repos))
+    return result
+
+
+def _skip(repo_slug: str, reason: str, detail: str | None = None) -> dict[str, Any]:
+    payload: dict[str, Any] = {"reason": reason}
+    if detail:
+        payload["detail"] = detail[:300]
+    raw = _post_json(f"/sbom/{quote(repo_slug, safe='')}/skip", payload)
+    if not isinstance(raw, dict) or raw.get("status") != "skipped":
+        raise RuntimeError(f"sbom-nexus skip returned an invalid outcome for {repo_slug}")
+    return raw
+
+
+def _ingest(repo_slug: str) -> dict[str, Any]:
+    raw = _post_json(f"/sbom/{quote(repo_slug, safe='')}/ingest")
+    if not isinstance(raw, dict) or raw.get("status") not in {"ingested", "skipped"}:
+        return _skip(repo_slug, "ingest-error", "invalid ingest outcome")
+    return raw
+
+
+def _apply_bounded_ingest(repos: list[dict[str, Any]]) -> dict[str, Any]:
+    updated: list[dict[str, Any]] = []
+    skipped: list[dict[str, Any]] = []
+
+    for repo in repos:
+        repo_slug = str(repo["repo_slug"])
+        try:
+            if repo.get("checkout_available") is False:
+                outcome = _skip(repo_slug, "no-checkout")
+            else:
+                outcome = _ingest(repo_slug)
+        except Exception as exc:
+            # A transport or contract failure still needs a terminal Nexus
+            # outcome so the same impossible repository cannot pin the queue.
+            outcome = _skip(repo_slug, "ingest-error", type(exc).__name__)
+
+        compact = {
+            key: outcome.get(key)
+            for key in (
+                "repo_slug",
+                "status",
+                "reason",
+                "snapshot_id",
+                "entry_count",
+                "snapshot_at",
+                "source_revision",
+            )
+            if outcome.get(key) is not None
+        }
+        compact.setdefault("repo_slug", repo_slug)
+        if outcome.get("status") == "ingested":
+            updated.append(compact)
+        else:
+            skipped.append(compact)
+
+    return {
+        "attempted_count": len(repos),
+        "updated": updated,
+        "skipped": skipped,
+    }
 
 
 class SbomNexusContextResolver(ContextResolver):

tests/test_sbom_nexus_context_resolver.py

@@ -1,8 +1,4 @@
-"""sbom-nexus catch_up resolver (ACTIVITY-WP-0030-T01).
-
-CUST-WP-0062-T03 has not landed, so the contract is exercised against a test
-double rather than a live nexus.
-"""
+"""sbom-nexus bounded catch-up resolver (ACTIVITY-WP-0030-T01/T02)."""
 
 from __future__ import annotations
 
@@ -40,6 +36,8 @@ class DummyClient:
         self._payload = payload
         self._status_code = status_code
         self.calls: list[tuple[str, dict[str, Any] | None]] = []
+        self.posts: list[tuple[str, dict[str, Any] | None]] = []
+        self.post_results: dict[str, list[DummyResponse]] = {}
 
     def __enter__(self) -> "DummyClient":
         return self
@@ -51,6 +49,13 @@ class DummyClient:
         self.calls.append((url, params))
         return DummyResponse(self._payload, self._status_code)
 
+    def post(self, url: str, json: dict[str, Any] | None = None) -> DummyResponse:
+        self.posts.append((url, json))
+        for suffix, results in self.post_results.items():
+            if url.endswith(suffix) and results:
+                return results.pop(0)
+        raise AssertionError(f"unexpected POST {url}")
+
 
 def _install(monkeypatch, payload: Any, status_code: int = 200) -> DummyClient:
     client = DummyClient(payload, status_code)
@@ -155,6 +160,127 @@ def test_catch_up_truncates_an_over_long_response(monkeypatch) -> None:
     assert result["selected_count"] == 2
 
 
+def test_apply_processes_at_most_limit_and_records_terminal_outcomes(monkeypatch) -> None:
+    repos = [
+        {
+            "repo_slug": "no-checkout",
+            "last_sbom_at": None,
+            "has_sbom": False,
+            "checkout_available": False,
+        },
+        {
+            "repo_slug": "scan-me",
+            "last_sbom_at": None,
+            "has_sbom": False,
+            "checkout_available": True,
+        },
+        {
+            "repo_slug": "not-selected",
+            "last_sbom_at": None,
+            "has_sbom": False,
+            "checkout_available": True,
+        },
+    ]
+    client = _install(monkeypatch, _payload(repos))
+    client.post_results = {
+        "/sbom/no-checkout/skip": [
+            DummyResponse(
+                {
+                    "repo_slug": "no-checkout",
+                    "status": "skipped",
+                    "reason": "no-checkout",
+                    "snapshot_id": "skip-1",
+                }
+            )
+        ],
+        "/sbom/scan-me/ingest": [
+            DummyResponse(
+                {
+                    "repo_slug": "scan-me",
+                    "status": "ingested",
+                    "snapshot_id": "scan-1",
+                    "entry_count": 12,
+                }
+            )
+        ],
+    }
+
+    result = SbomNexusContextResolver().resolve(
+        "catch_up", None, {"limit": 2, "apply": True}
+    )
+
+    assert result["attempted_count"] == 2
+    assert result["updated"] == [
+        {
+            "repo_slug": "scan-me",
+            "status": "ingested",
+            "snapshot_id": "scan-1",
+            "entry_count": 12,
+        }
+    ]
+    assert result["skipped"] == [
+        {
+            "repo_slug": "no-checkout",
+            "status": "skipped",
+            "reason": "no-checkout",
+            "snapshot_id": "skip-1",
+        }
+    ]
+    assert len(client.posts) == 2
+    assert all("not-selected" not in url for url, _body in client.posts)
+
+
+def test_apply_records_ingest_error_when_ingest_fails(monkeypatch) -> None:
+    client = _install(
+        monkeypatch,
+        _payload(
+            [
+                {
+                    "repo_slug": "broken",
+                    "last_sbom_at": None,
+                    "has_sbom": False,
+                    "checkout_available": True,
+                }
+            ]
+        ),
+    )
+    client.post_results = {
+        "/sbom/broken/ingest": [DummyResponse({}, status_code=503)],
+        "/sbom/broken/skip": [
+            DummyResponse(
+                {
+                    "repo_slug": "broken",
+                    "status": "skipped",
+                    "reason": "ingest-error",
+                    "snapshot_id": "skip-error-1",
+                }
+            )
+        ],
+    }
+
+    result = SbomNexusContextResolver().resolve(
+        "catch_up", None, {"limit": 3, "apply": True}
+    )
+
+    assert result["attempted_count"] == 1
+    assert result["updated"] == []
+    assert result["skipped"][0]["reason"] == "ingest-error"
+    assert client.posts[1][1] == {
+        "reason": "ingest-error",
+        "detail": "HTTPStatusError",
+    }
+
+
+def test_read_only_default_never_posts(monkeypatch) -> None:
+    client = _install(monkeypatch, _payload([{"repo_slug": "selected"}]))
+
+    result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 1})
+
+    assert result["selected_count"] == 1
+    assert "attempted_count" not in result
+    assert client.posts == []
+
+
 def test_catch_up_normalises_partial_entries(monkeypatch) -> None:
     _install(
         monkeypatch,
@@ -312,6 +438,7 @@ def test_daily_definition_is_bounded_and_disabled() -> None:
     assert source["type"] == "sbom-nexus"
     assert source["query"] == "catch_up"
     assert source["params"]["limit"] == 3
+    assert source["params"]["apply"] is True
     assert source["bind_to"] == "context.catchup"
     instruction = definition.instructions[0]
     assert instruction["model"] == "deterministic"