orchestrata_ultimate.orchestrator.impact.planner

Reference › orchestrata_ultimate › orchestrator › impact › planner

The last step of impact analysis: it turns a classified intent and its scope into a plan that several agents can run in parallel. analyzer calls it with the results of the classifier and the dependency analyzer; the plan types are in types.

ParallelPlanner.create_plan picks a primary agent for the intent type from the AGENT_FOR_TYPE table (default U:Cd), builds branches with _build_branches, and derives checkpoints. The branch builders create implementation tasks (_create_impl_tasks), test tasks (_create_test_tasks) and verification tasks (_create_verify_tasks), and group files into tasks of limited size (_group_files_as_tasks, _group_test_files). _build_checkpoints adds <agent>_complete for the primary agent's branch and tests_complete for the test branch. The plan reports the total task count, an estimated speedup of 0.8 per branch with at most four branches counted, and the first branch's task ids as the critical path.

At a glance

Class diagram

Dependency diagram

API

Classes

ParallelPlanner

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

Creates parallel execution plans for intents.

Source

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

"""Parallel execution planner - creates plans for multiple agents.

Analyzes impact scope and creates execution plans with parallel branches
that can run concurrently across different agent types.
"""

from collections import defaultdict

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.impact.types import (
    AgentTask,
    ImpactScope,
    IntentClassification,
    IntentType,
    ParallelPlan,
    TaskBranch,
)

AGENT_FOR_TYPE = {
    IntentType.FEATURE: "U:Cd",
    IntentType.BUGFIX: "U:Cd",
    IntentType.REFACTOR: "U:Cd",
    IntentType.NEW_REQUIREMENT: "U:Ar",
    IntentType.NEW_SPEC: "U:Ar",
}

AGENT_FOR_TESTS = "U:Qa"
AGENT_FOR_DESIGN = "U:Ar"
AGENT_FOR_VERIFICATION = "U:Vfy"


@traces(spec="019", story="US-005", reqs=["REQ-019-005"])
class ParallelPlanner:
    """Creates parallel execution plans for intents.

    Groups tasks by agent type and creates branches that
    can execute concurrently.
    """

    def __init__(self) -> None:
        """Initialize planner with zero task counter."""
        self._task_counter = 0

    def create_plan(
        self,
        classification: IntentClassification,
        scope: ImpactScope,
    ) -> ParallelPlan:
        """Create a parallel execution plan.

        Args:
            classification: The classified intent.
            scope: The impact scope analysis result.

        Returns:
            ParallelPlan with branches, speedup, and checkpoints.
        """
        primary_agent = AGENT_FOR_TYPE.get(classification.type, "U:Cd")
        branches = self._build_branches(classification, scope, primary_agent)
        checkpoints = self._build_checkpoints(branches, primary_agent)

        total_tasks = sum(len(b.tasks) for b in branches)
        speedup = min(len(branches), 4) * 0.8
        critical_path = [t.task_id for t in branches[0].tasks] if branches else []

        return ParallelPlan(
            total_tasks=total_tasks,
            parallel_branches=branches,
            estimated_speedup=speedup,
            critical_path=critical_path,
            verification_checkpoints=checkpoints,
        )

    def _build_branches(
        self,
        classification: IntentClassification,
        scope: ImpactScope,
        primary_agent: str,
    ) -> list[TaskBranch]:
        """Build all task branches for the plan.

        Args:
            classification: The classified intent.
            scope: The impact scope.
            primary_agent: Primary agent type string.

        Returns:
            List of TaskBranch instances.
        """
        branches: list[TaskBranch] = []

        impl_tasks = self._create_impl_tasks(classification, scope, primary_agent)
        if impl_tasks:
            branches.append(TaskBranch(agent_type=primary_agent, tasks=impl_tasks))

        if scope.affected_files:
            test_tasks = self._create_test_tasks(scope)
            if test_tasks:
                branches.append(
                    TaskBranch(agent_type=AGENT_FOR_TESTS, tasks=test_tasks)
                )

        verify_tasks = self._create_verify_tasks(scope)
        if verify_tasks:
            branches.append(
                TaskBranch(
                    agent_type=AGENT_FOR_VERIFICATION,
                    tasks=verify_tasks,
                )
            )

        return branches

    def _build_checkpoints(
        self,
        branches: list[TaskBranch],
        primary_agent: str,
    ) -> list[str]:
        """Build verification checkpoints from branches.

        Args:
            branches: The task branches.
            primary_agent: Primary agent type string.

        Returns:
            List of checkpoint identifier strings.
        """
        checkpoints = []
        for branch in branches:
            if branch.agent_type == primary_agent:
                checkpoints.append(f"{primary_agent}_complete")
            elif branch.agent_type == AGENT_FOR_TESTS:
                checkpoints.append("tests_complete")
        return checkpoints

    def _create_impl_tasks(
        self,
        classification: IntentClassification,
        scope: ImpactScope,
        agent_type: str,
    ) -> list[AgentTask]:
        """Create implementation tasks grouped by package.

        Args:
            classification: The classified intent.
            scope: The impact scope.
            agent_type: Agent type for these tasks.

        Returns:
            List of AgentTask for implementation work.
        """
        files = scope.affected_files or []

        if not files:
            label = classification.keywords[:3]
            return [
                AgentTask(
                    task_id=self._gen_id(),
                    description=(f"Implement {classification.type.value}: {label}"),
                    target_files=[],
                    agent_type=agent_type,
                    priority=10,
                )
            ]

        return self._group_files_as_tasks(files, agent_type, "Implement changes in")

    def _group_files_as_tasks(
        self,
        files: list[str],
        agent_type: str,
        description_prefix: str,
    ) -> list[AgentTask]:
        """Group files by package and create one task per group.

        Args:
            files: List of file paths.
            agent_type: Agent type for the tasks.
            description_prefix: Prefix for task descriptions.

        Returns:
            List of AgentTask grouped by package.
        """
        by_package: dict[str, list[str]] = defaultdict(list)
        for f in files:
            parts = f.split("/")
            pkg = next(
                (p for p in parts if p.startswith("orchestrata-")),
                "default",
            )
            by_package[pkg].append(f)

        return [
            AgentTask(
                task_id=self._gen_id(),
                description=f"{description_prefix} {pkg}",
                target_files=pkg_files,
                agent_type=agent_type,
                priority=10,
            )
            for pkg, pkg_files in by_package.items()
        ]

    def _create_test_tasks(self, scope: ImpactScope) -> list[AgentTask]:
        """Create test tasks from scope.

        Args:
            scope: The impact scope with test files.

        Returns:
            List of AgentTask for test work.
        """
        test_files = scope.test_files or []

        if not test_files:
            return [
                AgentTask(
                    task_id=self._gen_id(),
                    description="Write/update tests for affected code",
                    target_files=[],
                    agent_type=AGENT_FOR_TESTS,
                    priority=5,
                    dependencies=[],
                )
            ]

        return self._group_test_files(test_files)

    def _group_test_files(self, test_files: list[str]) -> list[AgentTask]:
        """Group test files by target source and create tasks.

        Args:
            test_files: List of test file paths.

        Returns:
            List of AgentTask for each test group.
        """
        by_target: dict[str, list[str]] = defaultdict(list)
        for tf in test_files:
            test_name = tf.replace("test_", "").replace("_test.py", ".py")
            by_target[test_name].append(tf)

        return [
            AgentTask(
                task_id=self._gen_id(),
                description=f"Update tests for {target}",
                target_files=tests,
                agent_type=AGENT_FOR_TESTS,
                priority=5,
                dependencies=[],
            )
            for target, tests in by_target.items()
        ]

    def _create_verify_tasks(self, scope: ImpactScope) -> list[AgentTask]:
        """Create verification tasks for affected packages.

        Args:
            scope: The impact scope.

        Returns:
            List of AgentTask for verification.
        """
        if not scope.affected_packages:
            return []

        pkgs = ", ".join(scope.affected_packages[:3])
        return [
            AgentTask(
                task_id=self._gen_id(),
                description=f"Verify changes in {pkgs}",
                target_files=(
                    scope.affected_files[:10] if scope.affected_files else []
                ),
                agent_type=AGENT_FOR_VERIFICATION,
                priority=1,
                dependencies=[],
            )
        ]

    def _gen_id(self) -> str:
        """Generate a unique sequential task ID.

        Returns:
            String like 'task-001', 'task-002', etc.
        """
        self._task_counter += 1
        return f"task-{self._task_counter:03d}"