# TopologyValidator API Reference

## Overview

`TopologyValidator` provides deterministic (non-AI), rule-based static analysis for subgraphs generated by `AdaptiveNode`. It is required for EU AI Act Art. 3(23) compliance when using runtime graph composition.

Every `AdaptiveNode` must be initialised with a `TopologyValidator` instance. The validator runs before any node in the generated spec executes. If the spec is rejected, execution falls through to `next_node` and the rejection reason is written to the audit log.

Compliance tags: **Art. 3(23)** (Substantial Modification control), **Art. 9** (Risk Management), **Art. 12** (Causal Trace logging)

## EU AI Act Article Mapping

| TopologyValidator invariant | EU AI Act article | Requirement satisfied |
|---|---|---|
| Cycle detection | Art. 9 — Risk Management | Prevents unbounded execution loops that constitute unmitigated runtime risk |
| Tool allowlist | Art. 9 — Risk Management | Privilege minimisation — restricts actions to pre-approved functions only |
| Structural integrity | Art. 3(23) — Substantial Modification | Ensures the injected subgraph is well-formed before it enters the execution path |
| Rejection logged to Causal Trace | Art. 12 — Record Keeping | Every validation decision — pass or fail — is traceable in the audit log |

## Class Signature

```python
class TopologyValidator:
    def __init__(self, allowed_tools: List[callable] = None)
```

## Parameters

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `allowed_tools` | `List[callable]` | No | Allowlist of Python functions that can appear in `ToolNode` entries of generated specs. If `None`, all tools are permitted — this is insecure and not recommended for production. |

## Methods

### `validate(graph_spec: Dict[str, Any]) -> None`

Validates a JSON graph specification for correctness and compliance.

**Raises**: `SecurityError` if the spec violates any invariant.

**Checks performed**:
1. **Cycle detection**: Ensures the graph is a DAG (Directed Acyclic Graph) — no execution path can loop
2. **Tool allowlist**: Verifies all `ToolNode` entries reference functions in `allowed_tools`
3. **Structural integrity**: Confirms all `next_node` references resolve to declared nodes
4. **Type safety**: Confirms node types are from the supported set (`LLMNode`, `ToolNode`, `BatchNode`, `AdaptiveNode`)

## Security Guarantees

### 1. Cycle Detection — Infinite Loop Prevention

`TopologyValidator` uses depth-first search to confirm the generated spec is acyclic before execution:

```python
validator = TopologyValidator()

# This raises SecurityError
invalid_graph = {
    "nodes": {
        "node_1": {"next_node": "node_2"},
        "node_2": {"next_node": "node_1"}  # Cycle
    },
    "start_node": "node_1"
}

validator.validate(invalid_graph)  # SecurityError: Cycle detected
```

### 2. Tool Allowlist — Privilege Minimisation

Only pre-approved functions can be referenced in `ToolNode` entries:

```python
def safe_tool():
    return "Safe operation"

def unapproved_function():
    return "Not permitted"

validator = TopologyValidator(allowed_tools=[safe_tool])

# Passes validation
safe_graph = {
    "nodes": {
        "node_1": {
            "type": "ToolNode",
            "config": {"tool_name": "safe_tool"},
            "next_node": None
        }
    },
    "start_node": "node_1"
}
validator.validate(safe_graph)  # Passes

# Rejected
rejected_graph = {
    "nodes": {
        "node_1": {
            "type": "ToolNode",
            "config": {"tool_name": "unapproved_function"},
            "next_node": None
        }
    },
    "start_node": "node_1"
}
validator.validate(rejected_graph)  # SecurityError: Tool not in allowlist
```

### 3. Reference Integrity

All `next_node` references must resolve to declared node IDs:

```python
broken_graph = {
    "nodes": {
        "node_1": {"next_node": "node_999"}  # node_999 not declared
    },
    "start_node": "node_1"
}

validator.validate(broken_graph)  # SecurityError: Invalid reference
```

## Example: Basic Validation

```python
from lar.adaptive import TopologyValidator, SecurityError

def approved_search(query: str) -> str:
    return f"Results for {query}"

def approved_filter(results: list) -> list:
    return [r for r in results if "important" in r]

validator = TopologyValidator(allowed_tools=[approved_search, approved_filter])

llm_generated_graph = {
    "nodes": {
        "search": {
            "type": "ToolNode",
            "config": {
                "tool_name": "approved_search",
                "input_keys": ["query"],
                "output_key": "raw_results"
            },
            "next_node": "filter"
        },
        "filter": {
            "type": "ToolNode",
            "config": {
                "tool_name": "approved_filter",
                "input_keys": ["raw_results"],
                "output_key": "filtered_results"
            },
            "next_node": None
        }
    },
    "start_node": "search"
}

try:
    validator.validate(llm_generated_graph)
    print("Graph passed validation")
except SecurityError as e:
    print(f"Validation failed: {e}")
```

## Example: Integration with AdaptiveNode

```python
from lar import AdaptiveNode, TopologyValidator

def fetch_wikipedia(topic: str) -> str:
    return f"Wikipedia summary of {topic}"

def summarize_text(text: str) -> str:
    return text[:100]

validator = TopologyValidator(allowed_tools=[fetch_wikipedia, summarize_text])

research_agent = AdaptiveNode(
    llm_model="gpt-4o",
    prompt_template="""
    Design a research subgraph for: {research_topic}

    Available tools:
    - fetch_wikipedia
    - summarize_text

    Return JSON graph spec with nodes and entry_point.
    """,
    validator=validator,
    context_keys=["research_topic"]
)
```

## Best Practices

### Principle of Least Privilege

Restrict the allowlist to the minimum set of functions required:

```python
# Not recommended — all tools permitted
validator = TopologyValidator(allowed_tools=None)

# Recommended — explicit, minimal allowlist
validator = TopologyValidator(allowed_tools=[read_only_function])
```

### Layered Validation

Extend `TopologyValidator` to add domain-specific constraints:

```python
class CustomValidator(TopologyValidator):
    def validate(self, graph_spec):
        super().validate(graph_spec)

        if len(graph_spec["nodes"]) > 10:
            raise SecurityError("Graph exceeds maximum node count (10)")

        for node in graph_spec["nodes"].values():
            if node["type"] == "LLMNode":
                if "gpt-4" in node["config"].get("model_name", ""):
                    raise SecurityError("GPT-4 not permitted (cost control policy)")
```

### Audit Logging

Log all validation outcomes for compliance review:

```python
try:
    validator.validate(graph_spec)
    logger.info(f"Graph validated: {graph_spec}")
except SecurityError as e:
    logger.error(f"Validation rejected: {e} — spec: {graph_spec}")
    raise
```

## Common Errors

| Error | Cause | Resolution |
|-------|-------|------------|
| `Cycle detected in graph topology` | Node A references Node B which references Node A | Remove the cyclic edge |
| `Tool 'X' not in allowlist` | Generated spec references a function not in `allowed_tools` | Add the function to `allowed_tools` or remove it from the spec |
| `start_node 'X' not found in nodes` | Entry point references a non-existent node ID | Ensure `start_node` matches a declared node key |
| `Invalid node type: 'X'` | Unsupported node class in generated spec | Use only: `LLMNode`, `ToolNode`, `BatchNode`, `AdaptiveNode` |

## See Also

- [AdaptiveNode API](adaptivenode.md) — runtime graph composition primitive
- [Adaptive Graphs](../core-concepts/9-adaptive-graphs.md) — when and why to use runtime graph composition
- [Red Teaming](../case-studies/red-teaming.md) — security testing
- [Security Firewall Example](https://github.com/snath-ai/lar/blob/main/examples/compliance/2_security_firewall.py)
