Route agents by graph structure, not just capabilities. Minimize H¹ to maximize coordination.
A standalone Python library that routes fleet agents using H¹ cohomology as the primary routing signal. Built on FM's constraint theory: the fleet's routing graph is a rigidity framework, and the routing goal is to stay below the emergence threshold (β₁ ≤ V-2).
Traditional fleet routing routes agents based on capabilities:
- "Agent A can do task T → route A to T"
- Simple, but misses structural constraints
H¹-driven routing routes agents based on graph structure:
- The fleet's routing graph has nodes (agents + tasks) and edges (routing relationships)
- H¹ cohomology measures how under-constrained the graph is
- When H¹ exceeds threshold (β₁ > V-2), coordination fails → emergence detected
- Route agents to minimize H¹ → better coordination
FM's Theorem: For a graph G = (V, E), H¹_dim = E - V + H0
- H¹ > V-2 → under-constrained → coordination fails (EMERGENCE)
- H¹ = V-2 → exactly constrained → Laman rigid (GOOD)
- H¹ < V-2 → over-constrained → redundant edges (INEFFICIENT)
Routing Goal: Keep H¹ as close to V-2 as possible without exceeding it.
pip install fleet-h1-routerfrom fleet_h1_router import H1Router, RoutingDecision
# Create router
router = H1Router()
# Register fleet agents
router.register_agent("oracle1", ["coordination", "plato", "research"])
router.register_agent("ccc", ["telegram", "messaging"])
router.register_agent("forgemaster", ["compilation", "llvm", "constraint-theory"])
# Register tasks
router.register_task("plato-sync", ["coordination", "plato"])
router.register_task("model-benchmark", ["compilation", "research"])
router.register_task("fleet-health-check", ["coordination"])
# Route a task — H1-aware routing
decision = router.route_task("plato-sync")
print(f"Routing: {decision.task_id} → {decision.agent_id}")
print(f" H¹ before: {decision.h1_before}, H¹ after: {decision.h1_after}")
print(f" Emergence risk: {decision.emergence_risk:.1%}")
# Check fleet constraint status
status = router.get_fleet_status()
print(f"Fleet H¹: {status['h1_dimension']}, {status['description']}")
print(f"Emergence alert: {status['emergence_detected']}")# Start H1 router service
fleet-h1-router serve --port 8902
# Route a task via HTTP
curl -X POST http://localhost:8902/route \
-H "Content-Type: application/json" \
-d '{"task_id": "plato-sync", "required_capabilities": ["coordination", "plato"]}'
# Get fleet constraint status
curl http://localhost:8902/status
# Get all agents
curl http://localhost:8902/agents┌──────────────────────────────────────────────────────────────┐
│ H1 ROUTER │
│ │
│ Route Request │
│ │ │
│ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ Capability │───▶│ H¹ Compute │───▶│ Agent Selector │ │
│ │ Filter │ │ (E-V+H0) │ │ (minimize H¹) │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ Eligible Emergence Selected │
│ Agents Check Agent │
│ │
│ ┌─────────────────────────────────────────────────────────┐│
│ │ Routing Decision Output ││
│ │ {agent_id, task_id, h1_before, h1_after, confidence} ││
│ └─────────────────────────────────────────────────────────┘│
└──────────────────────────────────────────────────────────────┘
def compute_h1(n_vertices: int, n_edges: int, n_components: int = 1) -> int:
"""
H¹ cohomology dimension for a constraint graph.
H¹ = E - V + H0
Where:
- E = number of edges (routing relationships)
- V = number of vertices (agents + tasks)
- H0 = number of connected components
Emergence threshold: H¹ > V - 2
"""
return n_edges - n_vertices + n_components
def is_emergence(h1: int, n_vertices: int) -> bool:
"""Check if graph is in emergence regime."""
return h1 > (n_vertices - 2) if n_vertices >= 2 else FalseThe router continuously monitors fleet H¹:
- β₁ ≤ V-2: Graph is rigid, coordination succeeds
- β₁ > V-2: Graph is under-constrained, coordination fails → EMERGENCE
- Router actively routes to reduce H¹ and prevent emergence
#!/usr/bin/env python3
"""
H1Router with holodeck integration.
Routes agent tasks based on constraint graph structure.
"""
from fleet_h1_router import H1Router, RoutingDecision, EmergenceAlert
from datetime import datetime
import json
class HolodeckH1Router(H1Router):
"""
H1 router with PLATO integration for holodeck sessions.
Posts emergence alerts to PLATO when β₁ exceeds threshold.
"""
def __init__(self, plato_url: str = "http://localhost:8848", **kwargs):
super().__init__(**kwargs)
self.plato_url = plato_url
# Pre-populate with known holodeck agents
self.register_agent("oracle1", ["coordination", "plato", "research", "publishing", "crates"])
self.register_agent("ccc", ["telegram", "messaging", "ai"])
self.register_agent("forgemaster", ["compilation", "llvm", "constraint-theory"])
def on_emergence_detected(self, alert: EmergenceAlert):
"""Called when emergence threshold is crossed."""
print(f"⚠️ EMERGENCE ALERT: β₁={alert.h1} exceeds V-2={alert.threshold}")
# Post to PLATO
self._post_alert_to_plato(alert)
def _post_alert_to_plato(self, alert: EmergenceAlert):
"""Post emergence alert to PLATO."""
import urllib.request
payload = json.dumps({
"domain": "gc/research/h1-routing-experiment",
"question": f"[EMERGENCE ALERT] β₁={alert.h1} > V-2={alert.threshold}",
"answer": json.dumps({
"timestamp": alert.timestamp,
"h1_dimension": alert.h1,
"v_threshold": alert.threshold,
"n_vertices": alert.n_vertices,
"n_edges": alert.n_edges,
"routing_decision": alert.routing_decision,
"recommendation": "Route agents to resolve under-constrained regions"
}, indent=2),
"tags": ["emergency", "h1", "emergence", "routing"],
"confidence": 0.95,
"source": "fleet-h1-router"
}).encode()
try:
req = urllib.request.Request(
f"{self.plato_url}/submit",
data=payload,
headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=5) as resp:
print(f" PLATO alert posted: {resp.read()[:100]}")
except Exception as e:
print(f" PLATO alert failed: {e}")
def route_session_task(self, session_id: str, task_type: str, required_caps: list) -> RoutingDecision:
"""
Route a session task with H1 awareness.
Uses both capability matching AND H1 minimization.
"""
# Get decision
decision = self.route_task(
task_id=f"{session_id}:{task_type}",
required_capabilities=required_caps
)
# Emit routing tile
self._emit_routing_tile(decision, session_id, task_type)
return decision
def _emit_routing_tile(self, decision: RoutingDecision, session_id: str, task_type: str):
"""Emit routing decision to PLATO."""
import urllib.request
payload = json.dumps({
"domain": "gc/research/h1-routing-experiment",
"question": f"[H1_ROUTE] {decision.agent_id} → {decision.task_id}",
"answer": json.dumps({
"session_id": session_id,
"task_type": task_type,
"agent_id": decision.agent_id,
"h1_before": decision.h1_before,
"h1_after": decision.h1_after,
"emergence_risk": decision.emergence_risk,
"capability_match_score": decision.capability_match_score
}, indent=2),
"tags": ["routing", "h1", decision.agent_id],
"confidence": decision.confidence,
"source": "fleet-h1-router"
}).encode()
try:
req = urllib.request.Request(
f"{self.plato_url}/submit",
data=payload,
headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=5):
pass # Fire and forget
except:
pass
if __name__ == "__main__":
router = HolodeckH1Router()
# Route a few tasks
for task_name, caps in [
("plato-sync", ["coordination", "plato"]),
("model-benchmark", ["compilation"]),
("fleet-health-check", ["coordination"]),
]:
decision = router.route_task(task_name, required_capabilities=caps)
print(f"Route: {task_name} → {decision.agent_id} (H¹: {decision.h1_before}→{decision.h1_after})")
# Check fleet status
status = router.get_fleet_status()
print(f"\nFleet Status:")
print(f" H¹ dimension: {status['h1_dimension']}")
print(f" Description: {status['description']}")
print(f" Emergence: {status['emergence_detected']}")from fleet_h1_router import H1Router
router = H1Router(
emergence_threshold_offset=-2, # β₁ > V + offset triggers alert
min_confidence=0.5, # Minimum routing confidence
lookahead_steps=3 # How far to project H1 changes
)| Method | Description |
|---|---|
register_agent(id, capabilities) |
Register an agent with capabilities |
unregister_agent(id) |
Remove an agent |
register_task(task_id, required_caps) |
Register a task with requirements |
route_task(task_id, required_capabilities) |
Route a task (H1-aware) |
get_fleet_status() |
Get current fleet constraint status |
get_h1_snapshot() |
Get H¹ computation details |
reset_graph() |
Clear all edges and recompute |
@dataclass
class RoutingDecision:
agent_id: str # Selected agent
task_id: str # Task being routed
h1_before: int # H¹ before routing
h1_after: int # H¹ after routing (projected)
confidence: float # Decision confidence (0.0-1.0)
emergence_risk: float # Probability of emergence
capability_match_score: float # How well agent matches requirements
reasoning: str # Why this agent was chosen@dataclass
class FleetStatus:
n_agents: int # Number of registered agents
n_tasks: int # Number of registered tasks
n_edges: int # Number of routing edges
h1_dimension: int # Current H¹
description: str # "under-constrained" | "rigid" | "over-constrained"
emergence_detected: bool # Is H¹ > V-2?
margin: int # H¹ - (V-2) — how close to emergenceFor a constraint graph G = (V, E) with H₀ connected components:
H¹(G) = ker(∂₁) / im(∂₂)
dim(H¹) = E - V + H₀
A graph is minimally rigid (just enough constraints) when:
- E = 2V - 3 (exactly enough edges)
- H¹ = V - 2 (exactly at the emergence boundary)
β₁ = H¹_dim
Emergence occurs when: β₁ > V - 2
This means:
- More independent cycles than vertices can support
- The graph has more degrees of freedom than constraints
- Coordination fails because positions can't be determined
| H¹ Value | Graph State | Routing Action |
|---|---|---|
| H¹ > V-2 | Under-constrained | Route agents to ADD constraints |
| H¹ = V-2 | Exactly rigid | Optimal routing, maintain status quo |
| H¹ < V-2 | Over-constrained | Route agents to SIMPLIFY graph |
H1_ROUTER_PORT=8902 # Service port (default: 8902)
PLATO_URL=http://localhost:8848 # PLATO server
EMERGENCE_THRESHOLD=-2 # H¹ offset from V to trigger alert
LOG_LEVEL=INFO # Logging level# fleet-h1-router.yaml
router:
emergence_threshold_offset: -2 # β₁ > V-2 triggers alert
min_confidence: 0.5
lookahead_steps: 3
agents:
- id: oracle1
capabilities: [coordination, plato, research]
- id: ccc
capabilities: [telegram, messaging]
tasks:
- id: plato-sync
required_capabilities: [coordination, plato]
plato:
url: http://localhost:8848
alert_room: gc/research/h1-routing-experimentPrometheus metrics for monitoring:
fleet_h1_router_h1_dimension{state="current"}
fleet_h1_router_emergence_alerts_total
fleet_h1_router_routing_decisions_total{agent_id, outcome}
fleet_h1_router_emergence_risk{agent_id}
fleet_h1_router_route_duration_seconds
- H¹ is the primary signal. Capabilities are a filter, not the decision.
- Stay below the emergence threshold. Every routing decision projects forward.
- Emergence is not failure. It's a signal that constraints need rebalancing.
- H¹-guided routing > capability-guided routing when graph structure matters.
- Marginal routing is optimal. Route to keep H¹ as close to V-2 as possible.
| Scenario | Capability Routing | H¹ Routing |
|---|---|---|
| All agents capable | Picks random agent | Picks agent that minimizes H¹ |
| Emergence threshold approaching | Ignores | Warns + adjusts |
| Under-constrained graph | No awareness | Routes to add constraints |
| Over-constrained graph | No awareness | Routes to simplify |
| Graph with cycles | Ignores | Detects via H¹ > V-2 |
# Run tests
pytest tests/ -v
# Test H1 computation
pytest tests/test_h1.py -v
# Test routing decisions
pytest tests/test_routing.py -v
# Test emergence detection
pytest tests/test_emergence.py -vSee CONTRIBUTING.md for development setup.
Apache 2.0 — See LICENSE.