orchestrata_ultimate.gateway.handlers.workflow

Reference › orchestrata_ultimate › gateway › handlers › workflow

WorkflowHandler serves the workflow.start, workflow.pause, workflow.resume and workflow.cancel RPCs (SPEC-014 US-014-005), which control pipeline runs. It does not run pipelines itself; it calls a WorkflowControlPort from workflow_types, attached with set_control, which the in-process PipelineKernelClient satisfies. rpc creates it. See gateway.

Every verb first calls _authorize, which applies the scope guard with read_write and then the policy gate with the action name and a context holding the actor and the call parameters. _require_control raises RPCError (code -32603) when no control port is wired, because workflow control needs the in-process pipeline kernel. handle_start requires an intent, takes an optional work_type (default feature) and thread_id, and returns the thread id with status running. handle_pause, handle_resume and handle_cancel require a thread_id through _require_thread_id and return the resulting workflow state; an unknown thread id surfaces as an RPCError with code -32602.

At a glance

Class diagram

Dependency diagram

API

Classes

WorkflowHandler

Line 29

workflow.start/pause/resume/cancel -- PAM++-gated pipeline facade.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/workflow.py

"""Workflow control RPC handlers -- start/pause/resume/cancel.

Implements SPEC-014 US-014-005 (REQ-014-076..080). Pure facade/command
handler in the shape of the sibling handlers (CxpHandler, GovernanceHandler):
every ``handle_*`` first runs :meth:`_authorize` (ScopeGuard read_write +
PolicyGate PAM++ check on ``workflow.<verb>``), then delegates to the
injected :class:`WorkflowControlPort` -- structurally satisfied by
``orchestrator.pipeline_kernel.PipelineKernelClient`` (D1/D9). No duplicate
pipeline engine lives here.
"""

import logging
from collections.abc import Callable
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.gateway.handlers._policy_gate import PolicyGate
from orchestrata_ultimate.gateway.handlers._scope_guard import ScopeGuard
from orchestrata_ultimate.gateway.workflow_types import (
    WorkflowControlPort,
    WorkflowStatus,
)

logger = logging.getLogger(__name__)


class WorkflowHandler:
    """workflow.start/pause/resume/cancel -- PAM++-gated pipeline facade."""

    def __init__(
        self,
        policy_gate: PolicyGate,
        scope_guard: ScopeGuard,
        control: WorkflowControlPort | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        """Wire the handler to its authorization strategies and (optionally)
        an already-connected control port.

        Args:
            policy_gate: PAM++ gate (REQ-014-080).
            scope_guard: Per-connection scope enforcement.
            control: The workflow engine facade, or None until
                :meth:`set_control` is called (e.g. at gateway boot, once
                the in-process ``PipelineKernelClient`` connects).
            tracer: Injected debug tracer; defaults to the process Singleton.
        """
        self._policy_gate = policy_gate
        self._scope_guard = scope_guard
        self._control = control
        self._tracer = tracer or Tracer.default()

    def set_control(self, client: WorkflowControlPort) -> None:
        """Wire the live in-process pipeline kernel client.

        Called from ``RPCHandler.set_pipeline_client`` -- the same seam
        ``server.py``'s ``_init_kernel`` fallback path already uses, so
        pause/resume/cancel lifecycle events reach subscribed UI clients
        with zero new wiring (the client is already constructed with
        ``broadcast=self.broadcast``). Never wired on a pure gRPC-kernel
        deployment (D3): ``_require_control`` then answers a structured
        refusal instead of a silent no-op.

        Args:
            client: A connected ``WorkflowControlPort`` (in practice, the
                orchestrator's ``PipelineKernelClient``).
        """
        self._control = client

    def set_policy_engine_factory(self, factory: Callable[[], Any]) -> None:
        """Rebuild the PAM++ gate around *factory*.

        Called from ``RPCHandler.set_policy_engine_factory`` -- the ONE
        ``_make_policy_engine`` the gateway registers at boot
        (``server.py``'s ``_init_policy_engine``), shared with
        ``GovernanceHandler`` so REQ-014-080 enforces with the same real
        rule base as ``governance.evaluate``.

        Args:
            factory: ``() -> PolicyEngine``.
        """
        self._policy_gate = PolicyGate(engine_factory=factory, tracer=self._tracer)

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-076", "REQ-014-080"])
    async def handle_start(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """workflow.start -- spawn a new pipeline run (REQ-014-076).

        Args:
            connection_id: Caller connection ID.
            params: ``intent`` (required), optional ``work_type`` (default
                ``"feature"``) and ``thread_id``.

        Returns:
            ``{"thread_id": ..., "status": "running"}``.
        """
        await self._authorize(connection_id, "workflow.start", params)
        control = self._require_control()
        intent = params.get("intent", "")
        if not intent:
            raise RPCError("workflow.start requires 'intent'", rpc_code=-32602)
        work_type = params.get("work_type", "feature")
        thread_id = await control.start_workflow(
            intent, work_type, params.get("thread_id")
        )
        return {"thread_id": thread_id, "status": WorkflowStatus.RUNNING.value}

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-077", "REQ-014-080"])
    async def handle_pause(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """workflow.pause -- stop at the last completed-stage checkpoint (D2).

        Args:
            connection_id: Caller connection ID.
            params: ``thread_id`` (required).

        Returns:
            The paused ``WorkflowState`` as a dict.
        """
        await self._authorize(connection_id, "workflow.pause", params)
        control = self._require_control()
        thread_id = self._require_thread_id(params)
        state = await control.pause_workflow(thread_id)
        return state.to_dict()

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-078", "REQ-014-080"])
    async def handle_resume(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """workflow.resume -- re-enter ``PipelineRunner.resume()`` from the
        last checkpoint.

        Args:
            connection_id: Caller connection ID.
            params: ``thread_id`` (required).

        Returns:
            The resumed ``WorkflowState`` as a dict.

        Raises:
            RPCError: -32602 when ``thread_id`` is unknown to the control
                port.
        """
        await self._authorize(connection_id, "workflow.resume", params)
        control = self._require_control()
        thread_id = self._require_thread_id(params)
        try:
            state = await control.resume_workflow(thread_id)
        except ValueError as exc:
            raise RPCError(str(exc), rpc_code=-32602) from exc
        return state.to_dict()

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-079", "REQ-014-080"])
    async def handle_cancel(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """workflow.cancel -- terminate the active task (idempotent).

        Args:
            connection_id: Caller connection ID.
            params: ``thread_id`` (required).

        Returns:
            The cancelled ``WorkflowState`` as a dict.
        """
        await self._authorize(connection_id, "workflow.cancel", params)
        control = self._require_control()
        thread_id = self._require_thread_id(params)
        state = await control.cancel_workflow(thread_id)
        return state.to_dict()

    @staticmethod
    @traces(
        spec="014",
        story="US-014-005",
        reqs=["REQ-014-077", "REQ-014-078", "REQ-014-079"],
    )
    def _require_thread_id(params: dict[str, Any]) -> str:
        """Extract and validate the required ``thread_id`` param."""
        thread_id = params.get("thread_id")
        if not thread_id:
            raise RPCError("thread_id is required", rpc_code=-32602)
        return str(thread_id)

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-080"])
    async def _authorize(
        self, connection_id: str, action: str, params: dict[str, Any]
    ) -> None:
        """REQ-014-080: ScopeGuard (read_write) then the PAM++ PolicyGate
        check, in that order, before any handler touches the pipeline.

        Args:
            connection_id: Caller connection ID.
            action: The workflow verb being authorized (e.g.
                ``"workflow.cancel"``).
            params: The RPC params, folded into the governance context.
        """
        await self._scope_guard.require(connection_id, "read_write")
        context: dict[str, Any] = {
            "actor": connection_id,
            "connection_id": connection_id,
            **params,
        }
        await self._policy_gate.check(action, context)

    @traces(
        spec="014",
        story="US-014-005",
        reqs=["REQ-014-076", "REQ-014-077", "REQ-014-078", "REQ-014-079"],
    )
    def _require_control(self) -> WorkflowControlPort:
        """The wired control port, or a structured refusal (D3)."""
        if self._control is None:
            raise RPCError(
                "workflow control requires the in-process pipeline kernel",
                rpc_code=-32603,
            )
        return self._control