Nexus

"""
Nexus: Multi-Tool Execution Engine with Topological DAG Scheduling
Detects cyclic tool dependencies and executes concurrent tool branches safely.
"""
import asyncio
from typing import Dict, List, Set, Any
from pydantic import BaseModel, Field

class ToolNode(BaseModel):
    tool_id: str
    dependencies: List[str] = Field(default_factory=list)
    parameters: Dict[str, Any]

class ExecutionPlan(BaseModel):
    is_acyclic: bool
    execution_order: List[List[str]]
    error: str = ""

class NexusDAGDispatcher:
    def __init__(self, max_concurrency: int = 8):
        self.max_concurrency = max_concurrency

    def build_topological_batches(self, nodes: Dict[str, ToolNode]) -> ExecutionPlan:
        # In-degree calculation
        in_degree = {nid: len(n.dependencies) for nid, n in nodes.items()}
        adj = {nid: [] for nid in nodes}
        for nid, n in nodes.items():
            for dep in n.dependencies:
                if dep in adj:
                    adj[dep].append(nid)

        queue = [nid for nid, deg in in_degree.items() if deg == 0]
        batches: List[List[str]] = []
        visited_count = 0

        while queue:
            batches.append(list(queue))
            visited_count += len(queue)
            next_queue = []
            for curr in queue:
                for neighbor in adj[curr]:
                    in_degree[neighbor] -= 1
                    if in_degree[neighbor] == 0:
                        next_queue.append(neighbor)
            queue = next_queue

        if visited_count != len(nodes):
            return ExecutionPlan(is_acyclic=False, execution_order=[], error="CYCLIC_DEPENDENCY_DETECTED")

        return ExecutionPlan(is_acyclic=True, execution_order=batches)