universal-agent-nexus 3.2.0
The Translation Layer: Compile Universal Agents to AWS, LangGraph, MCP, or UAA Kernel.
Requires Python: >=3.11

<div align="center">

# **Universal Agent Nexus**

### *The Translation Layer: Build Once, Run Anywhere*

**Compile universal agent architectures to LangGraph, AWS Step Functions, MCP, or the UAA Kernel—without rewriting code.**

[![PyPI version](https://img.shields.io/pypi/v/universal-agent-nexus.svg)](https://pypi.org/project/universal-agent-nexus/)
[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![Tests](https://github.com/mjdevaccount/universal_agent_nexus/actions/workflows/test.yml/badge.svg)](https://github.com/mjdevaccount/universal_agent_nexus/actions)
[![Code style: ruff](https://img.shields.io/badge/code%20style-ruff-000000.svg)](https://github.com/astral-sh/ruff)

[**Quick Start**](#-complete-beginner-tutorial) • [**Schema Reference**](#-manifest-schema-reference) • [**State Bridge**](#-cross-runtime-state-bridge) • [**Batch API**](#-batch-api--prompt-caching) • [**Extending Nexus**](#-extending-nexus-platform-extension-api) • [**Examples**](#-examples) • [**Troubleshooting**](#-troubleshooting) • [**Report Bug**](https://github.com/mjdevaccount/universal_agent_nexus/issues)

</div>

---

## 🎯 **What is Universal Agent Nexus?**

### **The Problem You're Solving**

You built an AI agent. It works great locally in LangGraph. Now you need to:

- **Deploy to production at scale** → AWS Step Functions
- **Expose to Claude Desktop for teammates** → MCP
- **Run through the UAA Kernel** → Native execution
- **Keep local dev fast for iteration** → LangGraph

**Traditional approach:** Rewrite your agent three times. Maintain three codebases. Deal with drift.

**Universal Agent Nexus approach:** Write once. Compile to any runtime.

```yaml
# One manifest
name: my-agent
graphs: [...]
```

```bash
nexus compile my-agent.yaml --target langgraph  # Local dev
nexus compile my-agent.yaml --target aws        # Production  
nexus compile my-agent.yaml --target mcp        # AI clients
nexus compile my-agent.yaml --target uaa        # UAA Kernel (native)
```

**Same logic. Four runtimes. Zero rewrites.**

---

## 🔗 **How Nexus + Fabric + Architecture Work Together**

> **New to Universal Agent?** This section explains the ecosystem.

The Universal Agent ecosystem consists of three interconnected repositories:

| Repository | Role | Metaphor |
|------------|------|----------|
| **[Architecture](https://github.com/mjdevaccount/universal_agent_architecture)** | Runtime kernel, state management, durability | Linux Kernel |
| **[Fabric](https://github.com/mjdevaccount/universal_agent_fabric)** | Roles, domains, policies → compiled manifests | Userland / distro |
| **[Nexus](https://github.com/mjdevaccount/universal_agent_nexus)** (this repo) | Adapters for AWS, LangGraph, MCP, UAA Kernel | Network / drivers |

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                         FABRIC (Composition Layer)                           │
│   github.com/mjdevaccount/universal_agent_fabric                             │
│                                                                              │
│   Role + Domain + Policy YAMLs → FabricBuilder → Enriched Manifest          │
└────────────────────────────────────┬────────────────────────────────────────┘
                                     │
                                     ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         NEXUS (Compiler Layer)                               │
│   github.com/mjdevaccount/universal_agent_nexus (this repo)                  │
│                                                                              │
│   ┌─────────────────────────────────────────────────────────────────────┐   │
│   │                        ManifestIR                                    │   │
│   └────────────────────────────┬────────────────────────────────────────┘   │
│                                │ Generators                                  │
│   ┌────────────┬───────────────┼───────────────┬────────────────────────┐   │
│   │ LangGraph  │      AWS      │      MCP      │     UAA Native         │   │
│   └────────────┴───────────────┴───────────────┴────────────────────────┘   │
│                                                                              │
│   + State Normalization Bridge (LangGraph ↔ AWS ↔ UAA)                       │
└────────────────────────────────────┬────────────────────────────────────────┘
                                     │
                                     ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                       ARCHITECTURE (Kernel Layer)                            │
│   github.com/mjdevaccount/universal_agent_architecture                       │
│                                                                              │
│   GraphEngine + Handlers + PolicyEngine + TaskStore + ContractRegistry      │
└─────────────────────────────────────────────────────────────────────────────┘
```

### **The Workflow**

```bash
# Step 1: Use Fabric to compose your agent
fabric build \
  --role researcher.yaml \
  --domain finance.yaml \
  --policy safety.yaml \
  --out my_agent.yaml

# Step 2: Use Nexus to compile to runtime
nexus compile my_agent.yaml --target langgraph --output agent.py

# Step 3: Run
python agent.py
```

### **Why Separate Repos?**

| Repo | Owner | Purpose |
|------|-------|---------|
| **Fabric** | Security/Business teams | Roles, capabilities, governance policies |
| **Nexus** | Platform/DevOps teams | Runtime translation, deployment pipelines |

This separation allows security teams to own policies while platform teams own infrastructure—without stepping on each other.

### **Can I Use Nexus Without Fabric?**

**Absolutely!** You can write manifest YAML files directly (see [Manifest Schema Reference](#-manifest-schema-reference)). Fabric just makes it easier for complex, enterprise scenarios.

---

## 🚀 **Complete Beginner Tutorial**

### **From Zero to Running Agent (5 minutes)**

**Goal:** Build a simple chatbot that greets users by name.

#### **Step 1: Install**

```bash
pip install "universal-agent-nexus[langgraph]"
```

#### **Step 2: Create manifest**

Create `chatbot.yaml`:

```yaml
name: greeter-bot
version: "1.0.0"
description: "A friendly greeting agent"

graphs:
  - name: main
    entry_node: greet
    nodes:
      - id: greet
        kind: task
        label: "Generate Greeting"
        config:
          prompt: "Generate a friendly greeting for {name}"
    edges: []

tools: []
routers: []
policies: []
```

#### **Step 3: Compile to LangGraph**

```bash
nexus compile chatbot.yaml --target langgraph --output bot.py
```

This generates `bot.py` with:
- ✅ LangGraph StateGraph definition
- ✅ Postgres checkpointing
- ✅ Async execution

#### **Step 4: Run locally**

```bash
# Start Postgres (Docker)
docker run -d -p 5432:5432 \
  -e POSTGRES_PASSWORD=password \
  -e POSTGRES_DB=nexus_dev \
  postgres:16-alpine

# Set environment
export UAA_POSTGRES_URL=postgresql://postgres:password@localhost:5432/nexus_dev

# Run
python bot.py
```

**Expected output:**

```
✅ Connected to PostgreSQL
✅ Graph initialized: main
Input: {"name": "Alice"}
Output: "Hello Alice! Welcome to Universal Agent Nexus! 👋"
```

#### **Step 5: Deploy to AWS (Optional)**

```bash
# Compile to AWS
nexus compile chatbot.yaml --target aws --output state_machine.json

# Deploy (requires AWS CLI configured)
cd terraform/environments/dev
terraform init
terraform apply
```

**✅ You just built and deployed a multi-runtime agent!**

---

## 📝 **Manifest Schema Reference**

### **Complete Example**

```yaml
name: "my-agent"
version: "1.0.0"
description: "Agent description"

graphs:
  - name: "main"
    entry_node: "start"
    nodes:
      - id: "start"
        kind: router              # Decision point
        label: "Risk Classifier"
        config:
          llm: "gpt-4o-mini"
          system_message: "You are a risk classifier..."
          
      - id: "validate"
        kind: tool                # External capability
        label: "Policy Validator"
        tool_ref: "policy_check"
        
      - id: "escalate"
        kind: task                # LLM generation
        label: "Generate Summary"
        config:
          prompt: "Summarize incident: {input}"
    
    edges:
      - from_node: "start"
        to_node: "validate"
        condition:
          expression: "risk_level > 0.5"
          
      - from_node: "validate"
        to_node: "escalate"
        condition:
          expression: "policy_violated == true"

tools:
  - name: "policy_check"
    protocol: "mcp"
    config:
      command: "mcp-policy-server"
      args: ["--strict"]

routers:
  - name: "risk_classifier"
    llm_provider: "openai"
    model: "gpt-4o-mini"
    system_message: "Classify risk as low/medium/high"

policies:
  - name: "require_approval"
    target_pattern: "execute_*"
    action: "require_approval"
    conditions:
      risk_level: ">0.8"
```

### **Node Types**

| Kind | Purpose | When to Use | Config Fields |
|------|---------|-------------|---------------|
| `router` | LLM makes a decision | Multi-path routing, classification | `llm`, `system_message`, `tools` |
| `tool` | Call external API/MCP | Database query, API call, file I/O | `tool_ref`, `input_mapping` |
| `task` | LLM generates text | Summarization, translation, writing | `prompt`, `llm`, `max_tokens` |

### **Edge Conditions**

Conditions use simple expression syntax:

```yaml
# Simple comparison
condition:
  expression: "risk_level > 0.5"

# Multiple conditions (AND)
condition:
  expression: "risk_level > 0.5 AND confidence > 0.8"

# Object field access
condition:
  expression: "user.role == 'admin'"

# List operations
condition:
  expression: "len(violations) > 0"
```

### **Tool Protocols**

| Protocol | Use Case | Example |
|----------|----------|---------|
| `mcp` | Model Context Protocol servers | Claude Desktop tools, Cursor plugins |
| `http` | REST APIs | External services, webhooks |
| `local` | Python functions | In-process tools |
| `subprocess` | CLI commands | Shell scripts, binaries |

### **Required vs Optional Fields**

| Field | Required | Default | Description |
|-------|----------|---------|-------------|
| `name` | ✅ | - | Unique agent identifier |
| `version` | ✅ | - | Semantic version (e.g., "1.0.0") |
| `description` | ❌ | `""` | Human-readable description |
| `graphs` | ✅ | - | Array of graph definitions |
| `graphs[].name` | ✅ | - | Graph identifier |
| `graphs[].entry_node` | ✅ | - | ID of starting node |
| `graphs[].nodes` | ✅ | - | Array of node definitions |
| `graphs[].edges` | ❌ | `[]` | Array of edge definitions |
| `tools` | ❌ | `[]` | External tool definitions |
| `routers` | ❌ | `[]` | Router configurations |
| `policies` | ❌ | `[]` | Governance policies |

---

## 📚 **Core Concepts**

### **1. Universal Agent Architecture (UAA)**

The **Universal Agent Architecture** is a specification for defining multi-step agent workflows as portable, runtime-agnostic manifests.

**Key components:**

- **Graph:** A directed graph of nodes (tasks, routers, tools) connected by edges
- **Node:** A single execution unit (LLM call, tool invocation, conditional logic)
- **Edge:** Transition between nodes (with optional conditions)
- **Tool:** External capability (API, database, MCP server)
- **Router:** Decision point that chooses next action(s) dynamically

### **2. Compilation Targets (Adapters)**

| Target | Best For | Execution Model | State Storage |
|--------|----------|-----------------|---------------|
| **LangGraph** | Local dev, debugging | Python async | PostgreSQL |
| **AWS** | Production scale | Step Functions, Lambda | DynamoDB |
| **MCP** | AI client integration | stdio transport | In-memory |

### **3. The IR-Based Compiler (v1.0.0)**

#### **Why an Intermediate Representation?**

Without IR, we'd need N² translators (LangGraph↔AWS, AWS↔MCP, etc.). With IR, we need only 2N (N parsers + N generators).

**Benefits:**

- ✅ **Bidirectional translation:** LangGraph ⇄ AWS ⇄ MCP
- ✅ **Optimization passes:** Dead code elimination, edge deduplication
- ✅ **Validation:** Type checking before code generation
- ✅ **Future-proof:** Add new runtimes without changing existing ones

#### **How it works:**

```
┌────────────────────────────────────────────────────────────┐
│                     FRONTENDS (Parsers)                    │
│  LangGraph → IR  |  AWS → IR  |  YAML → IR                │
└──────────────────────────┬─────────────────────────────────┘
                           │
                           ▼
┌────────────────────────────────────────────────────────────┐
│                  INTERMEDIATE REPRESENTATION               │
│            ManifestIR { GraphIR, ToolIR, RouterIR }        │
└──────────────────────────┬─────────────────────────────────┘
                           │
                           ▼
┌────────────────────────────────────────────────────────────┐
│                   TRANSFORMATION PASSES                    │
│  Dead code elimination | Edge deduplication | Validation   │
└──────────────────────────┬─────────────────────────────────┘
                           │
                           ▼
┌────────────────────────────────────────────────────────────┐
│                    BACKENDS (Generators)                   │
│  IR → LangGraph  |  IR → AWS  |  IR → YAML                │
└────────────────────────────────────────────────────────────┘
```

#### **Bidirectional translation in action:**

```bash
# AWS → LangGraph (reverse compile!)
nexus translate state_machine.json --to langgraph --out agent.py

# LangGraph → AWS
nexus translate agent.py --to aws --out state_machine.json

# Chain translations: AWS → LangGraph → MCP
nexus translate state_machine.json --to langgraph --out temp.py
nexus translate temp.py --to mcp --out server.py
```

---

## 🔧 **Extending Nexus: Platform Extension API**

> **For Platform Engineers & Framework Developers**
>
> Nexus exposes a comprehensive extension API that enables you to customize every aspect of the compilation pipeline. This section documents all extension points, APIs, and patterns for building on top of Nexus.

### **Extension Points Overview**

Nexus provides **8 tiers of abstraction exposure**, each enabling different levels of customization:

| Tier | Extension Point | Use Case | Complexity |
|------|----------------|----------|------------|
| **1** | Parser Registry | Add support for new source formats (Airflow, Prefect, Temporal) | Low |
| **2** | Generator Registry | Add support for new target runtimes (K8s, GCP, Azure) | Low |
| **3** | Pass Interface | Create custom optimization/validation passes | Medium |
| **4** | Annotations | Add typed metadata to IR nodes/edges | Low |
| **5** | Factories | Customize tool/router instantiation logic | Medium |
| **6** | Enrichment Strategy | Customize Fabric fragment composition | Medium |
| **7** | Compiler Builder | Configure entire compiler with custom components | High |
| **8** | IR Visitor | Traverse and analyze IR structure | Low |

---

### **Tier 1: Parser Registry**

**Purpose:** Register custom parsers to support new source formats.

**API Reference:**

```python
from universal_agent_nexus.parser_registry import (
    register_parser,
    get_parser,
    detect_source_type,
    list_parsers,
    ParserRegistry,
)

# Register a custom parser
register_parser(
    source_type: str,
    parser: Parser,
    *,
    detection_priority: int = 100,
    aliases: Optional[Set[str]] = None,
    description: Optional[str] = None,
) -> None

# Get parser by source type
parser = get_parser(source_type: str) -> Parser

# Auto-detect source type
detected = detect_source_type(source: str) -> str

# List all registered parsers
parsers = list_parsers() -> Dict[str, ParserInfo]
```

**Parser Protocol:**

All parsers must implement the `Parser` protocol:

```python
from universal_agent_nexus.ir.parser import Parser
from universal_agent_nexus.ir import ManifestIR

class MyFrameworkParser:
    """Custom parser for MyFramework."""
    
    def parse(self, source: str) -> ManifestIR:
        """
        Parse source and produce IR.
        
        Args:
            source: Path to source file or source string
            
        Returns:
            ManifestIR instance
        """
        # Parse your format → ManifestIR
        return ManifestIR(...)
    
    def can_parse(self, source: str) -> bool:
        """
        Check if this parser can handle the source.
        
        Args:
            source: Path to source file or source string
            
        Returns:
            True if parser can handle this source
        """
        # Detection logic (file extension, content sniffing, etc.)
        return source.endswith(".myfw")
```

**Complete Example: Airflow Parser**

```python
from universal_agent_nexus.parser_registry import register_parser
from universal_agent_nexus.ir import ManifestIR, GraphIR, NodeIR, NodeKind
from pathlib import Path
import yaml

class AirflowParser:
    """Parse Apache Airflow DAGs to ManifestIR."""
    
    def parse(self, source: str) -> ManifestIR:
        path = Path(source)
        if path.exists():
            dag_def = yaml.safe_load(path.read_text())
        else:
            dag_def = yaml.safe_load(source)
        
        # Convert Airflow DAG → ManifestIR
        nodes = [
            NodeIR(
                id=task_id,
                kind=NodeKind.TASK,
                label=task.get("task_name", task_id),
                description=task.get("description"),
            )
            for task_id, task in dag_def.get("tasks", {}).items()
        ]
        
        graph = GraphIR(
            name=dag_def.get("dag_id", "main"),
            entry_node=nodes[0].id if nodes else "start",
            nodes=nodes,
            edges=[],  # Convert Airflow dependencies to edges
        )
        
        return ManifestIR(
            name=dag_def.get("dag_id", "airflow-dag"),
            version="1.0.0",
            description=dag_def.get("description", ""),
            graphs=[graph],
        )
    
    def can_parse(self, source: str) -> bool:
        path = Path(source)
        if path.exists():
            return source.endswith((".airflow", ".dag.yaml"))
        try:
            data = yaml.safe_load(source)
            return "dag_id" in data and "tasks" in data
        except Exception:
            return False

# Register the parser
register_parser(
    "airflow",
    AirflowParser(),
    detection_priority=150,  # Higher = checked first
    aliases={"dag", "airflow_dag"},
    description="Apache Airflow DAG parser",
)

# Now compile works with Airflow!
from universal_agent_nexus.compiler import compile
result = compile("my_dag.airflow", target="aws")
```

**Best Practices:**

- **Detection Priority:** Set higher priority (200+) for formats with unique file extensions, lower (50-100) for ambiguous formats
- **Aliases:** Register common aliases (e.g., `{"uaa", "yaml"}` for YAML parser)
- **Error Handling:** Provide clear error messages with source location information
- **Performance:** Cache parsed results if parsing is expensive

**Available Parsers:**

```python
# List all registered parsers
parsers = list_parsers()
for source_type, info in parsers.items():
    print(f"{source_type}: {info.description}")
    print(f"  Aliases: {info.aliases}")
    print(f"  Priority: {info.detection_priority}")
```

---

### **Tier 2: Generator Registry**

**Purpose:** Register custom generators to support new target runtimes.

**API Reference:**

```python
from universal_agent_nexus.generator_registry import (
    register_generator,
    get_generator,
    list_generators,
    GeneratorRegistry,
)

# Register a custom generator
register_generator(
    target_type: str,
    generator: Generator,
    *,
    aliases: Optional[Set[str]] = None,
    default_options: Optional[Dict] = None,
    description: Optional[str] = None,
) -> None

# Get generator by target type
generator = get_generator(target_type: str) -> Generator

# List all registered generators
generators = list_generators() -> Dict[str, GeneratorInfo]
```

**Generator Protocol:**

All generators must implement the `Generator` protocol:

```python
from universal_agent_nexus.ir.generator import Generator
from universal_agent_nexus.ir import ManifestIR

class MyPlatformGenerator:
    """Custom generator for MyPlatform."""
    
    def generate(self, ir: ManifestIR) -> str:
        """
        Generate target format from IR.
        
        Args:
            ir: ManifestIR instance
            
        Returns:
            Generated code/config as string
        """
        # Convert ManifestIR → your platform's format
        lines = []
        for graph in ir.graphs:
            lines.append(f"# Graph: {graph.name}")
            # ... generate code ...
        return "\n".join(lines)
```

**Complete Example: Temporal Generator**

```python
from universal_agent_nexus.generator_registry import register_generator
from universal_agent_nexus.ir import ManifestIR, GraphIR, NodeIR, NodeKind

class TemporalGenerator:
    """Generate Temporal workflows from ManifestIR."""
    
    def generate(self, ir: ManifestIR) -> str:
        """Generate Temporal workflow code."""
        lines = [
            '"""',
            f"Auto-generated Temporal workflow: {ir.name}",
            '"""',
            "",
            "from temporalio import workflow",
            "",
        ]
        
        for graph in ir.graphs:
            lines.extend([
                f"@workflow.defn",
                f"class {graph.name.title()}Workflow:",
                f'    """{graph.name} workflow."""',
                "",
                "    @workflow.run",
                "    async def run(self, input_data: dict) -> dict:",
            ])
            
            # Generate workflow steps
            for node in graph.nodes:
                if node.kind == NodeKind.TASK:
                    lines.append(f"        result = await {node.id}_activity(input_data)")
                elif node.kind == NodeKind.TOOL:
                    lines.append(f"        result = await execute_tool('{node.tool_ref}', input_data)")
            
            lines.append("        return result")
            lines.append("")
        
        return "\n".join(lines)

# Register the generator
register_generator(
    "temporal",
    TemporalGenerator(),
    aliases={"temporalio"},
    description="Temporal.io workflow generator",
)

# Now compile works with Temporal!
from universal_agent_nexus.compiler import compile
result = compile("agent.yaml", target="temporal")
```

**Best Practices:**

- **Idempotency:** Generated code should be deterministic (same IR → same output)
- **Readability:** Include comments and clear structure in generated code
- **Error Handling:** Validate IR before generation, provide helpful error messages
- **Options:** Use `default_options` for generator-specific configuration

---

### **Tier 3: Pass Interface**

**Purpose:** Create custom transformation passes for optimization, validation, or instrumentation.

**API Reference:**

```python
from universal_agent_nexus.ir.pass_interface import (
    Transform,
    ValidationPass,
    OptimizationPass,
    InstrumentationPass,
    PassMetadata,
)
from universal_agent_nexus.ir import ManifestIR
from universal_agent_nexus.ir.pass_manager import PassManager

# Base interface
class Transform(ABC):
    @property
    @abstractmethod
    def metadata(self) -> PassMetadata: ...
    
    @abstractmethod
    def apply(self, ir: ManifestIR) -> ManifestIR: ...

# Specialized base classes
class ValidationPass(Transform):  # Don't modify IR
    @abstractmethod
    def validate(self, ir: ManifestIR) -> list[str]: ...

class OptimizationPass(Transform):  # Modify IR for performance
    ...

class InstrumentationPass(Transform):  # Add metadata/observability
    ...
```

**Pass Metadata:**

```python
from dataclasses import dataclass
from typing import Set

@dataclass
class PassMetadata:
    name: str                          # Unique pass identifier
    description: str                   # Human-readable description
    requires: Set[str] = set()         # Passes that must run before this
    invalidates: Set[str] = set()      # Analyses invalidated by this pass
    preserves: Set[str] = set()        # Analyses preserved by this pass
    category: str = "optimization"     # optimization, validation, instrumentation
```

**Complete Example: Custom Optimization Pass**

```python
from universal_agent_nexus.ir.pass_interface import OptimizationPass, PassMetadata
from universal_agent_nexus.ir import ManifestIR, GraphIR, NodeIR, NodeKind

class MergeDuplicateTools(OptimizationPass):
    """Merge duplicate tool nodes with identical configurations."""
    
    @property
    def metadata(self) -> PassMetadata:
        return PassMetadata(
            name="merge-duplicate-tools",
            description="Merge duplicate tool nodes to reduce redundancy",
            requires={"dead-node-elimination"},  # Run after dead code removal
            invalidates={"edge-analysis"},       # May change edges
            preserves={"cycle-free"},            # Preserves cycle-freedom
            category="optimization",
        )
    
    def apply(self, ir: ManifestIR) -> ManifestIR:
        """Merge duplicate tools."""
        for graph in ir.graphs:
            # Group nodes by tool signature
            tool_signatures = {}
            nodes_to_remove = []
            
            for node in graph.nodes:
                if node.kind == NodeKind.TOOL and node.tool_ref:
                    sig = (node.tool_ref, str(sorted(node.config.items())))
                    
                    if sig in tool_signatures:
                        # Found duplicate - redirect edges to original
                        original = tool_signatures[sig]
                        self._redirect_edges(graph, node.id, original.id)
                        nodes_to_remove.append(node.id)
                    else:
                        tool_signatures[sig] = node
            
            # Remove merged nodes
            if nodes_to_remove:
                graph.nodes = [n for n in graph.nodes if n.id not in nodes_to_remove]
                graph._build_indexes()
        
        return ir
    
    def _redirect_edges(self, graph: GraphIR, from_id: str, to_id: str) -> None:
        """Redirect all edges from one node to another."""
        for edge in graph.edges:
            if edge.from_node == from_id:
                edge.from_node = to_id
            if edge.to_node == from_id:
                edge.to_node = to_id

# Use the pass
from universal_agent_nexus.ir.pass_manager import PassManager, OptimizationLevel

manager = PassManager(opt_level=OptimizationLevel.DEFAULT)
manager.add_custom_pass(
    MergeDuplicateTools(),
    after="dead-node-elimination",  # Explicit ordering
)
ir = manager.run(ir)
```

**Complete Example: Validation Pass**

```python
from universal_agent_nexus.ir.pass_interface import ValidationPass, PassMetadata

class CostLimitValidation(ValidationPass):
    """Validate that estimated costs don't exceed limits."""
    
    @property
    def metadata(self) -> PassMetadata:
        return PassMetadata(
            name="cost-limit-validation",
            description="Validate estimated execution costs",
            requires={"dead-node-elimination"},
            category="validation",
        )
    
    def validate(self, ir: ManifestIR) -> list[str]:
        """Return list of error messages (empty if valid)."""
        errors = []
        max_cost_cents = 1000  # $10.00 limit
        
        from universal_agent_nexus.ir.visitor import DefaultIRVisitor, traverse
        
        class CostCollector(DefaultIRVisitor):
            def __init__(self):
                self.total_cost = 0.0
            
            def visit_node(self, node):
                from universal_agent_nexus.ir.annotations import CostAnnotation
                cost = node.metadata.get_annotation(CostAnnotation)
                if cost:
                    self.total_cost += cost.cost_cents
        
        collector = CostCollector()
        traverse(ir, collector)
        
        if collector.total_cost > max_cost_cents:
            errors.append(
                f"Estimated cost ${collector.total_cost/100:.2f} exceeds "
                f"limit ${max_cost_cents/100:.2f}"
            )
        
        return errors

# Use validation pass
manager = PassManager()
manager.add_custom_pass(CostLimitValidation())
try:
    ir = manager.run(ir)
except RuntimeError as e:
    print(f"Validation failed: {e}")
```

**Pass Ordering:**

Passes are automatically ordered based on `requires` dependencies. You can also specify explicit ordering:

```python
manager.add_custom_pass(
    MyPass(),
    before="some-other-pass",  # Run before this pass
    after="dead-node-elimination",  # Run after this pass
)
```

**Available Passes:**

```python
from universal_agent_nexus.ir.pass_manager import PassManager

manager = PassManager()
passes = manager.list_passes()
for name, metadata in passes.items():
    print(f"{name}: {metadata.description}")
    print(f"  Requires: {metadata.requires}")
    print(f"  Category: {metadata.category}")
```

---

### **Tier 4: Annotations**

**Purpose:** Add typed metadata to IR nodes and edges for analysis, optimization, or observability.

**API Reference:**

```python
from universal_agent_nexus.ir.annotations import (
    Annotation,
    CostAnnotation,
    MonitoringAnnotation,
    PerformanceAnnotation,
    BatchAnnotation,
)
from universal_agent_nexus.ir import NodeIR, Metadata

# Annotate a node
node.metadata.annotate(CostAnnotation(tokens=500, latency_ms=250, cost_cents=0.05))

# Get annotation by type
cost = node.metadata.get_annotation(CostAnnotation) -> Optional[CostAnnotation]

# Check if annotation exists
has_cost = node.metadata.has_annotation(CostAnnotation) -> bool

# List all annotations
annotations = node.metadata.list_annotations() -> Dict[str, Annotation]
```

**Built-in Annotations:**

```python
# Cost tracking
cost = CostAnnotation(
    tokens=500,           # Estimated token count
    latency_ms=250,       # Estimated latency in milliseconds
    cost_cents=0.05,      # Estimated cost in cents
)
node.metadata.annotate(cost)

# Monitoring configuration
monitoring = MonitoringAnnotation(
    trace=True,           # Enable distributed tracing
    log_level="INFO",     # Logging level
)
node.metadata.annotate(monitoring)

# Performance characteristics
performance = PerformanceAnnotation(
    expected_duration_ms=100.0,  # Expected execution time
    max_concurrency=5,           # Max concurrent executions
    timeout_ms=5000,              # Timeout in milliseconds
)
node.metadata.annotate(performance)

# Batch API eligibility (added by BatchOptimizationPass)
batch = BatchAnnotation(
    eligible=True,              # Whether node can be batched
    batch_group="claude:abc123", # Group for batching similar requests
    cache_key="sha256...",      # Cache key for prompt caching
    priority=100,               # Higher = process sooner
    max_wait_ms=5000,           # Max wait for batch accumulation
)
node.metadata.annotate(batch)
```

**Creating Custom Annotations:**

```python
from universal_agent_nexus.ir.annotations import Annotation

class SecurityAnnotation(Annotation):
    """Track security requirements for nodes."""
    
    @classmethod
    def key(cls) -> str:
        return "security"
    
    def __init__(
        self,
        requires_auth: bool = True,
        required_roles: Optional[List[str]] = None,
        encryption_required: bool = False,
    ):
        self.requires_auth = requires_auth
        self.required_roles = required_roles or []
        self.encryption_required = encryption_required

# Use custom annotation
security = SecurityAnnotation(
    requires_auth=True,
    required_roles=["admin", "operator"],
    encryption_required=True,
)
node.metadata.annotate(security)

# Retrieve later
security = node.metadata.get_annotation(SecurityAnnotation)
if security and "admin" in security.required_roles:
    # Apply admin-only logic
    pass
```

**Complete Example: Cost Analysis**

```python
from universal_agent_nexus.ir.annotations import CostAnnotation
from universal_agent_nexus.ir.visitor import DefaultIRVisitor, traverse

class CostAnalyzer(DefaultIRVisitor):
    """Analyze total estimated cost of IR."""
    
    def __init__(self):
        self.total_cost_cents = 0.0
        self.total_tokens = 0
        self.total_latency_ms = 0.0
        self.node_costs = {}
    
    def visit_node(self, node):
        cost = node.metadata.get_annotation(CostAnnotation)
        if cost:
            self.total_cost_cents += cost.cost_cents
            self.total_tokens += cost.tokens
            self.total_latency_ms += cost.latency_ms
            self.node_costs[node.id] = cost
    
    def get_summary(self) -> dict:
        return {
            "total_cost_dollars": self.total_cost_cents / 100,
            "total_tokens": self.total_tokens,
            "total_latency_seconds": self.total_latency_ms / 1000,
            "node_count": len(self.node_costs),
        }

# Analyze IR
analyzer = CostAnalyzer()
traverse(ir, analyzer)
summary = analyzer.get_summary()
print(f"Estimated cost: ${summary['total_cost_dollars']:.2f}")
print(f"Estimated latency: {summary['total_latency_seconds']:.2f}s")
```

**Best Practices:**

- **Type Safety:** Always use typed annotations (`get_annotation<T>()`) instead of raw dict access
- **Immutability:** Annotations should be immutable after creation
- **Key Uniqueness:** Each annotation class must have a unique `key()`
- **Optional Access:** Always check for `None` when retrieving annotations

---

### **Tier 5: Factories**

**Purpose:** Customize how tools and routers are instantiated during parsing.

**API Reference:**

```python
from universal_agent_nexus.ir.factories import (
    ToolFactory,
    RouterFactory,
    set_tool_factory,
    set_router_factory,
    get_tool_factory,
    get_router_factory,
    DefaultToolFactory,
    DefaultRouterFactory,
)

# Set global factories
set_tool_factory(factory: ToolFactory) -> None
set_router_factory(factory: RouterFactory) -> None

# Get current factories
tool_factory = get_tool_factory() -> ToolFactory
router_factory = get_router_factory() -> RouterFactory
```

**Tool Factory Interface:**

```python
from universal_agent_nexus.ir.factories import ToolFactory
from universal_agent_nexus.ir import ToolIR

class CustomToolFactory(ToolFactory):
    """Custom tool factory with retry logic."""
    
    def create_tool(
        self,
        name: str,
        description: str,
        protocol: str,
        config: Dict[str, Any],
    ) -> ToolIR:
        """Create ToolIR with custom logic."""
        # Add retry configuration to all tools
        if "retry" not in config:
            config["retry"] = {
                "max_attempts": 3,
                "backoff_factor": 2,
                "timeout_ms": 5000,
            }
        
        # Add observability hooks
        config["observability"] = {
            "trace": True,
            "log_input": True,
            "log_output": False,  # Don't log sensitive outputs
        }
        
        return ToolIR(
            name=name,
            description=description,
            protocol=protocol,
            config=config,
        )

# Set global factory
set_tool_factory(CustomToolFactory())

# Now all parsed tools will have retry logic
from universal_agent_nexus.compiler import parse
ir = parse("agent.yaml")  # Uses custom factory
```

**Router Factory Interface:**

```python
from universal_agent_nexus.ir.factories import RouterFactory
from universal_agent_nexus.ir import RouterIR, RouterStrategy

class SmartRouterFactory(RouterFactory):
    """Router factory with intelligent model selection."""
    
    def create_router(
        self,
        name: str,
        strategy: RouterStrategy,
        system_message: Optional[str],
        model_candidates: List[str],
        default_model: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
    ) -> RouterIR:
        """Create RouterIR with smart defaults."""
        # If no default model, select based on strategy
        if not default_model and model_candidates:
            if strategy == RouterStrategy.LLM:
                # Prefer faster models for LLM routing
                default_model = next(
                    (m for m in model_candidates if "mini" in m.lower()),
                    model_candidates[0]
                )
            else:
                default_model = model_candidates[0]
        
        # Add fallback configuration
        config = config or {}
        if "fallback" not in config:
            config["fallback"] = {
                "enabled": True,
                "model": model_