orchestrata_ultimate.kernel.pipeline_events

Reference › orchestrata_ultimate › kernel › pipeline_events

The milestone event stream of a pipeline run. Whenever the runner boots, takes an intent, escalates between volks, runs a review or tests, or finishes, it emits a PipelineEvent so that a UI or CLI can watch the run live. The module serves SPEC-001 story US-006 (REQ-001-US-006-020). It is imported by pipeline, pipeline_escalation, pipeline_session, the streaming client cli_watch, the gateway server and orchestrator.pipeline_kernel. It is separate from the system-wide event_bus.

PipelineEventType is a StrEnum that is the single source of truth for the taxonomy: boot, intent, escalation, swarm start, shard and done, review, test result, complete, stage, paused, resumed and cancelled, each a pipeline.-prefixed string, so existing string comparisons keep working. PipelineEvent carries the type, thread id, volk, iteration, timestamp, data and a driver identity; to_payload serialises it. The build_*_event functions construct the common events, and pipeline_topic returns the per-thread UI topic name.

PipelineEventBus.emit writes the JSON event to a Redis list, trimmed to a maximum length, forwards it to the UI, and calls the registered callbacks (added with on). The driver field is filled from the bus's own identity (inprocess by default) on the serialised copy only, and a producer-supplied driver is kept. _forward_to_ui adds the event to two Redis streams the gateway already reads: a per-thread ring buffer, so reconnecting clients can replay, and a global stream the gateway forwarder tails to broadcast onto the thread topic. Because the kernel and gateway share Redis, this works across processes. Without a Redis client, events only reach the log and the callbacks. Redis and callback errors are logged and never raised. The emit_* helpers (emit_boot, emit_intent, emit_escalation, emit_review, emit_test_result, emit_complete and others) wrap this for each event type.

At a glance

Class diagram

Dependency diagram

API

Classes

PipelineEventType (enum)

Bases: StrEnum · Decorators: traces(spec='001', story='US-006', reqs=['REQ-006-031', 'REQ-006-032', 'REQ-006-033']) · Line 38

The pipeline event taxonomy — the single source of truth.

PipelineEvent (dataclass)

Decorators: traces(spec='001', story='US-006', reqs=['REQ-001-US-006-020']), dataclass · Line 86

A milestone event in the pipeline lifecycle.

PipelineEventBus

Line 201

Publishes pipeline events to Redis and optional callbacks.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/pipeline_events.py

# mypy: ignore-errors
"""Pipeline milestone events — emitted at each stage transition.

Events are published to Redis (orchestrata:pipeline:events) and
optionally to the graph for persistent audit trail. External consumers
(goal_runner, dashboards, hooks) subscribe to detect milestones
without parsing conversation.md.

Event types follow the pipeline lifecycle (see ``PipelineEventType`` for the
authoritative, enumerable source of truth):
  pipeline.boot          — kernel booted, volks registered
  pipeline.intent        — intent submitted, thread created
  pipeline.escalation    — volk-to-volk handoff
  pipeline.swarm.start   — swarm fan-out begins
  pipeline.swarm.shard   — individual shard completed
  pipeline.swarm.done    — all shards completed
  pipeline.review        — self-review or deterministic review result
  pipeline.test.result   — test results (pass/fail/count/coverage)
  pipeline.complete      — pipeline finished (success or failure)
  pipeline.stage         — a pipeline stage finished (status/duration/artifacts)
  pipeline.paused        — pipeline execution paused
  pipeline.resumed       — pipeline execution resumed
  pipeline.cancelled     — pipeline execution cancelled
"""

import json
import logging
from collections.abc import Callable
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-032", "REQ-006-033"])
class PipelineEventType(StrEnum):
    """The pipeline event taxonomy — the single source of truth.

    Members equal the pre-existing string literals byte-for-byte
    (``StrEnum`` is a ``str`` subclass), so this is a purely additive,
    wire-compatible change: existing string comparisons/consumers keep
    working unchanged.
    """

    BOOT = "pipeline.boot"
    INTENT = "pipeline.intent"
    ESCALATION = "pipeline.escalation"
    SWARM_START = "pipeline.swarm.start"
    SWARM_SHARD = "pipeline.swarm.shard"
    SWARM_DONE = "pipeline.swarm.done"
    REVIEW = "pipeline.review"
    TEST_RESULT = "pipeline.test.result"
    COMPLETE = "pipeline.complete"
    STAGE = "pipeline.stage"
    PAUSED = "pipeline.paused"
    RESUMED = "pipeline.resumed"
    CANCELLED = "pipeline.cancelled"


_logger = logging.getLogger(__name__)

# Redis key for event stream
EVENTS_KEY = "orchestrata:pipeline:events"
MAX_EVENTS = 500

# Per-thread Redis stream prefix the gateway's RedisRingBuffer reads from
# (it prefixes stream names with ``gw:ringbuf:``), so the UI topic
# ``pipeline.<thread_id>`` is durably backed and replayable on reconnect.
RINGBUF_PREFIX = "gw:ringbuf:"
PIPELINE_TOPIC_PREFIX = "pipeline."
# Global stream the gateway forwarder tails to fan out live events to all
# ``pipeline.<thread_id>`` subscribers across processes.
PIPELINE_STREAM_KEY = f"{RINGBUF_PREFIX}pipeline.events"
PIPELINE_STREAM_MAXLEN = 1000


def pipeline_topic(thread_id: str) -> str:
    """Return the UI pub/sub topic for a pipeline thread."""
    return f"{PIPELINE_TOPIC_PREFIX}{thread_id}"


@traces(spec="001", story="US-006", reqs=["REQ-001-US-006-020"])
@dataclass
class PipelineEvent:
    """A milestone event in the pipeline lifecycle.

    Attributes:
        event_type: The event's type (``PipelineEventType`` member — a
            ``str`` subclass, so it serializes/compares identically to the
            legacy dotted string literal).
        thread_id: CXP thread this event belongs to.
        volk: Volk alias that triggered this event.
        data: Event-specific payload.
        timestamp: When the event occurred.
        iteration: Iteration number if known.
        driver: Provenance of the execution that produced this event
            (e.g. ``"inprocess"``, or ``"headless:<harness>"`` at Phase 2 —
            VolkDriver Phase 0, SPEC-001 US-006). Empty string means
            unattributed — legal, never an error.
    """

    event_type: PipelineEventType
    thread_id: str = ""
    volk: str = ""
    data: dict[str, Any] = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
    iteration: int = 0
    driver: str = ""

    def to_payload(self) -> dict[str, Any]:
        """The canonical broadcast serialization for this event.

        Implemented as exactly ``asdict(self)`` so the REQ-006-032
        golden-bytes wire-compat guard also pins this method: both
        producers (the kernel bus and the fallback ``PipelineKernelClient``)
        share this single serializer, so their payload shapes cannot drift.
        """
        return asdict(self)


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-033"])
def build_intent_event(thread_id: str, intent: str, volk: str = "") -> PipelineEvent:
    """Build a ``PipelineEventType.INTENT`` event — the single-source data
    shape shared by the kernel bus (``emit_intent``) and the fallback
    producer's legacy ``pipeline.start`` fold (``volk="U:Cx"``)."""
    return PipelineEvent(
        event_type=PipelineEventType.INTENT,
        thread_id=thread_id,
        volk=volk,
        data={"intent": intent[:200]},
    )


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-033"])
def build_stage_event(
    thread_id: str,
    stage: str,
    status: str,
    duration_ms: float,
    artifact_count: int,
    error: str | None = None,
) -> PipelineEvent:
    """Build a ``PipelineEventType.STAGE`` event — one pipeline stage's
    result. ``volk`` is the stage alias itself (the stage IS the actor)."""
    return PipelineEvent(
        event_type=PipelineEventType.STAGE,
        thread_id=thread_id,
        volk=stage,
        data={
            "stage": stage,
            "status": status,
            "duration_ms": duration_ms,
            "artifact_count": artifact_count,
            "error": error,
        },
    )


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-033"])
def build_complete_event(
    thread_id: str,
    success: bool,
    reason: str = "",
    extra: dict[str, Any] | None = None,
) -> PipelineEvent:
    """Build a ``PipelineEventType.COMPLETE`` event. ``extra`` merges
    additional data fields (e.g. ``stages_completed``/``total_artifacts``
    from the legacy ``pipeline.complete`` fold, or ``error`` from the
    legacy ``pipeline.error`` fold) on top of the canonical
    ``{success, reason}`` pair."""
    data: dict[str, Any] = {"success": success, "reason": reason}
    if extra:
        data.update(extra)
    return PipelineEvent(
        event_type=PipelineEventType.COMPLETE,
        thread_id=thread_id,
        data=data,
    )


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-033"])
def build_paused_event(thread_id: str) -> PipelineEvent:
    """Build a ``PipelineEventType.PAUSED`` event."""
    return PipelineEvent(event_type=PipelineEventType.PAUSED, thread_id=thread_id)


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-033"])
def build_resumed_event(thread_id: str) -> PipelineEvent:
    """Build a ``PipelineEventType.RESUMED`` event."""
    return PipelineEvent(event_type=PipelineEventType.RESUMED, thread_id=thread_id)


@traces(spec="001", story="US-006", reqs=["REQ-006-031", "REQ-006-033"])
def build_cancelled_event(thread_id: str) -> PipelineEvent:
    """Build a ``PipelineEventType.CANCELLED`` event."""
    return PipelineEvent(event_type=PipelineEventType.CANCELLED, thread_id=thread_id)


class PipelineEventBus:
    """Publishes pipeline events to Redis and optional callbacks.

    Attributes:
        redis: Redis client (optional — events still logged if None).
        callbacks: List of synchronous callback functions.
    """

    def __init__(
        self,
        redis_client: Any | None = None,
        events_key: str = EVENTS_KEY,
        ui_stream_key: str | None = None,
        driver: str = "inprocess",
    ) -> None:
        """Initialize the event bus.

        Args:
            redis_client: Redis client instance. If None, events
                are only logged and sent to callbacks.
            events_key: Redis list the bus publishes to. Production uses
                the module default; tests pass a test-scoped key so
                parallel suite batches can't interleave on one list.
            ui_stream_key: Global Redis stream ``_forward_to_ui`` XADDs to
                (mirrors the gateway pump's ``stream_key`` injection seam,
                :func:`_pump_pipeline_events_once`). Defaults to
                ``PIPELINE_STREAM_KEY`` — zero behaviour change when unset.
            driver: This bus's own producer identity (VolkDriver Phase 0,
                SPEC-001 US-006). ``emit`` resolves it onto the OUTPUT
                payload for any event the caller left unattributed —
                ``"inprocess"`` by default, ``"headless:<harness>"`` for a
                Phase 2 headless-harness bus.
        """
        self._redis: Any | None = redis_client
        self._events_key = events_key
        self._ui_stream_key = ui_stream_key or PIPELINE_STREAM_KEY
        self._callbacks: list[Callable[[PipelineEvent], None]] = []
        self._driver = driver

    def on(self, callback: Callable[[PipelineEvent], None]) -> None:
        """Register a callback for all events.

        Args:
            callback: Function taking a PipelineEvent.
        """
        self._callbacks.append(callback)

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-020"])
    def emit(self, event: PipelineEvent) -> None:
        """Publish an event to Redis and all callbacks.

        Resolves ``driver`` to this bus's own producer identity
        (``self._driver``, "inprocess" by default) on the OUTPUT payload
        when the caller left it unset — attribution belongs to the
        SERIALIZED wire representation, never mutated onto the caller's
        object (VolkDriver Phase 0). A caller-supplied driver (Phase 2
        headless-harness producers) is never overridden. Callbacks receive
        the original, unmutated *event* — driver resolution is a
        wire/serialization concern, not an in-process one.

        Args:
            event: The pipeline event to publish.
        """
        event_dict = event.to_payload()
        event_dict["driver"] = event_dict["driver"] or self._driver
        event_json = json.dumps(event_dict, default=str)

        _logger.debug(
            "[EVENT] %s volk=%s %s",
            event.event_type,
            event.volk,
            json.dumps(event.data, default=str)[:100],
        )

        # Publish to Redis
        if self._redis:
            try:
                self._redis.lpush(self._events_key, event_json)
                self._redis.ltrim(self._events_key, 0, MAX_EVENTS - 1)
            except Exception as exc:
                _logger.warning("Event publish failed: %s", exc)
            self._forward_to_ui(event, event_dict)

        # Fire callbacks
        for cb in self._callbacks:
            try:
                cb(event)
            except Exception as exc:
                _logger.warning("Event callback failed: %s", exc)

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-020"])
    def _forward_to_ui(self, event: PipelineEvent, event_dict: dict[str, Any]) -> None:
        """Bridge an event to the gateway's UI plumbing (Redis streams).

        Writes the event to two Redis streams the gateway already understands:
        - ``gw:ringbuf:pipeline.<thread_id>`` — durable per-thread replay, read
          directly by the gateway's ``RedisRingBuffer`` (which prefixes stream
          names with ``gw:ringbuf:``), so reconnecting clients see prior events.
        - ``gw:ringbuf:pipeline.events`` — a global stream the gateway forwarder
          tails to broadcast live onto ``pipeline.<thread_id>``.

        Cross-process safe: the kernel and gateway share Redis even when the
        kernel runs out-of-process behind gRPC. Best-effort and non-fatal.

        Args:
            event: The event being emitted.
            event_dict: ``asdict(event)`` (already computed by ``emit``).
        """
        if not self._redis:
            return
        # Redis streams require flat string fields; nested data is JSON-encoded.
        fields = {
            "event_type": event.event_type,
            "thread_id": event.thread_id,
            "volk": event.volk,
            "iteration": str(event.iteration),
            "timestamp": event.timestamp,
            "data": json.dumps(event.data, default=str),
            "driver": event_dict["driver"],
        }
        try:
            if event.thread_id:
                self._redis.xadd(
                    f"{RINGBUF_PREFIX}{pipeline_topic(event.thread_id)}",
                    fields,
                    maxlen=PIPELINE_STREAM_MAXLEN,
                    approximate=True,
                )
            self._redis.xadd(
                self._ui_stream_key,
                fields,
                maxlen=PIPELINE_STREAM_MAXLEN,
                approximate=True,
            )
        except Exception as exc:
            _logger.warning("UI event forward failed: %s", exc)

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-007"])
    def emit_boot(self, volks: list[str], thread_id: str = "") -> None:
        """Emit pipeline.boot event.

        Args:
            volks: List of registered volk aliases.
            thread_id: Thread ID if available.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.BOOT,
                thread_id=thread_id,
                data={"volks": volks, "count": len(volks)},
            )
        )

    def emit_intent(
        self,
        thread_id: str,
        intent: str,
    ) -> None:
        """Emit pipeline.intent event.

        Args:
            thread_id: CXP thread ID.
            intent: The submitted intent text.
        """
        self.emit(build_intent_event(thread_id, intent))

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-003"])
    def emit_escalation(
        self,
        thread_id: str,
        from_volk: str,
        to_volk: str,
        subject: str,
    ) -> None:
        """Emit pipeline.escalation event.

        Args:
            thread_id: CXP thread ID.
            from_volk: Source volk alias.
            to_volk: Destination volk alias.
            subject: CXP packet subject.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.ESCALATION,
                thread_id=thread_id,
                volk=from_volk,
                data={
                    "from": from_volk,
                    "to": to_volk,
                    "subject": subject,
                },
            )
        )

    def emit_swarm_start(
        self,
        thread_id: str,
        shard_count: int,
    ) -> None:
        """Emit pipeline.swarm.start event.

        Args:
            thread_id: CXP thread ID.
            shard_count: Number of parallel shards.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.SWARM_START,
                thread_id=thread_id,
                volk="U:Cd",
                data={"shard_count": shard_count},
            )
        )

    def emit_swarm_shard(
        self,
        thread_id: str,
        shard_id: int,
        file_path: str,
        success: bool,
    ) -> None:
        """Emit pipeline.swarm.shard event.

        Args:
            thread_id: CXP thread ID.
            shard_id: Shard number.
            file_path: File this shard generated.
            success: Whether the shard succeeded.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.SWARM_SHARD,
                thread_id=thread_id,
                volk="J:Junior",
                data={
                    "shard_id": shard_id,
                    "file": file_path,
                    "success": success,
                },
            )
        )

    def emit_swarm_done(
        self,
        thread_id: str,
        succeeded: int,
        total: int,
    ) -> None:
        """Emit pipeline.swarm.done event.

        Args:
            thread_id: CXP thread ID.
            succeeded: Number of successful shards.
            total: Total shards.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.SWARM_DONE,
                thread_id=thread_id,
                volk="U:Cd",
                data={"succeeded": succeeded, "total": total},
            )
        )

    def emit_review(
        self,
        thread_id: str,
        volk: str,
        review_type: str,
        passed: bool,
        cycle: int = 0,
        issues: int = 0,
    ) -> None:
        """Emit pipeline.review event.

        Args:
            thread_id: CXP thread ID.
            volk: Reviewing volk alias.
            review_type: "self_review" or "deterministic".
            passed: Whether the review passed.
            cycle: Review cycle number.
            issues: Number of issues found.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.REVIEW,
                thread_id=thread_id,
                volk=volk,
                data={
                    "type": review_type,
                    "passed": passed,
                    "cycle": cycle,
                    "issues": issues,
                },
            )
        )

    def emit_test_result(
        self,
        thread_id: str,
        passed: bool,
        pass_count: int,
        fail_count: int,
        coverage_pct: float = 0.0,
        duration_s: float = 0.0,
    ) -> None:
        """Emit pipeline.test.result event.

        Args:
            thread_id: CXP thread ID.
            passed: Whether all tests passed.
            pass_count: Number of passed tests.
            fail_count: Number of failed tests.
            coverage_pct: Line coverage percentage.
            duration_s: Test execution time.
        """
        self.emit(
            PipelineEvent(
                event_type=PipelineEventType.TEST_RESULT,
                thread_id=thread_id,
                volk="U:Qa",
                data={
                    "passed": passed,
                    "pass_count": pass_count,
                    "fail_count": fail_count,
                    "coverage_pct": coverage_pct,
                    "duration_s": duration_s,
                },
            )
        )

    def emit_complete(
        self,
        thread_id: str,
        success: bool,
        reason: str = "",
    ) -> None:
        """Emit pipeline.complete event.

        Args:
            thread_id: CXP thread ID.
            success: Whether the pipeline succeeded.
            reason: Completion reason (e.g. "tests_pass", "shutdown").
        """
        self.emit(build_complete_event(thread_id, success, reason))

    def emit_stage(
        self,
        thread_id: str,
        stage: str,
        status: str,
        duration_ms: float,
        artifact_count: int,
        error: str | None = None,
    ) -> None:
        """Emit pipeline.stage event — a pipeline stage's result.

        Args:
            thread_id: CXP thread ID.
            stage: Stage alias (also used as ``volk`` — the stage IS the actor).
            status: Stage outcome (e.g. "resolved", "failed").
            duration_ms: How long the stage took, in milliseconds.
            artifact_count: Number of artifacts the stage produced.
            error: Error message if the stage failed, else None.
        """
        self.emit(
            build_stage_event(
                thread_id, stage, status, duration_ms, artifact_count, error
            )
        )

    def emit_paused(self, thread_id: str) -> None:
        """Emit pipeline.paused event.

        Args:
            thread_id: CXP thread ID.
        """
        self.emit(build_paused_event(thread_id))

    def emit_resumed(self, thread_id: str) -> None:
        """Emit pipeline.resumed event.

        Args:
            thread_id: CXP thread ID.
        """
        self.emit(build_resumed_event(thread_id))

    def emit_cancelled(self, thread_id: str) -> None:
        """Emit pipeline.cancelled event.

        Args:
            thread_id: CXP thread ID.
        """
        self.emit(build_cancelled_event(thread_id))