diff --git a/activity-definitions/daily-sbom-catchup.md b/activity-definitions/daily-sbom-catchup.md index 30f15fa..7d8a570 100644 --- a/activity-definitions/daily-sbom-catchup.md +++ b/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 diff --git a/src/activity_core/context_resolvers/sbom_nexus.py b/src/activity_core/context_resolvers/sbom_nexus.py index f2d8960..88de281 100644 --- a/src/activity_core/context_resolvers/sbom_nexus.py +++ b/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): diff --git a/tests/test_sbom_nexus_context_resolver.py b/tests/test_sbom_nexus_context_resolver.py index 8e5fba2..4bc77fc 100644 --- a/tests/test_sbom_nexus_context_resolver.py +++ b/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"