feat(sbom): draft bounded daily SBOM catch-up (ACTIVITY-WP-0030-T01)
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s
Build and Publish Container Image / build-and-push (push) Successful in 23s

Add the sbom-nexus catch_up resolver contract and the daily replacement for
weekly-sbom-staleness, both disabled until CUST-WP-0062-T03 lands.

The weekly check used `for_each: context.repos.repos` and emitted one task per
stale repo — 75 tasks on 2026-08-17 against 111/111 stale repos. The
replacement asks sbom-nexus for only the N oldest-stale repos in one ranked
call and carries no rule block at all, so tasks_spawned is 0 by construction.

- context_resolvers/sbom_nexus.py: source type `sbom-nexus`, query `catch_up`,
  GET /sbom/catch-up?limit=N. Read-only; ingest is T02. Limit bounded 1..25 and
  the response truncated to it so an over-long reply cannot widen T02's
  side-effect.
- activity-definitions/daily-sbom-catchup.md: weekdays 09:15 Berlin, enabled:
  false, deterministic sbom_catchup progress sink.
- rules/executor.py: the deterministic report builder only special-cased
  context.repos, which would have emitted a contentless progress event for this
  definition. _sbom_catchup_report names the selected repos and reads
  updated/skipped from context when T02 populates them.
- 17 tests against a test double; no live nexus exists yet.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-08-21 08:45:40 +02:00
parent a446de1c45
commit e64af4102d
6 changed files with 667 additions and 2 deletions

View file

@ -692,6 +692,10 @@ def _execution_failure_report(instr: Any, error: str) -> dict[str, Any] | None:
def _deterministic_context_report(instr: Any, context: dict) -> InstructionResult:
"""Build a report from resolved context without calling an LLM."""
catchup = context.get("catchup") if isinstance(context, dict) else None
if isinstance(catchup, dict) and isinstance(catchup.get("repos"), list):
return _sbom_catchup_report(instr, catchup)
repos = context.get("repos") if isinstance(context, dict) else None
if isinstance(repos, dict):
repo_list = repos.get("repos") if isinstance(repos.get("repos"), list) else []
@ -746,6 +750,59 @@ def _deterministic_context_report(instr: Any, context: dict) -> InstructionResul
)
def _sbom_catchup_report(instr: Any, catchup: dict) -> InstructionResult:
"""Deterministic SBOM catch-up evidence (ACTIVITY-WP-0030).
Names the repos selected by the ranked sbom-nexus call, plus fleet counters
so ``never_count`` can be watched declining day over day. ``updated`` and
``skipped`` are read from the context when the bounded ingest side-effect
(ACTIVITY-WP-0030-T02) has populated them, and stay empty until then.
"""
selected = [r for r in catchup["repos"] if isinstance(r, dict)]
updated = [r for r in catchup.get("updated", []) if isinstance(r, dict)]
skipped = [r for r in catchup.get("skipped", []) if isinstance(r, dict)]
limit = catchup.get("limit")
report: dict[str, Any] = {
"summary": (
f"SBOM catch-up: {len(selected)} selected (limit {limit}), "
f"{len(updated)} updated, {len(skipped)} skipped; "
f"{catchup.get('never_count')} never scanned of "
f"{catchup.get('total_count')} repos."
),
"status": "deterministic",
"deterministic": True,
"limit": limit,
"selected_count": len(selected),
"stale_count": catchup.get("stale_count"),
"never_count": catchup.get("never_count"),
"total_count": catchup.get("total_count"),
"selected_repos": [
{
"repo_slug": r.get("repo_slug"),
"sbom_age_days": r.get("sbom_age_days"),
"has_sbom": r.get("has_sbom"),
"last_sbom_at": r.get("last_sbom_at"),
"checkout_available": r.get("checkout_available"),
}
for r in selected
],
"updated_repos": [r.get("repo_slug") for r in updated],
"skipped_repos": [
{"repo_slug": r.get("repo_slug"), "reason": r.get("reason")}
for r in skipped
],
}
return InstructionResult(
tasks=[],
report=report,
prompt_hash=None,
model=getattr(instr, "model", None),
output_validated=True,
review_required=bool(getattr(instr, "review_required", False)),
condition_matched=getattr(instr, "condition", "") or None,
)
def _validate_output(
raw_output: Any,
instr: Any,