Harden SBOM retries and align hub evidence

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f

src/activity_core/activities.py

@@ -199,6 +199,81 @@ async def resolve_context(
     return snapshot
 
 
+def _sbom_heartbeat_state(run_id: str) -> dict[str, Any]:
+    try:
+        details = activity.info().heartbeat_details
+    except RuntimeError:
+        return {"run_id": run_id, "outcomes_by_bind": {}}
+    if not details or not isinstance(details[0], dict):
+        return {"run_id": run_id, "outcomes_by_bind": {}}
+    state = dict(details[0])
+    if state.get("run_id") != run_id:
+        return {"run_id": run_id, "outcomes_by_bind": {}}
+    if not isinstance(state.get("outcomes_by_bind"), dict):
+        state["outcomes_by_bind"] = {}
+    return state
+
+
+def _heartbeat_sbom_state(state: dict[str, Any]) -> None:
+    try:
+        activity.heartbeat(state)
+    except RuntimeError:
+        # Direct unit invocation has no Temporal activity context.
+        pass
+
+
+@activity.defn
+async def apply_sbom_catchup(payload: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Apply declared SBOM writes to the fixed selection in workflow history."""
+    from activity_core.context_resolvers.sbom_nexus import apply_bounded_ingest
+
+    run_id = str(payload["run_id"])
+    context_sources = payload.get("context_sources") or []
+    context = payload.get("context") or {}
+    heartbeat_state = _sbom_heartbeat_state(run_id)
+    outcomes_by_bind = heartbeat_state["outcomes_by_bind"]
+    patches: dict[str, dict[str, Any]] = {}
+
+    for source in context_sources:
+        if not isinstance(source, dict):
+            continue
+        params = source.get("params") or {}
+        if not (
+            source.get("type") == "sbom-nexus"
+            and source.get("query") == "catch_up"
+            and params.get("apply") is True
+        ):
+            continue
+
+        raw_bind = source.get("bind_to") or source.get("name") or "sbom-nexus"
+        bind_key = str(raw_bind).removeprefix("context.")
+        selection = context.get(bind_key)
+        if not isinstance(selection, dict):
+            continue
+        repos = selection.get("repos")
+        if not isinstance(repos, list):
+            continue
+        try:
+            limit = int(selection.get("limit", params.get("limit", 3)))
+        except (TypeError, ValueError):
+            limit = 3
+        limit = max(1, min(25, limit))
+        fixed_repos = [repo for repo in repos[:limit] if isinstance(repo, dict)]
+
+        def record_progress(outcomes: list[dict[str, Any]]) -> None:
+            outcomes_by_bind[bind_key] = outcomes
+            _heartbeat_sbom_state(heartbeat_state)
+
+        patches[bind_key] = apply_bounded_ingest(
+            fixed_repos,
+            operation_id=run_id,
+            completed=outcomes_by_bind.get(bind_key),
+            on_progress=record_progress,
+        )
+
+    return patches
+
+
 @activity.defn
 async def log_run(run_payload: dict) -> str:
     """Persist an ActivityRun record to Postgres and return its run_id.

src/activity_core/context_resolvers/sbom_nexus.py

@@ -32,14 +32,10 @@ Ordering is the nexus's responsibility (never-scanned first, then oldest
 ``last_sbom_at``); this adapter validates the shape and normalises the entries
 so the deterministic report can render them without comprehensions.
 
-Until CUST-WP-0062-T03 lands there is no live endpoint — the query is exercised
-against a test double (``tests/test_sbom_nexus_context_resolver.py``) and the
-daily definition stays ``enabled: false``.
-
-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.
+The ranked query is always read-only, including when a definition declares
+``params.apply: true``. The workflow records that result in Temporal history,
+then a dedicated activity applies the already-truncated fixed target set. This
+prevents a retry from querying and advancing into a second batch.
 
 Config: SBOM_NEXUS_URL env var (default: http://127.0.0.1:8010).
 """
@@ -47,8 +43,10 @@ Config: SBOM_NEXUS_URL env var (default: http://127.0.0.1:8010).
 from __future__ import annotations
 
 import os
+from collections.abc import Callable
 from typing import Any
 from urllib.parse import quote
+from uuid import NAMESPACE_URL, uuid5
 
 import httpx
 
@@ -86,10 +84,22 @@ 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:
+def _post_json(
+    path: str,
+    payload: dict[str, Any] | None = None,
+    *,
+    idempotency_key: str,
+) -> Any:
     url = f"{_base_url()}{path}"
     with httpx.Client(timeout=_TIMEOUT_SECONDS) as client:
-        response = client.post(url, json=payload)
+        response = client.post(
+            url,
+            json=payload,
+            headers={
+                "Idempotency-Key": idempotency_key,
+                "X-Activity-Core-Operation-ID": idempotency_key,
+            },
+        )
         response.raise_for_status()
         return response.json()
 
@@ -145,10 +155,12 @@ def _catch_up(params: dict[str, Any]) -> dict[str, Any]:
         raise RuntimeError("sbom-nexus catch_up response missing required key: repos")
 
     repos: list[dict[str, Any]] = []
+    seen_slugs: set[str] = set()
     for raw in raw_repos:
         entry = _normalise_entry(raw)
-        if entry is not None:
+        if entry is not None and entry["repo_slug"] not in seen_slugs:
             repos.append(entry)
+            seen_slugs.add(entry["repo_slug"])
     # The nexus owns ranking, but the definition promises "at most N": never let
     # an over-long response widen the bounded side-effect in T02.
     repos = repos[:limit]
@@ -157,7 +169,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))
 
-    result = {
+    return {
         "repos": repos,
         "selected_count": len(repos),
         "stale_count": stale_count,
@@ -165,65 +177,135 @@ 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]:
+def _operation_key(operation_id: str, repo_slug: str) -> str:
+    return str(
+        uuid5(
+            NAMESPACE_URL,
+            f"activity-core:sbom-catchup:{operation_id}:{repo_slug}",
+        )
+    )
+
+
+def _skip(
+    repo_slug: str,
+    reason: str,
+    *,
+    operation_id: 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":
+    raw = _post_json(
+        f"/sbom/{quote(repo_slug, safe='')}/skip",
+        payload,
+        idempotency_key=_operation_key(operation_id, repo_slug),
+    )
+    if (
+        not isinstance(raw, dict)
+        or raw.get("status") != "skipped"
+        or raw.get("reason") not in {"no-checkout", "no-manifest", "ingest-error"}
+    ):
         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")
+def _ingest(repo_slug: str, *, operation_id: str) -> dict[str, Any]:
+    raw = _post_json(
+        f"/sbom/{quote(repo_slug, safe='')}/ingest",
+        idempotency_key=_operation_key(operation_id, repo_slug),
+    )
+    valid = isinstance(raw, dict) and raw.get("status") in {"ingested", "skipped"}
+    if not valid:
+        raise RuntimeError(f"sbom-nexus ingest returned an invalid outcome for {repo_slug}")
+    if raw.get("status") == "skipped" and raw.get("reason") not in {
+        "no-checkout",
+        "no-manifest",
+        "ingest-error",
+    }:
+        raise RuntimeError(f"sbom-nexus ingest returned an invalid skip for {repo_slug}")
     return raw
 
 
-def _apply_bounded_ingest(repos: list[dict[str, Any]]) -> dict[str, Any]:
-    updated: list[dict[str, Any]] = []
-    skipped: list[dict[str, Any]] = []
+def _compact_outcome(repo_slug: str, outcome: dict[str, Any]) -> dict[str, Any]:
+    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)
+    return compact
 
+
+def apply_bounded_ingest(
+    repos: list[dict[str, Any]],
+    *,
+    operation_id: str,
+    completed: list[dict[str, Any]] | None = None,
+    on_progress: Callable[[list[dict[str, Any]]], None] | None = None,
+) -> dict[str, Any]:
+    """Apply one fixed target set, resuming outcomes acknowledged by heartbeat.
+
+    Transport errors and malformed responses are deliberately not converted to
+    synthetic skips. The remote write may have committed, so only Nexus can
+    safely resolve that ambiguity through operation-id enforcement.
+    """
+    selected: list[dict[str, Any]] = []
+    selected_slugs: set[str] = set()
     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__)
+        if repo_slug not in selected_slugs:
+            selected.append(repo)
+            selected_slugs.add(repo_slug)
 
-        compact = {
-            key: outcome.get(key)
-            for key in (
-                "repo_slug",
-                "status",
-                "reason",
-                "snapshot_id",
-                "entry_count",
-                "snapshot_at",
-                "source_revision",
+    outcomes_by_slug = {
+        str(outcome.get("repo_slug")): dict(outcome)
+        for outcome in completed or []
+        if isinstance(outcome, dict)
+        and outcome.get("repo_slug") in selected_slugs
+        and outcome.get("status") in {"ingested", "skipped"}
+    }
+
+    for repo in selected:
+        repo_slug = str(repo["repo_slug"])
+        if repo_slug in outcomes_by_slug:
+            continue
+        if on_progress:
+            on_progress(list(outcomes_by_slug.values()))
+        if repo.get("checkout_available") is False:
+            outcome = _skip(
+                repo_slug,
+                "no-checkout",
+                operation_id=operation_id,
             )
-            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)
+            outcome = _ingest(repo_slug, operation_id=operation_id)
+        outcomes_by_slug[repo_slug] = _compact_outcome(repo_slug, outcome)
+        if on_progress:
+            on_progress(list(outcomes_by_slug.values()))
+
+    ordered_outcomes = [
+        outcomes_by_slug[str(repo["repo_slug"])] for repo in selected
+    ]
+    updated = [
+        outcome for outcome in ordered_outcomes if outcome.get("status") == "ingested"
+    ]
+    skipped = [
+        outcome for outcome in ordered_outcomes if outcome.get("status") == "skipped"
+    ]
 
     return {
-        "attempted_count": len(repos),
+        "attempted_count": len(ordered_outcomes),
         "updated": updated,
         "skipped": skipped,
     }

src/activity_core/ops_evidence_sinks.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 import json
 import os
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
+from uuid import NAMESPACE_URL, UUID, uuid5
 
 import httpx
 
@@ -22,10 +24,14 @@ _INTER_HUB_SINK_TYPES = {
     "inter-hub-event",
     "inter-hub-interaction-event",
 }
-_CORE_HUB_SINK_TYPES = {
+_LEGACY_CORE_HUB_SINK_TYPES = {
     "core-hub",
     "core-hub-interaction-event",
 }
+_HUB_CORE_SINK_TYPES = {
+    "hub-core",
+    "hub-core-interaction-event",
+}
 
 
 def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, Any]]:
@@ -77,7 +83,13 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An
                     results.append(
                         _post_state_hub_progress(payload, bind_key, probe_result, sink)
                     )
-                elif sink_type in _CORE_HUB_SINK_TYPES:
+                elif sink_type in _HUB_CORE_SINK_TYPES:
+                    results.append(
+                        _post_hub_core_interaction_event(
+                            payload, bind_key, probe_result, sink
+                        )
+                    )
+                elif sink_type in _LEGACY_CORE_HUB_SINK_TYPES:
                     results.append(
                         _post_core_hub_interaction_event(
                             payload, bind_key, probe_result, sink
@@ -215,6 +227,131 @@ def _progress_exists(base_url: str, event_type: str, idempotency_key: str) -> bo
     return False
 
 
+def _post_hub_core_interaction_event(
+    payload: dict[str, Any],
+    context_key: str,
+    probe_result: dict[str, Any],
+    sink: dict[str, Any],
+) -> dict[str, Any]:
+    """Append sanitized evidence through hub-core's canonical interaction port."""
+    raw_base_url = (
+        sink.get("hub_core_url")
+        or sink.get("base_url")
+        or os.environ.get("HUB_CORE_BASE_URL")
+        or ""
+    )
+    base_url = str(raw_base_url).rstrip("/")
+    if not base_url:
+        return {
+            "type": sink.get("type"),
+            "status": "skipped",
+            "reason": "missing_hub_core_config",
+            "missing": ["HUB_CORE_BASE_URL"],
+            "context_key": context_key,
+        }
+
+    endpoint = _selected_endpoint(probe_result, sink)
+    correlation_id = _hub_core_correlation_id(payload, context_key)
+    reported_event_type = str(sink.get("event_type", "ops-endpoint-verified"))
+    body = {
+        "schema_version": str(sink.get("schema_version", "0.1.0")),
+        "correlation_id": correlation_id,
+        # The port catalog separates the interaction envelope from the
+        # domain-specific observation carried in its payload.
+        "event_type": "hub.interaction.recorded",
+        "occurred_at": _hub_core_occurred_at(payload, probe_result),
+        "subject_refs": _hub_core_subject_refs(payload, context_key, endpoint),
+        "payload": {
+            "reported_event_type": reported_event_type,
+            "view_context": _core_hub_view_context(
+                payload, context_key, endpoint, sink
+            ),
+            **_core_hub_metadata(
+                payload, context_key, probe_result, endpoint
+            ),
+        },
+    }
+    timeout = float(sink.get("timeout_seconds", 10.0))
+    headers = {
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+        "User-Agent": "activity-core-ops-evidence/0.2",
+    }
+    resp = httpx.post(
+        f"{base_url}/ports/events/interaction",
+        json=body,
+        headers=headers,
+        timeout=timeout,
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    event_id = data.get("id")
+    if not event_id:
+        raise RuntimeError("Hub Core interaction port response did not include an id")
+    if not _hub_core_event_exists(base_url, str(event_id), timeout, headers):
+        raise RuntimeError("Hub Core interaction event was not visible after append")
+
+    return {
+        "type": sink.get("type"),
+        "status": "posted",
+        "event_type": "hub.interaction.recorded",
+        "reported_event_type": reported_event_type,
+        "event_id": event_id,
+        "correlation_id": data.get("correlation_id", correlation_id),
+        "verified": True,
+        "context_key": context_key,
+    }
+
+
+def _hub_core_correlation_id(payload: dict[str, Any], context_key: str) -> str:
+    raw = str(payload.get("run_id") or "")
+    try:
+        return str(UUID(raw))
+    except ValueError:
+        return str(uuid5(NAMESPACE_URL, f"activity-core:{raw}:{context_key}"))
+
+
+def _hub_core_occurred_at(
+    payload: dict[str, Any], probe_result: dict[str, Any]
+) -> str:
+    value = payload.get("scheduled_for") or probe_result.get("generated_at")
+    if isinstance(value, str) and value:
+        return value
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _hub_core_subject_refs(
+    payload: dict[str, Any], context_key: str, endpoint: dict[str, Any]
+) -> dict[str, str]:
+    refs = {
+        "activity": payload.get("activity_id"),
+        "activity_run": payload.get("run_id"),
+        "context": context_key,
+        "endpoint": endpoint.get("endpoint_id"),
+    }
+    return {key: str(value) for key, value in refs.items() if value}
+
+
+def _hub_core_event_exists(
+    base_url: str,
+    event_id: str,
+    timeout: float,
+    headers: dict[str, str],
+) -> bool:
+    resp = httpx.get(
+        f"{base_url}/ports/projections/interaction_events",
+        headers=headers,
+        timeout=timeout,
+    )
+    resp.raise_for_status()
+    body = resp.json()
+    data = body.get("data") if isinstance(body, dict) else None
+    items = data.get("items") if isinstance(data, dict) else None
+    if not isinstance(items, list):
+        return False
+    return any(isinstance(item, dict) and item.get("id") == event_id for item in items)
+
+
 def _post_core_hub_interaction_event(
     payload: dict[str, Any],
     context_key: str,

src/activity_core/worker.py

@@ -33,6 +33,7 @@ from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
 from temporalio.worker import Worker
 
 from activity_core.activities import (
+    apply_sbom_catchup,
     emit_tasks,
     evaluate_instructions,
     evaluate_rules,
@@ -104,6 +105,7 @@ async def run() -> None:
         activities=[
             load_activity_definition,
             resolve_context,
+            apply_sbom_catchup,
             log_run,
             evaluate_rules,
             evaluate_instructions,

src/activity_core/workflows.py

@@ -20,6 +20,7 @@ from temporalio.common import RetryPolicy, SearchAttributeKey, TypedSearchAttrib
 
 with workflow.unsafe.imports_passed_through():
     from activity_core.activities import (
+        apply_sbom_catchup,
         emit_tasks,
         evaluate_rules,
         evaluate_instructions,
@@ -57,10 +58,10 @@ class RunActivityWorkflow:
 
     Sequence:
       1. load_activity_definition(activity_id)  → defn dict
-      2. resolve_context(defn.context_sources)  → context snapshot
-      3. evaluate_rules(rules, event, context)  → matching rules → TaskSpec dicts
-      4. emit_tasks(task_specs)                 → TaskRef list via IssueSink
-      5. log_run(...)                           → activity_runs row
+      2. resolve_context(defn.context_sources)  → read-only context snapshot
+      3. apply_sbom_catchup(fixed selection)    → bounded outcome patch
+      4. evaluate rules/instructions            → TaskSpec dicts and reports
+      5. log run, then emit tasks
     """
 
     @workflow.run
@@ -99,7 +100,13 @@ class RunActivityWorkflow:
             ])
         )
 
-        # ── 2. Resolve context ────────────────────────────────────────────────
+        if trigger_key == SCHEDULED_TRIGGER_KEY:
+            dedup_source = workflow.info().workflow_id
+        else:
+            dedup_source = f"{activity_id}:{trigger_key}"
+        run_id = str(uuid.uuid5(uuid.NAMESPACE_URL, dedup_source))
+
+        # ── 2. Resolve context (read-only) ────────────────────────────────────
         context_snapshot: dict = await workflow.execute_activity(
             resolve_context,
             args=[defn["context_sources"], event_envelope_json],
@@ -107,11 +114,29 @@ class RunActivityWorkflow:
             retry_policy=_RETRY_POLICY,
         )
 
-        if trigger_key == SCHEDULED_TRIGGER_KEY:
-            dedup_source = workflow.info().workflow_id
-        else:
-            dedup_source = f"{activity_id}:{trigger_key}"
-        run_id = str(uuid.uuid5(uuid.NAMESPACE_URL, dedup_source))
+        # ── 3. Apply declared bounded side-effects to the fixed selection ────
+        if any(
+            isinstance(source, dict)
+            and source.get("type") == "sbom-nexus"
+            and source.get("query") == "catch_up"
+            and (source.get("params") or {}).get("apply") is True
+            for source in defn.get("context_sources", [])
+        ) and workflow.patched("activity-wp-0033-sbom-retry-boundary"):
+            context_patches: dict = await workflow.execute_activity(
+                apply_sbom_catchup,
+                {
+                    "context_sources": defn.get("context_sources", []),
+                    "context": context_snapshot,
+                    "run_id": run_id,
+                },
+                start_to_close_timeout=_ACTIVITY_TIMEOUT,
+                heartbeat_timeout=timedelta(seconds=60),
+                retry_policy=_RETRY_POLICY,
+            )
+            for bind_key, patch in context_patches.items():
+                current = context_snapshot.get(bind_key)
+                if isinstance(current, dict) and isinstance(patch, dict):
+                    current.update(patch)
 
         await workflow.execute_activity(
             persist_ops_evidence,
@@ -127,7 +152,7 @@ class RunActivityWorkflow:
             retry_policy=_RETRY_POLICY,
         )
 
-        # ── 3. Evaluate rules ─────────────────────────────────────────────────
+        # ── 4. Evaluate rules ─────────────────────────────────────────────────
         import json as _json
         event_attrs: dict = {}
         if event_envelope_json:
@@ -162,7 +187,7 @@ class RunActivityWorkflow:
             task_spec_dicts.extend(instruction_result.get("task_specs", []))
             report_dicts.extend(instruction_result.get("reports", []))
 
-        # ── 4. Persist reports ────────────────────────────────────────────────
+        # ── 5. Persist reports ────────────────────────────────────────────────
         if report_dicts:
             await workflow.execute_activity(
                 persist_instruction_reports,
@@ -177,7 +202,7 @@ class RunActivityWorkflow:
                 retry_policy=_RETRY_POLICY,
             )
 
-        # ── 5. Log the run BEFORE emit ────────────────────────────────────────
+        # ── 6. Log the run BEFORE emit ────────────────────────────────────────
         # ACTIVITY-WP-0021: emit_tasks sink failures used to abort the workflow
         # before log_run, so failed Binky/SBOM fires left no activity_runs row
         # and automation-status could not observe them. Always record the run;
@@ -196,7 +221,7 @@ class RunActivityWorkflow:
             retry_policy=_RETRY_POLICY,
         )
 
-        # ── 6. Emit tasks (may fail independently of run audit) ───────────────
+        # ── 7. Emit tasks (may fail independently of run audit) ───────────────
         if task_spec_dicts:
             # Cron schedules pass trigger_key="scheduled" for *every* fire.
             # ops_run idempotency is {def}:{source}:{triggering_event_id}, so