orchestrata_ultimate.gateway.handlers.orchestrator

Reference › orchestrata_ultimate › gateway › handlers › orchestrator

OrchestratorHandler serves the orchestrator.intent RPC: it takes a natural-language request from a client and turns it into a task and, when a kernel is wired, a dispatched pipeline run. rpc is its importer, and server wires the kernel client, reply consumer id, pub/sub manager and an optional intent evaluator through setters. See the orchestrator package and gateway.

handle_intent requires a non-empty string text (code -32602 otherwise). _parse_intent uses the injected SPEC-003 IntentEvaluator to extract action, target, title, priority, work type and resource path, and falls back to the coarse heuristic in _parse_natural_language_intent if no evaluator is wired or classification fails. If the evaluator reports that the request needs disambiguation, the reply carries needs_disambiguation and the reasons, and nothing is created. Otherwise, with a kernel client present, the handler creates a thread id, subscribes the caller to pipeline.<thread_id> through _subscribe_pipeline, and sends a CXPPacket with subject intent to the kernel, stamped with the reply consumer id. A dispatch failure is logged and the intent is kept as a task only. The handler then creates a task through the projects handler, keeping a synthetic task from _synthetic_task if that fails, and returns the intent id, task, summary, whether it was dispatched, the thread id and the pipeline topic.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

OrchestratorHandler

Line 40

Intent processing handler.

Functions

Source

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

"""Orchestrator RPC handlers -- intent processing, task creation.

Bridges natural-language user input to Orchestrata's intent system
and creates visible tasks so users can track work.
"""

from __future__ import annotations

import logging
import re
import uuid
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.gateway.exceptions import RPCError

logger = logging.getLogger(__name__)

# Map IntentPriority.name -> the UI/task priority vocabulary. NORMAL collapses
# to "medium" (the task default); BACKGROUND to "low".
_PRIORITY_TO_TASK: dict[str, str] = {
    "CRITICAL": "critical",
    "HIGH": "high",
    "NORMAL": "medium",
    "LOW": "low",
    "BACKGROUND": "low",
}


def _extract_assigned_volk(text: str) -> str | None:
    """Return the ``@volk`` mention in *text*, or ``None``.

    Shared by both the evaluator path and the heuristic parser so the assignee
    is extracted identically regardless of which parser classified the text.
    """
    match = re.search(r"@([A-Za-z_][A-Za-z0-9_:]*)", text)
    return match.group(1) if match else None


class OrchestratorHandler:
    """Intent processing handler.

    Accepts natural-language intents from the UI, parses them,
    and creates tasks so the user sees work in progress.
    """

    def __init__(self) -> None:
        self._projects_handler: Any = None
        self._kernel_client: Any = None
        self._pubsub: Any = None
        self._evaluator: Any = None
        # P-B6 (design D6, mirrors gateway/handlers/cxp.py's P-B3): this
        # gateway's minted reply-consumer id, stamped onto every kernel-bound
        # ``orchestrator.intent`` dispatch's metadata.reply_to so the
        # kernel's per-consumer ReplyRouter (P-B1) addresses the reply back
        # to THIS gateway. None (the default) is the back-compat path --
        # unstamped send -- for callers that never call
        # set_reply_consumer_id.
        self._reply_consumer_id: str | None = None

    def set_projects_handler(self, handler: Any) -> None:
        """Inject ProjectsHandler so we can create tasks."""
        self._projects_handler = handler

    def set_evaluator(self, evaluator: Any) -> None:
        """Inject the SPEC-003 ``IntentEvaluator`` as the single NL parser.

        When wired, free text is classified through the extensible evaluator
        (parse -> classify -> workflow-map -> confidence gate) instead of the
        coarse :func:`_parse_natural_language_intent` heuristic — so the UI
        ingress and the IntentBroker share ONE parser. Truly unclassifiable
        text (no work-type signal) is returned for disambiguation rather than
        spawning pipeline work. When unset, the handler falls back to the
        heuristic (degraded, but functional)."""
        self._evaluator = evaluator

    def set_kernel_client(self, client: Any) -> None:
        """Inject the CXP gRPC client to dispatch intents to the live volk
        kernel. When unset, intents run in task-only (degraded) mode."""
        self._kernel_client = client

    def set_reply_consumer_id(self, consumer_id: str) -> None:
        """Inject this gateway's minted reply-consumer id (design D6).

        Stamped onto every kernel-bound ``orchestrator.intent`` dispatch's
        ``metadata.reply_to`` so the kernel's per-consumer ``ReplyRouter``
        (P-B1) routes the volk's reply back to THIS gateway's private queue
        instead of the single legacy ``"HUMAN"`` queue every gateway used to
        compete over (G3). Mirrors ``CxpHandler.set_reply_consumer_id``
        (P-B3).

        Args:
            consumer_id: This gateway's minted id (``HUMAN@<hex>``).
        """
        self._reply_consumer_id = consumer_id

    def set_pubsub(self, pubsub: Any) -> None:
        """Inject the gateway pub/sub manager.

        Used to subscribe the calling UI connection to ``pipeline.<thread_id>``
        *before* the intent is dispatched, so early pipeline events (emitted by
        the kernel's PipelineEventBus and forwarded by the gateway) aren't lost
        to the subscribe-after-publish race. When unset, dispatch still works;
        the UI just relies on its own ``cxp.subscribe`` instead."""
        self._pubsub = pubsub

    async def handle_intent(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle orchestrator.intent -- process a user intent.

        Args:
            connection_id: Caller connection ID.
            params: Must contain ``text`` (the natural-language intent).

        Returns:
            Dict with ``intent_id``, ``task``, and ``summary``.
        """
        text = params.get("text", "")
        if not text or not isinstance(text, str):
            raise RPCError(
                "Missing 'text' parameter",
                rpc_code=-32602,
            )

        parsed = self._parse_intent(text)
        intent_id = f"intent_{uuid.uuid4().hex[:8]}"

        # Ingress governance gate: when the shared evaluator cannot classify the
        # text into any actionable work type, ask the user to clarify instead of
        # spawning pipeline work on a guess (mirrors IntentBroker.process_text).
        if parsed.get("needs_disambiguation"):
            return {
                "intent_id": intent_id,
                "action": parsed.get("action", "create"),
                "target": parsed.get("target", "task"),
                "summary": "Could not determine what to do — please clarify.",
                "task": None,
                "dispatched": False,
                "thread_id": "",
                "pipeline_topic": "",
                "needs_disambiguation": True,
                "reasons": parsed.get("reasons", []),
            }

        # Dispatch the intent to the live volk kernel when wired. send_packet is
        # ack-only; the volk's reply streams back asynchronously and is delivered
        # to the UI by the gateway's reply consumer. The task below remains a
        # visible side-record. Degrades to task-only if the kernel is unreachable.
        thread_id = ""
        dispatched = False
        if self._kernel_client is not None:
            try:
                from orchestrata_ultimate.core.packet import CXPPacket

                thread_id = uuid.uuid4().hex
                # Subscribe the UI to this thread's pipeline topic BEFORE
                # dispatch so the first stage events (which the kernel emits as
                # soon as the intent lands) aren't lost to the subscribe-after-
                # publish race; the per-thread ring buffer replays any it does.
                await self._subscribe_pipeline(connection_id, thread_id)
                # P-B6 (design D6): stamp this gateway's reply-consumer id so
                # the kernel's ReplyRouter (P-B1/P-B2) addresses the reply
                # back to THIS gateway. Back-compat: when unset (no
                # set_reply_consumer_id call), send unstamped -- byte-
                # identical to pre-P-B6 output -- and the kernel/ReplyRouter
                # fall back to the legacy "HUMAN" key.
                metadata = (
                    {"reply_to": self._reply_consumer_id}
                    if self._reply_consumer_id
                    else None
                )
                await self._kernel_client.send_packet(
                    CXPPacket(
                        from_agent="HUMAN",
                        to_agent="U:Cx",
                        subject="intent",
                        payload={"text": text},
                        thread_id=thread_id,
                        metadata=metadata,
                    )
                )
                dispatched = True
            except Exception:
                thread_id = ""
                logger.warning("Kernel dispatch failed; intent kept as task only")

        # Always present a task to the UI. The synthetic task carries the parsed
        # title/priority/assignee; when the projects handler persists to the graph we
        # adopt its id. When persistence is unavailable (no handler, Neo4j down,
        # or an error result with no task) we keep the synthetic task so the
        # intent flow still works in degraded mode.
        task = _synthetic_task(parsed, text)
        if self._projects_handler:
            try:
                task_result = await self._projects_handler.handle_task_create(
                    connection_id,
                    {
                        "title": parsed["title"],
                        "description": parsed.get("description", text),
                        "status": "not_started",
                        "priority": parsed.get("priority", "medium"),
                        "assigned_volk": parsed.get("assigned_volk"),
                    },
                )
                created = task_result.get("task")
                if created and created.get("id"):
                    task["id"] = created["id"]
            except Exception:
                pass  # keep the synthetic task

        return {
            "intent_id": intent_id,
            "action": parsed.get("action", "create"),
            "target": parsed.get("target", "task"),
            "work_type": parsed.get("work_type", ""),
            "resource_path": parsed.get("resource_path", ""),
            "summary": f"Created task: {parsed['title']}",
            "task": task,
            "dispatched": dispatched,
            "thread_id": thread_id,
            "pipeline_topic": f"pipeline.{thread_id}" if thread_id else "",
        }

    def _parse_intent(self, text: str) -> dict[str, Any]:
        """Parse free text to a structured-intent dict.

        Prefers the injected SPEC-003 ``IntentEvaluator`` (the single, shared,
        extensible NL parser); falls back to the coarse heuristic when no
        evaluator is wired or classification raises. The returned dict is a
        superset of :func:`_parse_natural_language_intent`'s keys, adding
        ``work_type``, ``resource_path``, ``needs_disambiguation`` and
        ``reasons``.
        """
        if self._evaluator is None:
            return _parse_natural_language_intent(text)
        try:
            evaluation = self._evaluator.evaluate(text, requester="U:Cx")
            intent = evaluation.intent
            title = intent.parameters.get("title") or text
            return {
                "action": intent.action.value,
                "target": intent.target.value,
                "title": title,
                "description": text,
                "priority": _PRIORITY_TO_TASK.get(intent.priority.name, "medium"),
                "assigned_volk": _extract_assigned_volk(text),
                "work_type": evaluation.work_type.value,
                "resource_path": intent.resource_path,
                "needs_disambiguation": evaluation.needs_disambiguation,
                "reasons": list(evaluation.reasons),
            }
        except Exception:
            logger.warning("Evaluator classification failed; using heuristic parser")
            return _parse_natural_language_intent(text)

    async def _subscribe_pipeline(self, connection_id: str, thread_id: str) -> None:
        """Subscribe *connection_id* to ``pipeline.<thread_id>`` if pubsub wired.

        Best-effort and non-fatal: a failure to subscribe must not block the
        intent dispatch (the UI can still ``cxp.subscribe`` itself).

        Args:
            connection_id: Caller connection to subscribe.
            thread_id: Pipeline thread whose events the UI should receive.
        """
        if self._pubsub is None or not thread_id:
            return
        try:
            await self._pubsub.subscribe(connection_id, f"pipeline.{thread_id}")
        except Exception:
            logger.warning("Pipeline subscribe failed for thread %s", thread_id)


def _synthetic_task(parsed: dict[str, Any], text: str) -> dict[str, Any]:
    """Build an in-memory task from a parsed intent.

    Used when Neo4j persistence is unavailable so the intent flow still
    returns a usable task to the UI.

    Args:
        parsed: Output of :func:`_parse_natural_language_intent`.
        text: Original natural-language intent.

    Returns:
        Task dict with id, title, description, status, priority, assignee.
    """
    return {
        "id": f"task_{uuid.uuid4().hex[:8]}",
        "title": parsed["title"],
        "description": text,
        "status": "not_started",
        "priority": parsed.get("priority", "medium"),
        "assigned_volk": parsed.get("assigned_volk"),
        "created_at": datetime.now(UTC).isoformat(),
    }


def _parse_natural_language_intent(text: str) -> dict[str, Any]:
    """Heuristic parser: natural language → structured intent.

    Rules (simplest thing that works):
      - Sentences starting with "create", "add", "make", "fix",
        "implement", "write", "build" → action=create
      - Keywords "critical", "urgent", "asap" → priority=critical
      - Keywords "high", "important" → priority=high
      - Keywords "low", "minor", "nice to have" → priority=low
      - Everything else → priority=medium
      - The rest of the sentence becomes the title.
    """
    lower = text.lower()

    # Action detection
    action = "create"
    action_words = ["create", "add", "make", "fix", "implement", "write", "build"]
    for word in action_words:
        if lower.startswith(word):
            action = "create"
            break

    # Priority detection
    priority = "medium"
    if any(w in lower for w in ("critical", "urgent", "asap", "emergency")):
        priority = "critical"
    elif any(w in lower for w in ("high", "important", "priority")):
        priority = "high"
    elif any(w in lower for w in ("low", "minor", "nice to have", "trivial")):
        priority = "low"

    # Clean up title: strip action prefix and extra whitespace
    title = text.strip()
    for prefix in action_words:
        if lower.startswith(prefix):
            # Remove prefix + optional " a " / " an " / " the "
            rest = title[len(prefix) :].strip()
            rest = re.sub(r"^(a|an|the)\s+", "", rest, flags=re.IGNORECASE)
            title = rest
            break

    # Capitalize first letter
    if title:
        title = title[0].upper() + title[1:]

    # Extract assigned volk if mentioned
    assigned_volk: str | None = None
    volk_match = re.search(r"@([A-Za-z_][A-Za-z0-9_:]*)", text)
    if volk_match:
        assigned_volk = volk_match.group(1)
        # Remove @mention from title
        title = re.sub(r"\s*@\S+", "", title).strip()

    return {
        "action": action,
        "target": "task",
        "title": title or text,
        "description": text,
        "priority": priority,
        "assigned_volk": assigned_volk,
    }