"""Temporal workflow definitions for activity-core. Two workflows are registered here: - RunActivityWorkflow → orchestrator-tq - TaskExecutorWorkflow → task-execution-tq Workflow IDs follow the conventions in docs/conventions.md: RunActivityWorkflow: activity-{activity_id}:{trigger_key} TaskExecutorWorkflow: task-{run_id}:{task_type}:{index} """ 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, evaluate_rules, evaluate_instructions, load_activity_definition, log_run, persist_instruction_reports, persist_ops_evidence, persist_task_instance, resolve_context, ) 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")) ) _TASK_QUEUE = "task-execution-tq" @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. apply_sbom_catchup(fixed selection) → bounded outcome patch 4. evaluate rules/instructions → TaskSpec dicts and reports 5. 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. 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, { "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, ) # ── 4. 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", [])) # ── 5. 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, ) # ── 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; # 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, ) # ── 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 # 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, }, start_to_close_timeout=_ACTIVITY_TIMEOUT, retry_policy=_RETRY_POLICY, ) return {"run_id": run_id, "tasks_spawned": len(task_spec_dicts)} @workflow.defn class TaskExecutorWorkflow: """LEGACY NO-OP — not a production execution surface (ACTIVITY-WP-0023-T08). Historical compatibility stub. Real task execution belongs in per-repo workers / agent-harness, not activity-core. Behaviour is controlled by ``ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB``: - unset / false (default): refuse to run (logs error, raises) so the stub cannot attract production work by accident. - true: legacy behaviour — persist a done ``task_instances`` row for idempotent dev/test callers only. """ @workflow.run async def run(self, run_id: str, task_type: str, params: dict) -> dict: enabled = ( os.environ.get("ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB", "") .strip() .lower() in {"1", "true", "yes", "on"} ) task_id = str( uuid.uuid5(uuid.NAMESPACE_URL, workflow.info().workflow_id) ) if not enabled: workflow.logger.error( "TaskExecutorWorkflow refused: stub disabled " "(set ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true only for legacy tests). " "Real execution belongs in per-repo workers / agent-harness. " "See docs/task-emission-consumer-contract.md" ) raise RuntimeError( "TaskExecutorWorkflow is disabled (ACTIVITY-WP-0023-T08). " "Use per-repo executors; do not route production work here." ) workflow.logger.warning( "TaskExecutorWorkflow stub running (legacy mode)", extra={"run_id": run_id, "task_type": task_type, "task_id": task_id}, ) await workflow.execute_activity( persist_task_instance, { "id": task_id, "run_id": run_id, "type": task_type, "params": params, "status": "done", }, task_queue=_TASK_QUEUE, start_to_close_timeout=_ACTIVITY_TIMEOUT, retry_policy=_RETRY_POLICY, ) return {"task_id": task_id, "status": "done", "legacy_stub": True}