fleet-router-integration

Route tasks across a fleet using Laman rigidity, holonomy consensus, and deadband filtering.

What it does

When you have N agents in a distributed system, you need to answer three questions:

1. Is the topology rigid? — Can the network survive agent failures without losing connectivity?
2. Do agents agree on coordinates? — Are their transform frames holonomy-consistent?
3. Which agent should handle this task? — Route to the nearest capable healthy agent.

This package answers all three. It's the integration layer that combines:

• Laman rigidity — topological guarantee: 2V−3 edges, no floppy degrees of freedom
• Holonomy checking — verify coordinate frame consistency across all agent pairs
• Deadband filtering — suppress health report noise (only propagate changes > threshold)
• Capability routing — match queries to agents by type, distance, and health

Install

pip install fleet-router-integration

Quick start

from fleet_router_integration.agent import Agent
from fleet_router_integration.router import FleetRouter, build_minimal_rigid_edges

# Create a fleet router
router = FleetRouter()

# Register agents with positions, capabilities, and coordinate transforms
router.register(Agent("alpha", (0.0, 0.0), ["query", "mutation"]), transform=(1.0, 0.0, 0.0, 1.0))
router.register(Agent("beta",  (3.0, 1.0), ["query", "analytics"]),  transform=(1.0, 0.01, -0.01, 1.0))
router.register(Agent("gamma", (1.0, 5.0), ["mutation", "analytics"]), transform=(0.99, 0.0, 0.0, 1.01))

# Check topology rigidity (Laman condition: |E| = 2V-3)
print(f"Topology rigid: {router.topology_is_rigid()}")

# Check coordinate frame consistency
print(f"Holonomy consistent: {router.holonomy_consistent()}")

# Route a query to the nearest capable agent
agent = router.route("query", source_pos=(2.0, 2.0))
print(f"Routed to: {agent.id}")  # → beta (nearest agent that handles "query")

# Update health with deadband filtering
router.update_health("beta", 0.95)   # first report → always propagates
router.update_health("beta", 0.94)   # change < 0.05 → suppressed
router.update_health("beta", 0.80)   # change ≥ 0.05 → propagated

Example workflow: build fleet, verify rigidity, route tasks

from fleet_router_integration.agent import Agent
from fleet_router_integration.router import (
    FleetRouter,
    build_minimal_rigid_edges,
    si_laman_edges,
    si_is_rigid,
)

# 1. Build a fleet of 8 agents
router = FleetRouter()
positions = [(0,0), (2,1), (4,0), (1,3), (3,3), (5,2), (2,5), (4,5)]
capabilities = [
    ["query", "mutation"],
    ["query", "analytics"],
    ["mutation", "analytics"],
    ["query"],
    ["analytics"],
    ["query", "mutation"],
    ["mutation"],
    ["query", "analytics"],
]

for i, (pos, caps) in enumerate(zip(positions, capabilities)):
    agent = Agent(f"agent-{i}", pos, caps)
    router.register(agent)

# 2. Verify Laman rigidity
n = len(router.agents)
needed_edges = si_laman_edges(n)
edges = build_minimal_rigid_edges(n)
print(f"Agents: {n}, Laman edges needed: {needed_edges}, built: {len(edges)}")
print(f"Rigid: {si_is_rigid(n, edges)}")

# 3. Route tasks
for task in ["query", "mutation", "analytics"]:
    agent = router.route(task, source_pos=(2.5, 2.5))
    if agent:
        print(f"Task '{task}' → {agent.id} (health={agent.health:.2f})")

Performance

Check	Complexity	Notes
Laman rigidity	O(2^N × N)	Exact check; Henneberg construction is O(N)
Holonomy consistency	O(N²)	All pairs; O(N) for cycles
Deadband filter	O(1) per update	Hash lookup + threshold compare
Route query	O(N)	Scan candidates, sort by distance

For large fleets (>100 agents), use build_minimal_rigid_edges() for O(N) topology construction instead of the exact rigidity check.

Architecture

┌──────────────────────────────────────────────┐
│              FleetRouter                      │
│                                              │
│  ┌──────────┐  ┌──────────────┐  ┌────────┐ │
│  │  Agent    │  │ HolonomyChk  │  │Deadband│ │
│  │  Registry │  │ (2×2 matrix) │  │ Filter │ │
│  └────┬─────┘  └──────┬───────┘  └───┬────┘ │
│       │               │              │       │
│       └───────────────┴──────────────┘       │
│                       │                      │
│              route() / topology_is_rigid()    │
└──────────────────────────────────────────────┘

Running tests

pip install pytest
pytest tests/

License

MIT