diff --git a/.env.example b/.env.example index b8f7071..e6914b6 100644 --- a/.env.example +++ b/.env.example @@ -15,6 +15,9 @@ NATS_URL=nats://nats:4222 # ── Service integrations (gracefully degraded if unavailable) ───────────────── # State Hub — used by the state-hub context adapter. Binds {} on failure. STATE_HUB_URL=http://127.0.0.1:8000 +# Hub Core canonical extension ports (ACTIVITY-WP-0029). No Core Hub runtime +# token or widget mapping is needed for port.events.interaction. +HUB_CORE_BASE_URL=http://127.0.0.1:8010 # Consistency sweep engine is repo-manager (ACTIVITY-WP-0029). Default POST # still hits State Hub dual-run. Set REPO_MANAGER_URL or CONSISTENCY_SWEEP_URL # when repo-manager exposes the HTTP sweep. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index f914563..91549a1 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -32,10 +32,11 @@ | workplan | ACTIVITY-WP-0026 | finished | — | workplans/ACTIVITY-WP-0026-ops-run-claim-queue.md | | workplan | ACTIVITY-WP-0027 | finished | — | workplans/ACTIVITY-WP-0027-llm-connect-and-run-artifacts.md | | workplan | ACTIVITY-WP-0028 | finished | — | workplans/ACTIVITY-WP-0028-activity-review-cli.md | -| workplan | ACTIVITY-WP-0029 | active | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | -| workplan | ACTIVITY-WP-0030 | done | — | workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md | +| workplan | ACTIVITY-WP-0029 | finished | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | +| workplan | ACTIVITY-WP-0030 | finished | — | workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md | | workplan | ACTIVITY-WP-0031 | active | — | workplans/ACTIVITY-WP-0031-production-execution-reliability-cleanup.md | | workplan | ACTIVITY-WP-0032 | active | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | +| workplan | ACTIVITY-WP-0033 | finished | — | workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md | | workplan | ADHOC-2026-06-01 | finished | — | workplans/ADHOC-2026-06-01.md | | workplan | ADHOC-2026-08-20 | finished | — | workplans/ADHOC-2026-08-20.md | | workplan | custodian-WP-0001 | done | — | workplans/custodian-WP-0001-temporal-backbone.md | @@ -183,7 +184,7 @@ | task | ACTIVITY-WP-0028-T07 | done | — | workplans/ACTIVITY-WP-0028-activity-review-cli.md | | task | ACTIVITY-WP-0029-T01 | done | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | | task | ACTIVITY-WP-0029-T02 | done | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | -| task | ACTIVITY-WP-0029-T03 | wait | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | +| task | ACTIVITY-WP-0029-T03 | done | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | | task | ACTIVITY-WP-0029-T04 | done | — | workplans/ACTIVITY-WP-0029-hub-port-alignment.md | | task | ACTIVITY-WP-0030-T01 | done | — | workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md | | task | ACTIVITY-WP-0030-T02 | done | — | workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md | @@ -199,6 +200,9 @@ | task | ACTIVITY-WP-0032-T03 | todo | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | | task | ACTIVITY-WP-0032-T04 | wait | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | | task | ACTIVITY-WP-0032-T05 | wait | — | workplans/ACTIVITY-WP-0032-glas-profile-execution-contract.md | +| task | ACTIVITY-WP-0033-T01 | done | — | workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md | +| task | ACTIVITY-WP-0033-T02 | done | — | workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md | +| task | ACTIVITY-WP-0033-T03 | done | — | workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md | | task | ADHOC-2026-06-01-T01 | done | — | workplans/ADHOC-2026-06-01.md | | task | ADHOC-2026-06-01-T02 | done | — | workplans/ADHOC-2026-06-01.md | | task | ADHOC-2026-06-01-T03 | done | — | workplans/ADHOC-2026-06-01.md | diff --git a/activity-definitions/daily-sbom-catchup.md b/activity-definitions/daily-sbom-catchup.md index c038210..437b7c8 100644 --- a/activity-definitions/daily-sbom-catchup.md +++ b/activity-definitions/daily-sbom-catchup.md @@ -79,15 +79,24 @@ 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. -## Bounded side-effect (ACTIVITY-WP-0030-T02) +## Bounded side-effect (ACTIVITY-WP-0030-T02 / ACTIVITY-WP-0033) -`params.apply: true` calls sbom-nexus for each of the already-truncated N -targets. A target with an explicitly unavailable checkout is recorded through +`params.apply: true` declares the write, but context resolution remains +read-only. The workflow records the already-truncated N targets in Temporal +history, then a dedicated activity applies exactly that fixed set. Completed +per-repository outcomes are heartbeated so an ordinary activity retry resumes +after them instead of selecting or writing a second batch. + +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. +route, which returns `ingested`, `no-manifest`, or `ingest-error`. Each request +sends a stable per-run/per-repository `Idempotency-Key`. Transport failures, +timeouts, and malformed responses fail visibly and are never converted into a +synthetic skip because the remote write may have committed. Nexus must enforce +the operation key to close the remaining crash window between a committed POST +and Activity Core's next heartbeat. The resulting `updated` and `skipped` +arrays are part of the run context and progress report. There is no task or +issue emission. ## Enable checklist diff --git a/docs/runbook.md b/docs/runbook.md index d24a876..17519d3 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -595,9 +595,15 @@ export FORGEJO_TOKEN=$(bao kv get -field=API_TOKEN platform/workloads/forgejo/fo fleet-wide `for_each: context.repos.repos` emitted 75 tasks on 2026-08-17 and must not be re-enabled. The bounded replacement is `daily-sbom-catchup`, active weekdays at 09:15 Europe/Berlin. It makes one ranked sbom-nexus request, -processes at most three terminal ingest/skip outcomes, and emits no tasks. Two -production proof fires on 2026-08-22 processed six distinct repositories as -documented `no-checkout` skips with zero task spawns; see +records that read-only selection in workflow history, and processes at most +three terminal ingest/skip outcomes in a dedicated heartbeat-enabled activity. +Retries retain the same fixed targets and resume acknowledged outcomes. Each +write carries a stable per-run/per-repository `Idempotency-Key`; an ambiguous +response fails the activity and is not converted into a synthetic skip. Until +sbom-nexus enforces that key, a worker crash after a committed POST but before +the next heartbeat remains a possible duplicate-write window. The automation +emits no tasks. Two production proof fires on 2026-08-22 processed six distinct +repositories as documented `no-checkout` skips with zero task spawns; see `docs/evidence/ACTIVITY-WP-0030-daily-sbom-catchup-2026-08-22.md`. `weekly-coding-retro` follows the same cron -> context resolver -> per-repo task @@ -617,8 +623,18 @@ when the workflow completes cleanly twice in a row. ## Ops inventory evidence posture -The current accepted live backend for activity-core ops inventory probes is -State Hub progress with `event_type=ops_inventory_probe`. +Hub-core has been the production runtime authority since 2026-08-21. The +Railiance projection of the (still disabled) ops-inventory probe now writes +sanitized evidence through canonical `port.events.interaction` using sink type +`hub-core-interaction-event`. `HUB_CORE_BASE_URL` targets the in-cluster +hub-core runtime. No widget mapping or Core Hub runtime token is required. + +The sink wraps the observation as catalog event `hub.interaction.recorded`, +keeps the logical observation name in `payload.reported_event_type`, and reads +the `interaction_events` projection after append to verify persistence. The old +`core-hub-interaction-event` `/api/v2` sink remains implemented for rollback +compatibility during CORE-WP-0010 stabilization, but is no longer the Railiance +probe configuration. Inter-Hub / ops-hub per-entity submission remains intentionally deferred until all of these are true: diff --git a/docs/state-hub-caller-map.md b/docs/state-hub-caller-map.md index a627725..de4a285 100644 --- a/docs/state-hub-caller-map.md +++ b/docs/state-hub-caller-map.md @@ -13,7 +13,9 @@ activity-core stays the scheduler even when it is not the engine. Hub-core port names are from `helixforge.hub-extension` 0.1 (`prj-state-hub-retirement/architecture/hub-extension-contract_v0.1.yaml`). -HUB-WP-0004 has not shipped those ports yet. +HUB-WP-0004 shipped the named ports on 2026-08-21; hub-core became production +authority the same day. Core Hub remains a no-writer rollback only through the +stabilization window ending no earlier than 2026-08-28. | Caller | Current URL | SHR-INV | Target owner | Port (when named) | | --- | --- | --- | --- | --- | @@ -51,11 +53,25 @@ HUB-WP-0004 has not shipped those ports yet. (dual-run when `RM_RECONCILE` is on). Override with `CONSISTENCY_SWEEP_URL` or `REPO_MANAGER_URL` + optional `CONSISTENCY_SWEEP_PATH`. -## Evidence sink (T03 — wait) +## Evidence sink (T03 — aligned 2026-08-22) -All `POST /progress/` callers stay on the State Hub beachhead until -**HUB-WP-0004** publishes `port.events.progress` / `port.events.interaction`. -Do not add a second permanent progress host. +Ops-inventory evidence now has a canonical `hub-core-interaction-event` sink: + +- `POST /ports/events/interaction` with catalog event + `hub.interaction.recorded`; +- the sanitized domain observation remains in + `payload.reported_event_type` (for example `ops_inventory_probe`); +- append verification reads + `GET /ports/projections/interaction_events`; and +- `HUB_CORE_BASE_URL` points at the surviving hub-core runtime. The Railiance + Service retains the compatibility name `core-hub-api` during stabilization, + but the process and authority behind it are hub-core. + +The legacy `core-hub-interaction-event` `/api/v2` sink remains available only +for rollback compatibility. It is not the configured Railiance probe sink. +Unmigrated general progress callers can continue through the State Hub +beachhead during the retirement strangler; this task does not create a second +permanent progress authority. ## Execution (T04) diff --git a/k8s/railiance/20-runtime.yaml b/k8s/railiance/20-runtime.yaml index 8f2f08a..9fd2aab 100644 --- a/k8s/railiance/20-runtime.yaml +++ b/k8s/railiance/20-runtime.yaml @@ -29,6 +29,9 @@ data: CUSTODIAN_REPO_ROOT: /var/custodian ACTIVITY_CORE_ROOT: /etc/activity-core OPS_INVENTORY_PATH: /etc/activity-core/ops/service-inventory.yml + # Canonical hub-core runtime port; the Service name remains core-hub-api + # during the CORE-WP-0010 stabilization/rollback window. + HUB_CORE_BASE_URL: http://core-hub-api.core-hub.svc.cluster.local:8010 INTER_HUB_URL: "" OPS_HUB_WIDGET_MAPPING: "" PROMETHEUS_BIND_ADDR: 0.0.0.0:9090 @@ -68,8 +71,10 @@ data: # Daily SBOM Catch-up - Ranked, bounded SBOM catch-up. Each fire processes at most three - repositories through sbom-nexus and deliberately contains no task rule. + Ranked, bounded SBOM catch-up. Each fire records one read-only selection, + then processes that fixed set of at most three repositories in a dedicated + heartbeat-enabled activity. Ambiguous writes fail visibly; stable operation + identity is sent to sbom-nexus. This definition contains no task rule. ```instruction id: daily-sbom-catchup-report @@ -646,9 +651,8 @@ data: - https allow_network: true evidence_sinks: - - type: state-hub-progress + - type: hub-core-interaction-event event_type: ops_inventory_probe - author: activity-core bind_to: context.ops_inventory_probe --- @@ -656,7 +660,9 @@ data: Disabled Railiance projection of the Custodian-owned definition in `/home/worsch/the-custodian/activity-definitions/ops-service-inventory-probes.md`. - Keep disabled until ops-hub Inter-Hub evidence intake is active. + Keep disabled until the operator selects the desired probe cadence. Evidence + uses hub-core `port.events.interaction`; no widget mapping or runtime secret + is required. fi-daily-research-brief.md: | --- id: fi-daily-research-brief diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 69d50d6..4d35fde 100644 --- a/src/activity_core/activities.py +++ b/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. diff --git a/src/activity_core/context_resolvers/sbom_nexus.py b/src/activity_core/context_resolvers/sbom_nexus.py index 88de281..4d44dcc 100644 --- a/src/activity_core/context_resolvers/sbom_nexus.py +++ b/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, } diff --git a/src/activity_core/ops_evidence_sinks.py b/src/activity_core/ops_evidence_sinks.py index b0bcd8b..567521a 100644 --- a/src/activity_core/ops_evidence_sinks.py +++ b/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, diff --git a/src/activity_core/worker.py b/src/activity_core/worker.py index 100cfa3..e98a5cf 100644 --- a/src/activity_core/worker.py +++ b/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, diff --git a/src/activity_core/workflows.py b/src/activity_core/workflows.py index 1c2b90c..f112871 100644 --- a/src/activity_core/workflows.py +++ b/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 diff --git a/tests/test_ops_evidence_sinks.py b/tests/test_ops_evidence_sinks.py index afd9ad9..96e5a95 100644 --- a/tests/test_ops_evidence_sinks.py +++ b/tests/test_ops_evidence_sinks.py @@ -367,6 +367,90 @@ def test_core_hub_interaction_event_sink_posts_and_verifies_compact_event(monkey assert "token=secret" not in serialized +def test_hub_core_interaction_port_posts_and_verifies_sanitized_event(monkeypatch) -> None: + posts: list[dict[str, Any]] = [] + + def fake_post(url: str, **kwargs: Any) -> DummyResponse: + assert url == "http://hub-core.test/ports/events/interaction" + assert "Authorization" not in kwargs["headers"] + posts.append({"url": url, **kwargs}) + return DummyResponse( + { + "id": "event-port-1", + "status": "accepted", + "correlation_id": _run_id(), + }, + status_code=202, + ) + + def fake_get(url: str, **kwargs: Any) -> DummyResponse: + assert url == "http://hub-core.test/ports/projections/interaction_events" + return DummyResponse( + { + "id": "interaction_events", + "data": {"items": [{"id": "event-port-1"}]}, + "provenance": {}, + } + ) + + monkeypatch.setattr(httpx, "post", fake_post) + monkeypatch.setattr(httpx, "get", fake_get) + + result = persist_ops_inventory_evidence( + _payload([ + { + "type": "hub-core-interaction-event", + "hub_core_url": "http://hub-core.test", + "event_type": "ops-endpoint-verified", + } + ]) + ) + + assert result == [ + { + "type": "hub-core-interaction-event", + "status": "posted", + "event_type": "hub.interaction.recorded", + "reported_event_type": "ops-endpoint-verified", + "event_id": "event-port-1", + "correlation_id": _run_id(), + "verified": True, + "context_key": "ops_probe", + } + ] + body = posts[0]["json"] + assert body["schema_version"] == "0.1.0" + assert body["correlation_id"] == _run_id() + assert body["event_type"] == "hub.interaction.recorded" + assert body["payload"]["reported_event_type"] == "ops-endpoint-verified" + assert body["payload"]["endpoint"]["url"] == "http://state-hub.test/health" + assert body["subject_refs"]["endpoint"] == "state-hub-health" + + serialized = json.dumps(body, sort_keys=True) + assert "secret response body" not in serialized + assert "Authorization" not in serialized + assert "user:pass" not in serialized + assert "token=secret" not in serialized + + +def test_hub_core_interaction_port_skips_when_base_url_missing(monkeypatch) -> None: + monkeypatch.delenv("HUB_CORE_BASE_URL", raising=False) + + result = persist_ops_inventory_evidence( + _payload([{"type": "hub-core-interaction-event"}]) + ) + + assert result == [ + { + "type": "hub-core-interaction-event", + "status": "skipped", + "reason": "missing_hub_core_config", + "missing": ["HUB_CORE_BASE_URL"], + "context_key": "ops_probe", + } + ] + + def test_core_hub_sink_skips_cleanly_when_config_missing(monkeypatch) -> None: monkeypatch.delenv("CORE_HUB_BASE_URL", raising=False) monkeypatch.delenv("CORE_HUB_RUNTIME_TOKEN", raising=False) diff --git a/tests/test_railiance_ops_inventory_wiring.py b/tests/test_railiance_ops_inventory_wiring.py index 3bd1301..4e87df6 100644 --- a/tests/test_railiance_ops_inventory_wiring.py +++ b/tests/test_railiance_ops_inventory_wiring.py @@ -41,6 +41,9 @@ def test_runtime_config_has_ops_inventory_placeholders() -> None: assert config["data"]["OPS_INVENTORY_PATH"] == ( "/etc/activity-core/ops/service-inventory.yml" ) + assert config["data"]["HUB_CORE_BASE_URL"] == ( + "http://core-hub-api.core-hub.svc.cluster.local:8010" + ) assert config["data"]["INTER_HUB_URL"] == "" assert config["data"]["OPS_HUB_WIDGET_MAPPING"] == "" @@ -68,9 +71,8 @@ def test_external_configmap_projects_disabled_ops_probe_definition(tmp_path) -> "allow_network": True, "evidence_sinks": [ { - "type": "state-hub-progress", + "type": "hub-core-interaction-event", "event_type": "ops_inventory_probe", - "author": "activity-core", } ], }, @@ -266,12 +268,23 @@ def test_disabled_ops_probe_definition_can_emit_fixture_evidence( posts: list[dict[str, Any]] = [] def fake_progress_get(url: str, **kwargs: Any) -> _JsonResponse: - return _JsonResponse([]) + assert url.endswith("/ports/projections/interaction_events") + return _JsonResponse({"data": {"items": [{"id": "event-1"}]}}) def fake_progress_post(url: str, **kwargs: Any) -> _JsonResponse: + assert url.endswith("/ports/events/interaction") posts.append({"url": url, **kwargs}) - return _JsonResponse({"id": "progress-1"}) + return _JsonResponse( + { + "id": "event-1", + "status": "accepted", + "correlation_id": "12345678-aaaa-bbbb-cccc-123456789abc", + }, + status_code=202, + ) + runtime_config = _by_kind_name("ConfigMap", "actcore-runtime-config") + monkeypatch.setenv("HUB_CORE_BASE_URL", runtime_config["data"]["HUB_CORE_BASE_URL"]) monkeypatch.setattr(httpx, "get", fake_progress_get) monkeypatch.setattr(httpx, "post", fake_progress_post) @@ -288,8 +301,9 @@ def test_disabled_ops_probe_definition_can_emit_fixture_evidence( assert definition.enabled is False assert result[0]["status"] == "posted" - assert posts[0]["json"]["event_type"] == "ops_inventory_probe" - assert posts[0]["json"]["detail"]["probe"]["summary"]["ok"] == 4 + assert posts[0]["json"]["event_type"] == "hub.interaction.recorded" + assert posts[0]["json"]["payload"]["reported_event_type"] == "ops_inventory_probe" + assert posts[0]["json"]["payload"]["probe"]["summary"]["ok"] == 4 class _HttpResponse: diff --git a/tests/test_sbom_nexus_context_resolver.py b/tests/test_sbom_nexus_context_resolver.py index f6a16e3..b0cd33c 100644 --- a/tests/test_sbom_nexus_context_resolver.py +++ b/tests/test_sbom_nexus_context_resolver.py @@ -8,7 +8,10 @@ import httpx import pytest from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY -from activity_core.context_resolvers.sbom_nexus import SbomNexusContextResolver +from activity_core.context_resolvers.sbom_nexus import ( + SbomNexusContextResolver, + apply_bounded_ingest, +) class DummyResponse: @@ -36,7 +39,7 @@ 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.posts: list[tuple[str, dict[str, Any] | None, dict[str, str]]] = [] self.post_results: dict[str, list[DummyResponse]] = {} def __enter__(self) -> "DummyClient": @@ -49,8 +52,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)) + 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) @@ -160,7 +168,7 @@ 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: +def test_declared_apply_remains_read_only_during_selection(monkeypatch) -> None: repos = [ { "repo_slug": "no-checkout", @@ -182,6 +190,31 @@ def test_apply_processes_at_most_limit_and_records_terminal_outcomes(monkeypatch }, ] 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( @@ -205,8 +238,9 @@ def test_apply_processes_at_most_limit_and_records_terminal_outcomes(monkeypatch ], } - result = SbomNexusContextResolver().resolve( - "catch_up", None, {"limit": 2, "apply": True} + result = apply_bounded_ingest( + repos, + operation_id="run-1", ) assert result["attempted_count"] == 2 @@ -227,48 +261,141 @@ def test_apply_processes_at_most_limit_and_records_terminal_outcomes(monkeypatch } ] assert len(client.posts) == 2 - assert all("not-selected" not in url for url, _body in client.posts) + 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_records_ingest_error_when_ingest_fails(monkeypatch) -> None: +def test_ambiguous_ingest_failure_does_not_create_synthetic_skip(monkeypatch) -> None: client = _install( monkeypatch, - _payload( - [ - { - "repo_slug": "broken", - "last_sbom_at": None, - "has_sbom": False, - "checkout_available": True, - } - ] - ), + _payload([]), ) client.post_results = { "/sbom/broken/ingest": [DummyResponse({}, status_code=503)], - "/sbom/broken/skip": [ + } + + 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": "broken", - "status": "skipped", - "reason": "ingest-error", - "snapshot_id": "skip-error-1", + "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") - result = SbomNexusContextResolver().resolve( - "catch_up", None, {"limit": 3, "apply": True} + 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 result["updated"] == [] - assert result["skipped"][0]["reason"] == "ingest-error" - assert client.posts[1][1] == { - "reason": "ingest-error", - "detail": "HTTPStatusError", - } + assert len(client.posts) == 1 def test_read_only_default_never_posts(monkeypatch) -> None: @@ -281,6 +408,106 @@ def test_read_only_default_never_posts(monkeypatch) -> None: 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, diff --git a/workplans/ACTIVITY-WP-0029-hub-port-alignment.md b/workplans/ACTIVITY-WP-0029-hub-port-alignment.md index d5281fd..8dc9215 100644 --- a/workplans/ACTIVITY-WP-0029-hub-port-alignment.md +++ b/workplans/ACTIVITY-WP-0029-hub-port-alignment.md @@ -4,11 +4,11 @@ type: workplan title: "Hub port and schedule alignment for State Hub retirement" domain: infotech repo: activity-core -status: active +status: finished owner: grok topic_slug: infotech created: "2026-08-09" -updated: "2026-08-18" +updated: "2026-08-22" parent_project: prj-state-hub-retirement parent_workplan: SHR-WP-0001 stream: S5 @@ -107,16 +107,27 @@ inside this repo. ```task id: ACTIVITY-WP-0029-T03 -status: wait +status: done priority: medium state_hub_task_id: "3fbadbfe-9107-4869-b53f-3b9b46e0555a" ``` -Wait on **HUB-WP-0004** (ports still `proposed`) and the CORE-WP-0010 -absorption path. Interaction-event / progress evidence follows the -**hub-core** runtime, not a permanent core-hub or State Hub host. Update -smokes when the port exists. Until then, keep the State Hub beachhead -(ACTIVITY-WP-0015) as the compatibility sink. +HUB-WP-0004 finished and hub-core became production authority on 2026-08-21. +Activity-core now implements `hub-core-interaction-event` against canonical +`POST /ports/events/interaction`, then verifies the accepted id through +`GET /ports/projections/interaction_events`. The Railiance projection points +`HUB_CORE_BASE_URL` at the in-cluster hub-core runtime and uses this sink; it +does not require the legacy Core Hub widget mapping/runtime-token contract. + +The old `core-hub-interaction-event` compatibility sink remains available for +the no-writer rollback window, but is no longer the configured probe path. +General progress callers not yet moved by the retirement strangler continue to +use the State Hub beachhead rather than creating another permanent authority. +Unit and deployment-wiring tests cover catalog envelope shape, readback, +sanitization, missing configuration, and the in-cluster target. A disposable +integration smoke against hub-core 0.2.0 returned `202 Accepted` from the +canonical interaction port and verified the new event through the interaction +projection. ## Execution queue boundary @@ -138,6 +149,6 @@ index, issue-core lifecycle, ITC Task Model types, policy-nexus publication. (`docs/state-hub-caller-map.md`) - [x] Sweep still fires from activity-core and hits repo-manager (or dual-run) (`CONSISTENCY_SWEEP_URL` / `REPO_MANAGER_URL`; default State Hub adapter) -- [x] Evidence sink plan names hub-core as the permanent host; T03 stays wait until that port exists +- [x] Ops evidence uses hub-core's canonical interaction port; the legacy Core Hub sink is rollback-only - [x] Execution boundary is written so `/execution/*` cannot be read as "activity-core owns tasks" (`docs/execution-queue-boundary.md`, `GET /execution/semantics`, 410 on workplan routes) diff --git a/workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md b/workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md index 1c80eea..98e17ad 100644 --- a/workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md +++ b/workplans/ACTIVITY-WP-0030-daily-sbom-catchup.md @@ -4,7 +4,7 @@ type: workplan title: "Daily bounded SBOM catch-up via sbom-nexus" domain: infotech repo: activity-core -status: done +status: finished owner: grok topic_slug: infotech created: "2026-08-18" diff --git a/workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md b/workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md new file mode 100644 index 0000000..1c6c7c2 --- /dev/null +++ b/workplans/ACTIVITY-WP-0033-sbom-catchup-retry-boundary.md @@ -0,0 +1,90 @@ +--- +id: ACTIVITY-WP-0033 +type: workplan +title: "Bound SBOM catch-up across Temporal retries" +domain: infotech +repo: activity-core +status: finished +owner: codex +topic_slug: infotech +created: "2026-08-22" +updated: "2026-08-22" +related: + - ACTIVITY-WP-0030 + - CUST-WP-0062 + - SBOM-WP-0002 +--- + +# Bound SBOM catch-up across Temporal retries + +## Goal + +Preserve ACTIVITY-WP-0030's oldest-N bound when Temporal retries an activity or +an SBOM Nexus response is ambiguous. Activity Core can independently guarantee +that one workflow fire retains one fixed set of at most N distinct repository +targets. Exact once-only snapshot creation remains a Nexus-side idempotency +follow-up. + +## Separate selection from mutation + +```task +id: ACTIVITY-WP-0033-T01 +status: done +priority: high +``` + +Keep `resolve_context` and the `sbom-nexus / catch_up` resolver read-only even +when the definition declares `apply: true`. Compute the deterministic run id +before context resolution, then pass the already-truncated selection to a +dedicated Temporal activity. Retries must reuse that same selection rather than +querying the next batch. + +## Resume bounded outcomes safely + +```task +id: ACTIVITY-WP-0033-T02 +status: done +priority: high +``` + +Heartbeat completed per-repository outcomes so an ordinary activity retry does +not repeat acknowledged targets. Send a stable per-run/per-repository +`Idempotency-Key` for forward compatibility with Nexus enforcement. Never turn +an HTTP timeout, transport error, or malformed success response into a second +synthetic skip: the result is ambiguous and must fail visibly against the same +fixed target set. + +## Verification and handoff + +```task +id: ACTIVITY-WP-0033-T03 +status: done +priority: medium +``` + +Cover read-only selection, fixed-target retry, heartbeat resume, duplicate +target collapse, stable request identity, and ambiguous-response behavior. +Update the runbook and hand the remaining exactly-once enforcement requirement +to SBOM Nexus without claiming it is solved locally. + +Completed 2026-08-22. `resolve_context` now only selects and normalises a +deduplicated oldest-N set. `RunActivityWorkflow` records that selection before +calling the registered `apply_sbom_catchup` activity, protected by Temporal +patch marker `activity-wp-0033-sbom-retry-boundary`. The apply activity resumes +terminal outcomes from heartbeat details and sends a deterministic UUID +operation key for each workflow-run/repository pair. Ambiguous HTTP and +contract failures remain failures against the same fixed target instead of +creating `ingest-error` skip snapshots. + +Verification: 27 focused SBOM tests and 417 repository tests passed; the one +live NATS-to-Temporal bridge test was deselected because it requires the local +integration stack. Python compilation and `git diff --check` also passed. + +## Acceptance + +- [x] Context resolution performs no SBOM mutation +- [x] One workflow fire retains at most N distinct targets across retries +- [x] Heartbeat replay skips outcomes already acknowledged by the activity +- [x] Ambiguous ingest responses do not create synthetic skip snapshots +- [x] Stable operation identity is sent for future Nexus enforcement +- [x] Focused and repository-wide Activity Core verification pass