activity-core/src/activity_core/workflows.py

"""Temporal workflow definitions for activity-core.

RunActivityWorkflow is registered on ``orchestrator-tq``. Workflow IDs follow
``activity-{activity_id}:{trigger_key}``; see docs/conventions.md.
"""

from __future__ import annotations

import os
import uuid
from datetime import timedelta

from temporalio import workflow
from temporalio.common import RetryPolicy, SearchAttributeKey, TypedSearchAttributes, SearchAttributePair

with workflow.unsafe.imports_passed_through():
    from activity_core.activities import (
        apply_sbom_catchup,
        emit_tasks,
        execute_bounded_operation,
        evaluate_rules,
        evaluate_instructions,
        load_activity_definition,
        log_run,
        persist_instruction_reports,
        persist_ops_evidence,
        resolve_context,
    )
    from activity_core.bounded_operations import operation_sources
    from activity_core.ops_run_queue import emit_triggering_event_id
    from activity_core.schedule_manager import SCHEDULED_TRIGGER_KEY

# T32: Custom search attributes for Temporal visibility (must be registered in Temporal first).
# Registration: temporal operator search-attribute create --name ActivityId --type Keyword
_ACTIVITY_ID_KEY = SearchAttributeKey.for_keyword("ActivityId")
_ACTIVITY_NAME_KEY = SearchAttributeKey.for_keyword("ActivityName")

_RETRY_POLICY = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(minutes=5),
    maximum_attempts=10,
)

_ACTIVITY_TIMEOUT = timedelta(
    seconds=int(os.environ.get("ACTIVITY_TIMEOUT_SECONDS", "900"))
)
_NO_RETRY_POLICY = RetryPolicy(maximum_attempts=1)


@workflow.defn
class RunActivityWorkflow:
    """Durable orchestration workflow.

    Sequence:
      1. load_activity_definition(activity_id)  → defn dict
      2. resolve_context(defn.context_sources)  → read-only context snapshot
      3. execute admitted bounded operations    → bounded outcome patches
      4. persist operation evidence
      5. evaluate rules/instructions            → TaskSpec dicts and reports
      6. log run, then emit tasks
    """

    @workflow.run
    async def run(
        self,
        activity_id: str,
        trigger_key: str,
        scheduled_for: str | None = None,
        event_envelope_json: str | None = None,
    ) -> dict:
        """
        Args:
            activity_id:          UUID of the ActivityDefinition row.
            trigger_key:          event_id (event trigger) or "scheduled" (cron).
            scheduled_for:        ISO-8601 nominal scheduled time (cron only).
            event_envelope_json:  JSON-serialised EventEnvelope (event trigger only).

        Returns:
            {"run_id": str, "tasks_spawned": int}
        """
        # ── 1. Load definition ────────────────────────────────────────────────
        defn: dict = await workflow.execute_activity(
            load_activity_definition,
            activity_id,
            start_to_close_timeout=_ACTIVITY_TIMEOUT,
            retry_policy=_RETRY_POLICY,
        )

        # T32: Tag this workflow execution with activity metadata so runs are
        # filterable in the Temporal UI (requires ActivityId + ActivityName to be
        # registered as custom search attributes — see docs/runbook.md).
        workflow.upsert_search_attributes(
            TypedSearchAttributes([
                SearchAttributePair(_ACTIVITY_ID_KEY, activity_id),
                SearchAttributePair(_ACTIVITY_NAME_KEY, defn.get("name", "")),
            ])
        )

        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],
            start_to_close_timeout=_ACTIVITY_TIMEOUT,
            retry_policy=_RETRY_POLICY,
        )

        # ── 3. Execute admitted bounded operations ──────────────────────────
        # SBOM retains its dedicated heartbeat/retry activity because it has a
        # stable remote idempotency key.  Shell-backed operations are one-shot:
        # their platform tools do not expose a receipt activity-core can use to
        # prove an ambiguous mutation safe to repeat.
        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)

        for source, operation_spec in operation_sources(
            defn.get("context_sources", [])
        ):
            if operation_spec.operation_id == "sbom_nexus_ingest":
                continue
            operation_patch: dict = await workflow.execute_activity(
                execute_bounded_operation,
                {"source": source, "run_id": run_id},
                start_to_close_timeout=timedelta(
                    seconds=operation_spec.max_timeout_seconds
                ),
                retry_policy=_NO_RETRY_POLICY,
            )
            context_snapshot.update(operation_patch)

        # ── 4. Persist bounded operation/read evidence ───────────────────────
        await workflow.execute_activity(
            persist_ops_evidence,
            {
                "context_sources": defn.get("context_sources", []),
                "context": context_snapshot,
                "activity_id": activity_id,
                "run_id": run_id,
                "scheduled_for": scheduled_for,
                "version_used": defn["version"],
            },
            start_to_close_timeout=_ACTIVITY_TIMEOUT,
            retry_policy=_RETRY_POLICY,
        )

        # ── 5. Evaluate rules ─────────────────────────────────────────────────
        import json as _json
        event_attrs: dict = {}
        if event_envelope_json:
            try:
                event_attrs = _json.loads(event_envelope_json).get("attributes", {})
            except Exception:
                pass

        task_spec_dicts: list[dict] = await workflow.execute_activity(
            evaluate_rules,
            {
                "rules": defn.get("rules", []),
                "event": event_attrs,
                "context": context_snapshot,
            },
            start_to_close_timeout=_ACTIVITY_TIMEOUT,
            retry_policy=_RETRY_POLICY,
        )

        report_dicts: list[dict] = []
        if defn.get("instructions"):
            instruction_result: dict = await workflow.execute_activity(
                evaluate_instructions,
                {
                    "instructions": defn.get("instructions", []),
                    "event": event_attrs,
                    "context": context_snapshot,
                },
                start_to_close_timeout=_ACTIVITY_TIMEOUT,
                retry_policy=_RETRY_POLICY,
            )
            task_spec_dicts.extend(instruction_result.get("task_specs", []))
            report_dicts.extend(instruction_result.get("reports", []))

        # ── 6. Persist reports ────────────────────────────────────────────────
        if report_dicts:
            await workflow.execute_activity(
                persist_instruction_reports,
                {
                    "reports": report_dicts,
                    "activity_id": activity_id,
                    "run_id": run_id,
                    "scheduled_for": scheduled_for,
                    "version_used": defn["version"],
                },
                start_to_close_timeout=_ACTIVITY_TIMEOUT,
                retry_policy=_RETRY_POLICY,
            )

        # ── 7. 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;
        # then emit and re-raise so Temporal still surfaces sink failures.
        await workflow.execute_activity(
            log_run,
            {
                "run_id": run_id,
                "activity_id": activity_id,
                "scheduled_for": scheduled_for,
                "context_snapshot": context_snapshot,
                "tasks_spawned": len(task_spec_dicts),
                "version_used": defn["version"],
            },
            start_to_close_timeout=_ACTIVITY_TIMEOUT,
            retry_policy=_RETRY_POLICY,
        )

        # ── 8. 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
            # a bare "scheduled" key collides after the first day and silently
            # skips create (claim loop starves; dual-clock timers paper over it).
            # run_id is uuid5(workflow_id) for schedules and is unique per fire
            # while stable on Temporal redelivery of the same fire.
            emit_trigger_id = emit_triggering_event_id(
                trigger_key, run_id, scheduled_for=scheduled_for
            )
            await workflow.execute_activity(
                emit_tasks,
                {
                    "task_specs": task_spec_dicts,
                    "activity_id": activity_id,
                    "triggering_event_id": emit_trigger_id,
                    "run_id": run_id,
                    "version_used": defn["version"],
                },
                start_to_close_timeout=_ACTIVITY_TIMEOUT,
                retry_policy=_RETRY_POLICY,
            )

        return {"run_id": run_id, "tasks_spawned": len(task_spec_dicts)}