"""sbom-nexus bounded catch-up resolver (ACTIVITY-WP-0030-T01/T02).""" from __future__ import annotations from typing import Any import httpx import pytest from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY from activity_core.context_resolvers.sbom_nexus import ( SbomNexusContextResolver, apply_bounded_ingest, ) class DummyResponse: def __init__(self, payload: Any, status_code: int = 200) -> None: self._payload = payload self.status_code = status_code self.text = str(payload) def raise_for_status(self) -> None: if self.status_code >= 400: raise httpx.HTTPStatusError( f"HTTP {self.status_code}", request=httpx.Request("GET", "http://sbom-nexus.test"), response=None, # type: ignore[arg-type] ) def json(self) -> Any: return self._payload class DummyClient: """Records the single ranked call the contract allows.""" def __init__(self, payload: Any, status_code: int = 200) -> None: 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, dict[str, str]]] = [] self.post_results: dict[str, list[DummyResponse]] = {} def __enter__(self) -> "DummyClient": return self def __exit__(self, *args: Any) -> None: return None def get(self, url: str, params: dict[str, Any] | None = None) -> DummyResponse: self.calls.append((url, params)) return DummyResponse(self._payload, self._status_code) def post( self, url: str, json: dict[str, Any] | None = None, headers: dict[str, str] | None = None, ) -> DummyResponse: self.posts.append((url, json, headers or {})) 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) monkeypatch.setattr(httpx, "Client", lambda **kwargs: client) return client def _payload(repos: list[dict[str, Any]], **counts: Any) -> dict[str, Any]: base = {"stale_count": 111, "never_count": 93, "total_count": 111} base.update(counts) base["repos"] = repos return base def test_registered_as_sbom_nexus_source_type() -> None: assert CONTEXT_RESOLVER_REGISTRY["sbom-nexus"] is SbomNexusContextResolver def test_catch_up_returns_ranked_targets_and_fleet_counts(monkeypatch) -> None: client = _install( monkeypatch, _payload( [ { "repo_slug": "never-scanned", "last_sbom_at": None, "sbom_age_days": 9999, "has_sbom": False, "checkout_available": True, "source_ref": { "kind": "forgejo-archive-v1", "repository": "coulomb/never-scanned", "revision": "a" * 40, "observed_ref": "refs/heads/main", }, }, { "repo_slug": "activity-core", "last_sbom_at": "2026-04-26T00:00:00Z", "sbom_age_days": 117, "has_sbom": True, "checkout_available": True, }, { "repo_slug": "no-checkout", "last_sbom_at": "2026-05-01T00:00:00Z", "sbom_age_days": 112, "has_sbom": True, "checkout_available": False, }, ] ), ) result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 3}) # One ranked call, not a per-repo walk. assert len(client.calls) == 1 url, params = client.calls[0] assert url.endswith("/sbom/catch-up") assert params == {"limit": 3} assert [repo["repo_slug"] for repo in result["repos"]] == [ "never-scanned", "activity-core", "no-checkout", ] assert result["selected_count"] == 3 assert result["limit"] == 3 assert result["stale_count"] == 111 assert result["never_count"] == 93 assert result["total_count"] == 111 assert result["repos"][0]["has_sbom"] is False assert result["repos"][0]["source_ref"]["revision"] == "a" * 40 assert result["repos"][2]["checkout_available"] is False def test_catch_up_defaults_to_three(monkeypatch) -> None: client = _install(monkeypatch, _payload([])) result = SbomNexusContextResolver().resolve("catch_up", None, {}) assert client.calls[0][1] == {"limit": 3} assert result["limit"] == 3 assert result["repos"] == [] assert result["selected_count"] == 0 @pytest.mark.parametrize("requested", [0, -5, 500, "seven", None]) def test_catch_up_limit_stays_bounded(monkeypatch, requested: Any) -> None: _install(monkeypatch, _payload([])) result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": requested}) assert 1 <= result["limit"] <= 25 def test_catch_up_truncates_an_over_long_response(monkeypatch) -> None: """The bounded side-effect in T02 must never exceed the declared N.""" repos = [ {"repo_slug": f"repo-{index}", "last_sbom_at": None, "has_sbom": False} for index in range(10) ] _install(monkeypatch, _payload(repos)) result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 2}) assert len(result["repos"]) == 2 assert result["selected_count"] == 2 def test_declared_apply_remains_read_only_during_selection(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)) result = SbomNexusContextResolver().resolve( "catch_up", None, {"limit": 2, "apply": True} ) assert result["selected_count"] == 2 assert "attempted_count" not in result assert client.posts == [] def test_apply_processes_only_fixed_targets_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, }, ] client = _install(monkeypatch, _payload([])) 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 = apply_bounded_ingest( repos, operation_id="run-1", ) 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 first_key = client.posts[0][2]["Idempotency-Key"] assert first_key == client.posts[0][2]["X-Activity-Core-Operation-ID"] assert first_key != client.posts[1][2]["Idempotency-Key"] def test_apply_posts_frozen_source_ref_and_accepts_source_outcome(monkeypatch) -> None: source_ref = { "kind": "forgejo-archive-v1", "repository": "coulomb/scan-me", "revision": "b" * 40, "observed_ref": "refs/heads/main", } client = _install(monkeypatch, _payload([])) client.post_results = { "/sbom/scan-me/ingest": [ DummyResponse( { "repo_slug": "scan-me", "status": "skipped", "reason": "source-unavailable", "snapshot_id": "source-skip-1", } ) ] } result = apply_bounded_ingest( [ { "repo_slug": "scan-me", "checkout_available": False, "source_ref": source_ref, } ], operation_id="run-controlled-source", ) assert result["attempted_count"] == 1 assert result["skipped"][0]["reason"] == "source-unavailable" assert client.posts[0][1] == {"source_ref": source_ref} def test_ambiguous_ingest_failure_does_not_create_synthetic_skip(monkeypatch) -> None: client = _install( monkeypatch, _payload([]), ) client.post_results = { "/sbom/broken/ingest": [DummyResponse({}, status_code=503)], } with pytest.raises(httpx.HTTPStatusError): apply_bounded_ingest( [{"repo_slug": "broken", "checkout_available": True}], operation_id="run-ambiguous", ) assert len(client.posts) == 1 assert client.posts[0][0].endswith("/sbom/broken/ingest") def test_retry_resumes_heartbeat_outcomes(monkeypatch) -> None: repos = [ {"repo_slug": "first", "checkout_available": True}, {"repo_slug": "second", "checkout_available": True}, ] first_client = _install(monkeypatch, _payload([])) first_client.post_results = { "/sbom/first/ingest": [ DummyResponse( { "repo_slug": "first", "status": "ingested", "snapshot_id": "snapshot-first", } ) ], "/sbom/second/ingest": [DummyResponse({}, status_code=503)], } heartbeats: list[list[dict[str, Any]]] = [] with pytest.raises(httpx.HTTPStatusError): apply_bounded_ingest( repos, operation_id="run-retry", on_progress=lambda outcomes: heartbeats.append(outcomes), ) completed = heartbeats[-1] retry_client = _install(monkeypatch, _payload([])) retry_client.post_results = { "/sbom/second/ingest": [ DummyResponse( { "repo_slug": "second", "status": "ingested", "snapshot_id": "snapshot-second", } ) ] } result = apply_bounded_ingest( repos, operation_id="run-retry", completed=completed, ) assert [outcome["repo_slug"] for outcome in result["updated"]] == [ "first", "second", ] assert len(retry_client.posts) == 1 assert retry_client.posts[0][0].endswith("/sbom/second/ingest") def test_retry_reuses_stable_identity_for_same_run_and_repo(monkeypatch) -> None: repo = {"repo_slug": "ambiguous", "checkout_available": True} first_client = _install(monkeypatch, _payload([])) first_client.post_results = { "/sbom/ambiguous/ingest": [DummyResponse({}, status_code=503)], } with pytest.raises(httpx.HTTPStatusError): apply_bounded_ingest([repo], operation_id="run-stable") retry_client = _install(monkeypatch, _payload([])) retry_client.post_results = { "/sbom/ambiguous/ingest": [ DummyResponse( { "repo_slug": "ambiguous", "status": "ingested", "snapshot_id": "snapshot-ambiguous", } ) ], } apply_bounded_ingest([repo], operation_id="run-stable") first_headers = first_client.posts[0][2] retry_headers = retry_client.posts[0][2] assert retry_headers["Idempotency-Key"] == first_headers["Idempotency-Key"] assert retry_headers["X-Activity-Core-Operation-ID"] == first_headers[ "X-Activity-Core-Operation-ID" ] def test_apply_collapses_duplicate_targets(monkeypatch) -> None: client = _install(monkeypatch, _payload([])) client.post_results = { "/sbom/duplicate/skip": [ DummyResponse( { "repo_slug": "duplicate", "status": "skipped", "reason": "no-checkout", } ) ] } result = apply_bounded_ingest( [ {"repo_slug": "duplicate", "checkout_available": False}, {"repo_slug": "duplicate", "checkout_available": False}, ], operation_id="run-duplicates", ) assert result["attempted_count"] == 1 assert len(client.posts) == 1 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 == [] @pytest.mark.asyncio async def test_dedicated_activity_applies_truncated_workflow_selection(monkeypatch) -> None: from activity_core import activities captured: dict[str, Any] = {} def fake_apply(repos, *, operation_id, completed, on_progress): captured.update( repos=repos, operation_id=operation_id, completed=completed, ) return {"attempted_count": len(repos), "updated": [], "skipped": []} monkeypatch.setattr( "activity_core.context_resolvers.sbom_nexus.apply_bounded_ingest", fake_apply, ) patch = await activities.apply_sbom_catchup( { "run_id": "run-fixed", "context_sources": [ { "type": "sbom-nexus", "query": "catch_up", "bind_to": "context.catchup", "params": {"limit": 2, "apply": True}, } ], "context": { "catchup": { "limit": 2, "repos": [ {"repo_slug": "a"}, {"repo_slug": "b"}, {"repo_slug": "must-not-run"}, ], } }, } ) assert [repo["repo_slug"] for repo in captured["repos"]] == ["a", "b"] assert captured["operation_id"] == "run-fixed" assert patch["catchup"]["attempted_count"] == 2 @pytest.mark.asyncio async def test_dedicated_activity_forwards_heartbeat_outcomes(monkeypatch) -> None: from activity_core import activities completed = [ { "repo_slug": "already-done", "status": "skipped", "reason": "no-checkout", } ] captured: dict[str, Any] = {} monkeypatch.setattr( activities, "_sbom_heartbeat_state", lambda run_id: { "run_id": run_id, "outcomes_by_bind": {"catchup": completed}, }, ) def fake_apply(repos, *, operation_id, completed, on_progress): captured["completed"] = completed return {"attempted_count": 1, "updated": [], "skipped": completed} monkeypatch.setattr( "activity_core.context_resolvers.sbom_nexus.apply_bounded_ingest", fake_apply, ) await activities.apply_sbom_catchup( { "run_id": "run-resumed", "context_sources": [ { "type": "sbom-nexus", "query": "catch_up", "bind_to": "context.catchup", "params": {"limit": 1, "apply": True}, } ], "context": { "catchup": { "limit": 1, "repos": [{"repo_slug": "already-done"}], } }, } ) assert captured["completed"] == completed def test_catch_up_normalises_partial_entries(monkeypatch) -> None: _install( monkeypatch, _payload( [ {"repo_slug": "sparse"}, {"repo_slug": ""}, "not-an-object", {"last_sbom_at": "2026-01-01T00:00:00Z"}, ] ), ) result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 10}) assert [repo["repo_slug"] for repo in result["repos"]] == ["sparse"] sparse = result["repos"][0] assert sparse["has_sbom"] is False assert sparse["last_sbom_at"] is None assert sparse["sbom_age_days"] == 9999 assert sparse["checkout_available"] is None def test_catch_up_rejects_a_response_without_repos(monkeypatch) -> None: _install(monkeypatch, {"stale_count": 111}) with pytest.raises(RuntimeError, match="missing required key: repos"): SbomNexusContextResolver().resolve("catch_up", None, {}) def test_catch_up_rejects_a_non_object_response(monkeypatch) -> None: _install(monkeypatch, ["activity-core"]) with pytest.raises(RuntimeError, match="non-object response"): SbomNexusContextResolver().resolve("catch_up", None, {}) def test_unknown_query_returns_empty(monkeypatch) -> None: _install(monkeypatch, _payload([])) assert SbomNexusContextResolver().resolve("ingest", None, {}) == {} def test_base_url_is_configurable(monkeypatch) -> None: client = _install(monkeypatch, _payload([])) monkeypatch.setenv("SBOM_NEXUS_URL", "http://sbom-nexus.test:9000/") SbomNexusContextResolver().resolve("catch_up", None, {}) assert client.calls[0][0] == "http://sbom-nexus.test:9000/sbom/catch-up" def _instruction(): from activity_core.models import InstructionDef return InstructionDef.model_validate( { "id": "daily-sbom-catchup-report", "trusted_fields": [], "model": "deterministic", "temperature": 0, "max_tokens": 1, "prompt": "Deterministic SBOM catch-up report.", "output_schema": "", "review_advisory": False, "report_sinks": [ { "type": "state-hub-progress", "event_type": "sbom_catchup", "author": "activity-core", } ], } ) def test_deterministic_report_names_selected_repos() -> None: from activity_core.rules.executor import _deterministic_context_report context = { "catchup": { "repos": [ { "repo_slug": "never-scanned", "last_sbom_at": None, "sbom_age_days": 9999, "has_sbom": False, "checkout_available": True, "source_ref": { "kind": "forgejo-archive-v1", "repository": "coulomb/never-scanned", "revision": "a" * 40, "archive_url": "must-not-enter-progress-evidence", }, }, { "repo_slug": "no-checkout", "last_sbom_at": "2026-05-01T00:00:00Z", "sbom_age_days": 112, "has_sbom": True, "checkout_available": False, "source_ref": { "kind": "forgejo-archive-v1", "repository": "coulomb/no-checkout", "revision": "not-a-full-sha", }, }, ], "selected_count": 2, "stale_count": 111, "never_count": 93, "total_count": 111, "limit": 3, } } result = _deterministic_context_report(_instruction(), context) assert result.tasks == [] report = result.report assert report["deterministic"] is True assert report["never_count"] == 93 assert report["controlled_source_count"] == 1 assert [r["repo_slug"] for r in report["selected_repos"]] == [ "never-scanned", "no-checkout", ] assert "93 never scanned of 111 repos" in report["summary"] assert report["updated_repos"] == [] assert report["skipped_repos"] == [] assert report["selected_repos"][0]["source_ref"] == { "kind": "forgejo-archive-v1", "repository": "coulomb/never-scanned", "revision": "a" * 40, } assert "source_ref" not in report["selected_repos"][1] def test_deterministic_report_carries_t02_ingest_outcomes() -> None: """Forward-compatible with the bounded ingest side-effect (T02).""" from activity_core.rules.executor import _deterministic_context_report context = { "catchup": { "repos": [{"repo_slug": "a"}, {"repo_slug": "b"}], "updated": [{"repo_slug": "a"}], "skipped": [{"repo_slug": "b", "reason": "no-checkout"}], "stale_count": 110, "never_count": 92, "total_count": 111, "limit": 3, } } report = _deterministic_context_report(_instruction(), context).report assert report["updated_repos"] == ["a"] assert report["skipped_repos"] == [{"repo_slug": "b", "reason": "no-checkout"}] assert "1 updated, 1 skipped" in report["summary"] def test_daily_definition_is_bounded_and_enabled() -> None: from pathlib import Path from activity_core.definition_parser import parse_file definition = parse_file(Path("activity-definitions/daily-sbom-catchup.md")) assert definition.enabled is True # No rule block: the weekly flood came from `for_each` over every stale repo. assert definition.rules == [] source = definition.context_sources[0] 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" assert instruction["report_sinks"][0]["event_type"] == "sbom_catchup" def test_railiance_projection_enables_the_same_bounded_contract(tmp_path) -> None: from pathlib import Path import yaml from activity_core.definition_parser import parse_file documents = list( yaml.safe_load_all(Path("k8s/railiance/20-runtime.yaml").read_text()) ) config = next( item for item in documents if isinstance(item, dict) and item.get("kind") == "ConfigMap" and item.get("metadata", {}).get("name") == "actcore-external-activity-definitions" ) projected = tmp_path / "daily-sbom-catchup.md" projected.write_text(config["data"]["daily-sbom-catchup.md"]) definition = parse_file(projected) assert definition.enabled is True assert definition.rules == [] source = definition.context_sources[0] assert source["params"] == {"limit": 3, "apply": True}