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

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:

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:

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:

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

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

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:

# 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:

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:

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`