orchestrata_ultimate.orchestrator.intent_broker

Reference › orchestrata_ultimate › orchestrator › intent_broker

IntentBroker is the front door of the intent pipeline and the place where a human intent becomes work for the volks. Concierge, the gateway or a tool interceptor hands it a sentence, a tool call or a ready-made Intent, and the broker drives that intent through validation, optional impact analysis, routing and dispatch, recording each state change. It is assembled by intent_broker_factory and implements spec 003; see the orchestrator package narrative for the wider picture.

Dependencies arrive in one IntentBrokerConfig dataclass: an IntentValidator, an IntentRouter and IntentDispatcher, an IntentLifecycleManager, and optional history, metrics, security, impact analyzer and NL evaluator. Three entry points feed the same core:

• process_text runs the IntentEvaluator (normalize, classify, map to a workflow, confidence gate). Ambiguous text, including questions, comes back as a failed DispatchResult with an "ambiguous intent" message rather than being dispatched on a guess. Without an evaluator it fails and does not guess.
• process_tool_call uses IntentExtractor from intent_parser to turn an agent tool call into an intent.
• process_intent is the core. It stores the intent, validates it (a rejection moves it to REJECTED), optionally runs the impact analyzer and attaches a summary to the intent's parameters (failure of the analysis is logged and ignored), routes it, and dispatches it, moving the status to DISPATCHED on success or FAILED otherwise. Any exception becomes a failed DispatchResult; the method does not raise.

An intent classified as a refactor is diverted by _handle_refactor to the in-place refactor agent (SPEC-103), with target volk U:Cd. Its sandbox root comes from the root or workspace parameter, else the resource path, else the ORCHESTRATA_WORKSPACE variable. validate_only, route_only, get_intent, get_metrics and close are inspection helpers; BrokerMetrics counts started, successful, failed and rejected intents and average latency.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentBrokerError

Bases: Exception · Decorators: traces(spec='003', story='US-004', reqs=['REQ-003-US-004-001']) · Line 52

Base exception for IntentBroker errors.

IntentBrokerConfig (dataclass)

Decorators: dataclass · Line 57

Configuration object for IntentBroker dependency injection.

IntentBroker

Decorators: traces(spec='003', story='US-003', reqs=['REQ-003-US-003-001', 'REQ-003-US-009-024']) · Line 85

Main intent processing pipeline.

BrokerMetrics

Decorators: traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 431

Internal metrics tracker for the IntentBroker.

Source

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

# mypy: ignore-errors
"""Intent Broker — core orchestrator for the intent pipeline.

Coordinates parsing, validation, routing, dispatch, and lifecycle
management for intents in the Orchestrata multi-agent system.

The broker is the single entry point for intent processing:
    text/tool_call → parse → validate → route → dispatch → track

Spec: 003-intent
Author: Mohamed
"""

# Spec: 003 - Intent Broker

import logging
import time
from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.impact.analyzer import IntentImpactAnalyzer
from orchestrata_ultimate.orchestrator.impact.types import ImpactReport, IntentType
from orchestrata_ultimate.orchestrator.intent_classifier import IntentEvaluator
from orchestrata_ultimate.orchestrator.intent_history import IntentHistory
from orchestrata_ultimate.orchestrator.intent_lifecycle import IntentLifecycleManager
from orchestrata_ultimate.orchestrator.intent_metrics import IntentMetricsCollector
from orchestrata_ultimate.orchestrator.intent_parser import IntentExtractor
from orchestrata_ultimate.orchestrator.intent_router import (
    IntentDispatcher,
    IntentRouter,
)
from orchestrata_ultimate.orchestrator.intent_security import IntentSecurity
from orchestrata_ultimate.orchestrator.intent_types import (
    DispatchResult,
    Intent,
    IntentAction,
    IntentPriority,
    IntentStatus,
    IntentTarget,
    RoutingResult,
    ValidationResult,
)
from orchestrata_ultimate.orchestrator.intent_validator import IntentValidator

logger = logging.getLogger(__name__)


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
class IntentBrokerError(Exception):
    """Base exception for IntentBroker errors."""


@dataclass
class IntentBrokerConfig:
    """Configuration object for IntentBroker dependency injection.

    Groups all constructor dependencies so IntentBroker.__init__
    stays within the 5-parameter limit (Article V: DI).
    """

    validator: IntentValidator
    router: IntentRouter
    dispatcher: IntentDispatcher
    lifecycle: IntentLifecycleManager
    extractor: IntentExtractor | None = field(default=None)
    history: IntentHistory | None = field(default=None)
    metrics: IntentMetricsCollector | None = field(default=None)
    security: IntentSecurity | None = field(default=None)
    impact_analyzer: IntentImpactAnalyzer | None = field(default=None)
    # SPEC-003 US-001: the NL evaluator turns free natural-language text into a
    # structured Intent (parse -> classify -> workflow-map -> confidence gate)
    # so process_text() can run the full pipeline on plain text. Optional:
    # when unset, only pre-structured Intents (process_intent) are accepted.
    evaluator: IntentEvaluator | None = field(default=None)
    enable_impact_analysis: bool = field(default=True)
    # SPEC-103 FR-009: route REFACTOR intents to the tool-calling refactor
    # capability (in-place edits) instead of greenfield volk scaffolding.
    enable_tool_refactor: bool = field(default=True)


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001", "REQ-003-US-009-024"])
class IntentBroker:
    """Main intent processing pipeline.

    Dependencies are injected via constructor (Article V: DI).
    All methods are async (Article VI: Async-first).

    Pipeline:
        1. Parse: text or tool call → Intent
        2. Validate: Intent → ValidationResult (Prolog-based)
        3. Impact Analysis: Intent → ImpactReport (KG-based)
        4. Route: Intent → target Volk(s)
        5. Dispatch: Intent → CXP packet → Volk
        6. Track: lifecycle state machine

    Fail-closed: any error in validation → DENIED.
    """

    def __init__(self, config: IntentBrokerConfig) -> None:
        self._validator = config.validator
        self._router = config.router
        self._dispatcher = config.dispatcher
        self._lifecycle = config.lifecycle
        self._extractor = config.extractor or IntentExtractor()
        self._metrics = BrokerMetrics()
        self._history = config.history
        self._intent_metrics = config.metrics
        self._security = config.security
        self._impact_analyzer = config.impact_analyzer
        self._evaluator = config.evaluator
        self._enable_impact_analysis = config.enable_impact_analysis
        self._enable_tool_refactor = config.enable_tool_refactor

    @property
    def evaluator(self) -> IntentEvaluator | None:
        """The NL evaluator (parse -> classify -> workflow-map -> confidence
        gate), or ``None`` if this broker was built without NL support.

        Exposed so an ingress (e.g. the gateway ``OrchestratorHandler``) can
        reuse the SAME configured evaluator — including its Prolog DCG-grammar
        tier — as the single NL parser, instead of a parallel heuristic."""
        return self._evaluator

    @traces(
        spec="003",
        story="US-003",
        reqs=["REQ-003-US-003-001", "REQ-003-US-009-024"],
    )
    async def process_intent(self, intent: Intent) -> DispatchResult:
        """Process a fully-formed intent through the pipeline.

        Returns DispatchResult on success or failure.
        """
        start = time.monotonic()
        self._metrics.record_start()

        try:
            # Store initial state
            self._lifecycle.store_intent(intent)

            # Step 1: Validate
            validation = await self._validator.validate(intent)

            if validation.is_valid:
                self._lifecycle.transition(intent, IntentStatus.VALIDATED)
            else:
                self._lifecycle.transition(
                    intent, IntentStatus.REJECTED, reason=validation.reason
                )
                self._metrics.record_rejected()
                return DispatchResult(
                    intent_id=intent.id,
                    packet_id="",
                    target_volk="",
                    success=False,
                    error_message=f"Validation failed: {validation.reason}",
                )

            # Step 2: Impact Analysis (between validate and route)
            impact_report: ImpactReport | None = None
            if self._enable_impact_analysis and self._impact_analyzer:
                try:
                    intent_text = self._build_intent_text(intent)
                    impact_report = await self._impact_analyzer.analyze(intent_text)
                    logger.info(
                        f"Impact analysis complete: {impact_report.classification.type.value} "
                        f"- {impact_report.scope.severity.value} severity, "
                        f"{len(impact_report.scope.affected_files)} affected files"
                    )
                    intent.parameters["_impact_report"] = {
                        "classification": impact_report.classification.type.value,
                        "severity": impact_report.scope.severity.value,
                        "affected_files": impact_report.scope.affected_files,
                        "parallel_tasks": (
                            impact_report.parallel_plan.total_tasks
                            if impact_report.parallel_plan
                            else 0
                        ),
                    }
                except Exception as e:
                    logger.warning(f"Impact analysis failed: {e}")

            # Step 2b: REFACTOR intents dispatch to the SPEC-103 tool-calling
            # refactor capability (in-place edits), not the greenfield route.
            if (
                self._enable_tool_refactor
                and impact_report is not None
                and impact_report.classification.type == IntentType.REFACTOR
            ):
                return await self._handle_refactor(intent)

            # Step 3: Route
            self._lifecycle.transition(intent, IntentStatus.ROUTING)
            routing = await self._router.route(intent)

            if not routing.is_routed:
                self._lifecycle.transition(
                    intent, IntentStatus.FAILED, reason="No route found"
                )
                self._metrics.record_failed()
                return DispatchResult(
                    intent_id=intent.id,
                    packet_id="",
                    target_volk="",
                    success=False,
                    error_message=f"Routing failed: {routing.routing_reason}",
                )

            intent.target_volk = routing.primary_volk

            # RoutingResult invariant (intent_router.route()): every path
            # that populates target_volks also sets primary_volk to its
            # first element; is_routed (checked above) is exactly
            # `len(target_volks) > 0`, so primary_volk is non-None here.
            assert routing.primary_volk is not None

            # Step 3: Dispatch
            result = await self._dispatcher.dispatch(intent, routing.primary_volk)

            if result.success:
                self._lifecycle.transition(intent, IntentStatus.DISPATCHED)
                self._metrics.record_success(time.monotonic() - start)
            else:
                self._lifecycle.transition(
                    intent,
                    IntentStatus.FAILED,
                    reason=result.error_message or "Dispatch failed",
                )
                self._metrics.record_failed()

            return result

        except Exception as e:
            logger.error(f"Intent {intent.id} processing error: {e}")
            try:
                self._lifecycle.transition(intent, IntentStatus.FAILED, reason=str(e))
            except Exception as transition_exc:
                logger.warning(
                    "Failed to transition intent %s to FAILED: %s",
                    intent.id,
                    transition_exc,
                )
            self._metrics.record_failed()
            return DispatchResult(
                intent_id=intent.id,
                packet_id="",
                target_volk="",
                success=False,
                error_message=str(e),
            )

    async def process_tool_call(
        self,
        tool_name: str,
        parameters: dict[str, Any],
        requester: str,
        session_id: str | None = None,
        priority: IntentPriority = IntentPriority.NORMAL,
    ) -> DispatchResult:
        """Process a tool call by extracting and processing the intent."""
        extracted = self._extractor.extract(tool_name, parameters)

        if extracted is None:
            return DispatchResult(
                intent_id="",
                packet_id="",
                target_volk="",
                success=False,
                error_message=f"Cannot extract intent from tool: {tool_name}",
            )

        intent = Intent(
            action=IntentAction(extracted.action.value),
            target=IntentTarget(extracted.resource_type.value),
            resource_path=extracted.resource_path,
            parameters=parameters,
            requester=requester,
            priority=priority,
            session_id=session_id,
        )

        return await self.process_intent(intent)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-045"])
    async def process_text(
        self,
        text: str,
        requester: str = "U:Ox",
        session_id: str | None = None,
        priority: IntentPriority | None = None,
        context: Sequence[str] | None = None,
    ) -> DispatchResult:
        """Process a free natural-language intent end to end.

        Runs the NL evaluator (parse -> classify -> workflow-map ->
        confidence gate) to turn free text into a structured ``Intent``,
        then dispatches it through the full pipeline via
        :meth:`process_intent`. Text whose classification confidence is
        below the evaluator's threshold is rejected with a disambiguation
        message rather than dispatched on a guess. Requires an
        ``IntentEvaluator`` (see ``build_intent_broker``); without one this
        returns a failed ``DispatchResult`` instead of guessing structure.

        ``context`` is the preceding conversation turns, oldest first. It
        feeds the normalization stage, which is what lets a turn whose
        subject is a pronoun ("it hangs again when I drain") classify at all
        — no vocabulary can reach that sentence on its own.
        """
        if self._evaluator is None:
            return DispatchResult(
                intent_id="",
                packet_id="",
                target_volk="",
                success=False,
                error_message=(
                    "natural-language processing unavailable: "
                    "no IntentEvaluator wired"
                ),
            )

        evaluation = await self._evaluator.evaluate_async(
            text, requester=requester, context=context
        )
        if evaluation.needs_disambiguation:
            return DispatchResult(
                intent_id=evaluation.intent.id,
                packet_id="",
                target_volk="",
                success=False,
                error_message=f"ambiguous intent: {'; '.join(evaluation.reasons)}",
            )

        intent = evaluation.intent
        if session_id:
            intent.session_id = session_id
        if priority is not None:
            intent.priority = priority
        return await self.process_intent(intent)

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
    async def validate_only(self, intent: Intent) -> ValidationResult:
        """Validate an intent without routing or dispatching."""
        return await self._validator.validate(intent)

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
    async def route_only(self, intent: Intent) -> RoutingResult:
        """Route an intent without dispatching."""
        return await self._router.route(intent)

    @traces(spec="003", story="US-005", reqs=["REQ-003-US-005-019"])
    def get_intent(self, intent_id: str) -> Intent | None:
        """Retrieve an intent by ID."""
        return self._lifecycle.get_intent(intent_id)

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
    def get_metrics(self) -> dict:
        """Get broker processing metrics."""
        return self._metrics.to_dict()

    def close(self) -> None:
        """Clean up resources."""
        self._lifecycle.close()

    def _build_intent_text(self, intent: Intent) -> str:
        """Build a text representation of an intent for impact analysis."""
        parts = [intent.action.value, intent.target.value]
        if intent.resource_path:
            parts.append(intent.resource_path)
        if intent.parameters:
            params_str = ", ".join(
                f"{k}={v}" for k, v in list(intent.parameters.items())[:3]
            )
            parts.append(params_str)
        return " ".join(parts)

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-025"])
    async def _handle_refactor(self, intent: Intent) -> DispatchResult:
        """Dispatch a REFACTOR intent to the tool-calling refactor capability.

        Runs in-place, tool-driven edits on the target codebase instead of the
        greenfield volk scaffolding path (SPEC-103 FR-009). The sandbox root is
        an explicit ``root``/``workspace`` parameter, else the directory of the
        intent's resource_path, else ``ORCHESTRATA_WORKSPACE``.
        """
        import os
        from pathlib import Path

        from orchestrata_ultimate.refactor.agent import arefactor

        task = self._build_intent_text(intent)
        root = intent.parameters.get("root") or intent.parameters.get("workspace")
        if not root and intent.resource_path:
            p = Path(intent.resource_path)
            root = str(p.parent if p.suffix else p)
        if not root:
            root = os.environ.get("ORCHESTRATA_WORKSPACE", ".")

        self._lifecycle.transition(intent, IntentStatus.ROUTING)
        intent.target_volk = "U:Cd"
        logger.info(
            "REFACTOR intent %s -> tool-calling refactor (root=%s)", intent.id, root
        )
        start = time.monotonic()
        result = await arefactor(task, root)
        success = result.status == "completed"
        if success:
            self._lifecycle.transition(intent, IntentStatus.DISPATCHED)
            self._metrics.record_success(time.monotonic() - start)
        else:
            self._lifecycle.transition(
                intent, IntentStatus.FAILED, reason=f"refactor {result.status}"
            )
            self._metrics.record_failed()
        return DispatchResult(
            intent_id=intent.id,
            packet_id="",
            target_volk="U:Cd",
            success=success,
            error_message=(
                None
                if success
                else f"refactor {result.status}: {result.final_message[:200]}"
            ),
        )


@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class BrokerMetrics:
    """Internal metrics tracker for the IntentBroker."""

    def __init__(self) -> None:
        self.total_processed = 0
        self.total_success = 0
        self.total_failed = 0
        self.total_rejected = 0
        self.total_latency_ms = 0.0
        self._started_at = datetime.now(UTC)

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-012"])
    def record_start(self) -> None:
        self.total_processed += 1

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
    def record_success(self, elapsed_seconds: float) -> None:
        self.total_success += 1
        self.total_latency_ms += elapsed_seconds * 1000

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
    def record_failed(self) -> None:
        self.total_failed += 1

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
    def record_rejected(self) -> None:
        self.total_rejected += 1

    @property
    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
    def avg_latency_ms(self) -> float:
        if self.total_success == 0:
            return 0.0
        return self.total_latency_ms / self.total_success

    @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
    def to_dict(self) -> dict:
        return {
            "total_processed": self.total_processed,
            "total_success": self.total_success,
            "total_failed": self.total_failed,
            "total_rejected": self.total_rejected,
            "avg_latency_ms": round(self.avg_latency_ms, 2),
            "uptime_seconds": (datetime.now(UTC) - self._started_at).total_seconds(),
        }