orchestrata_ultimate.orchestrator.impact.analyzer

Reference › orchestrata_ultimate › orchestrator › impact › analyzer

IntentImpactAnalyzer is the entry point of impact analysis, called between validation and routing by intent_broker (wired in by intent_broker_factory when impact analysis is enabled) and by the CLI. It composes the three collaborators of the subpackage.

analyze takes the intent text and runs the steps in order: IntentClassifier.classify gives an IntentClassification, _find_impact_scope asks the DependencyAnalyzer for the affected files and builds an ImpactScope with _build_scope, and ParallelPlanner.create_plan makes the plan. The result is an ImpactReport with a UTC timestamp. _calculate_severity grades the scope from counts: critical above 20 files or more than 5 downstream dependents, high above 10 files or more than 2 downstream, medium above 3 files or more than 2 upstream, otherwise low. A dependency analyzer can be injected; close releases its graph client.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentImpactAnalyzer

Decorators: traces(spec='019', story='US-001', reqs=['REQ-019-001']) · Line 27

Main impact analyzer for intents.

Source

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

"""Intent Impact Analyzer - main orchestrator for impact analysis.

Combines classification, dependency analysis, interface verification,
and parallel planning into a single impact analysis pipeline.
"""

import logging
from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.impact.classifier import IntentClassifier
from orchestrata_ultimate.orchestrator.impact.dependency_analyzer import (
    DependencyAnalyzer,
)
from orchestrata_ultimate.orchestrator.impact.planner import ParallelPlanner
from orchestrata_ultimate.orchestrator.impact.types import (
    ImpactReport,
    ImpactScope,
    ImpactSeverity,
    IntentClassification,
)

logger = logging.getLogger(__name__)


@traces(spec="019", story="US-001", reqs=["REQ-019-001"])
class IntentImpactAnalyzer:
    """Main impact analyzer for intents.

    Pipeline:
        1. Classify intent type
        2. Query dependency graph for affected code
        3. Create parallel execution plan
    """

    def __init__(
        self,
        dependency_analyzer: DependencyAnalyzer | None = None,
    ) -> None:
        """Initialize the analyzer with optional dependency analyzer.

        Args:
            dependency_analyzer: Custom DependencyAnalyzer instance.
                Creates a default one if not provided.
        """
        self._classifier = IntentClassifier()
        self._dependency_analyzer = dependency_analyzer or DependencyAnalyzer()
        self._planner = ParallelPlanner()

    async def analyze(self, intent_text: str) -> ImpactReport:
        """Perform full impact analysis on an intent.

        Args:
            intent_text: The human intent text to analyze.

        Returns:
            Complete ImpactReport with classification, scope,
            and parallel execution plan.
        """
        classification = self._classifier.classify(intent_text)
        logger.info("Classified intent as: %s", classification.type.value)

        scope = await self._find_impact_scope(classification)
        logger.info("Found %d affected files", len(scope.affected_files))

        parallel_plan = self._planner.create_plan(classification, scope)
        logger.info("Created plan with %d tasks", parallel_plan.total_tasks)

        return ImpactReport(
            classification=classification,
            scope=scope,
            parallel_plan=parallel_plan,
            analysis_timestamp=datetime.now(UTC).isoformat(),
        )

    async def _find_impact_scope(
        self,
        classification: IntentClassification,
    ) -> ImpactScope:
        """Find the scope of impact based on classification.

        Args:
            classification: The classified intent result.

        Returns:
            ImpactScope describing affected code areas.
        """
        target_package = classification.target_package
        target_module = classification.target_module

        try:
            affected = await self._dependency_analyzer.find_affected_scope(
                target_package=target_package,
                target_module=target_module,
            )
            return self._build_scope(target_package, target_module, affected)
        except (ConnectionError, RuntimeError) as exc:
            logger.error("Dependency analysis failed: %s", exc)

        return ImpactScope(
            target_package=target_package or "unknown",
            target_module=target_module,
            severity=ImpactSeverity.MEDIUM,
        )

    def _build_scope(
        self,
        target_package: str | None,
        target_module: str | None,
        affected: dict,
    ) -> ImpactScope:
        """Build ImpactScope from dependency analysis results.

        Args:
            target_package: The target package name.
            target_module: The target module name.
            affected: Raw dict from dependency analyzer.

        Returns:
            Populated ImpactScope instance.
        """
        packages = affected.get("downstream", [])
        if target_package:
            packages = packages + [target_package]

        return ImpactScope(
            target_package=target_package or "unknown",
            target_module=target_module,
            affected_packages=packages,
            affected_modules=[m["name"] for m in affected.get("affected_modules", [])],
            affected_files=[f["path"] for f in affected.get("affected_files", [])],
            upstream_dependencies=affected.get("upstream", []),
            downstream_dependencies=affected.get("downstream", []),
            test_files=[t["path"] for t in affected.get("tests", [])],
            severity=self._calculate_severity(affected),
        )

    def _calculate_severity(self, affected: dict) -> ImpactSeverity:
        """Calculate impact severity based on affected code.

        Args:
            affected: Dict with affected_files, upstream,
                downstream lists.

        Returns:
            ImpactSeverity enum value.
        """
        file_count = len(affected.get("affected_files", []))
        upstream_count = len(affected.get("upstream", []))
        downstream_count = len(affected.get("downstream", []))

        if file_count > 20 or downstream_count > 5:
            return ImpactSeverity.CRITICAL
        elif file_count > 10 or downstream_count > 2:
            return ImpactSeverity.HIGH
        elif file_count > 3 or upstream_count > 2:
            return ImpactSeverity.MEDIUM
        return ImpactSeverity.LOW

    async def close(self) -> None:
        """Close the analyzer and release resources."""
        await self._dependency_analyzer.close()