diff --git a/changelog.d/tsk-hydxwo-orb-bridges.md b/changelog.d/tsk-hydxwo-orb-bridges.md new file mode 100644 index 000000000..ab04c2769 --- /dev/null +++ b/changelog.d/tsk-hydxwo-orb-bridges.md @@ -0,0 +1,11 @@ +### Added + +- Orb bridges adapter contract (tinyagentos/orb/bridges/__init__.py): + - BridgeResult dataclass with bridge_id, title, installed, targets, instruction and to_dict() + - Pure helper functions: orbs_running, orb_host, target + - Auto-discovery via pkgutil.iter_modules in adapters() + - evaluate_all runs all adapters, logs errors per adapter, one failure does not hide others +- MimiClaw reference adapter (tinyagentos/orb/bridges/mimiclaw.py): + - BRIDGE_ID="mimiclaw", TITLE="MimiClaw", APP_ID="mimiclaw", WS_PORT=18789 + - installed() returns None (no host-side tool) + - evaluate() builds targets from LAN orbs running MimiClaw with WebSocket URL instruction \ No newline at end of file diff --git a/tests/orb/test_bridges.py b/tests/orb/test_bridges.py new file mode 100644 index 000000000..76b46fdd9 --- /dev/null +++ b/tests/orb/test_bridges.py @@ -0,0 +1,244 @@ +#!/usr/bin/env python3 +"""Tests for tinyagentos/orb/bridges.""" + +import sys +from types import ModuleType, SimpleNamespace + +import pytest + + +def test_orbs_running_filters_on_app_and_keeps_order(): + """Test orbs_running filters on app and keeps input order.""" + from tinyagentos.orb.bridges import orbs_running + + lan_orbs = [ + {"name": "orb-1", "app": "mimiclaw", "addresses": ["10.0.0.1"], "port": 80}, + {"name": "orb-2", "app": "esphome", "addresses": ["10.0.0.2"], "port": 80}, + {"name": "orb-3", "app": "mimiclaw", "addresses": ["10.0.0.3"], "port": 80}, + {"name": "orb-4", "app": None, "addresses": ["10.0.0.4"], "port": 80}, + ] + + result = orbs_running(lan_orbs, "mimiclaw") + + assert len(result) == 2 + assert result[0]["name"] == "orb-1" + assert result[1]["name"] == "orb-3" + + +def test_orb_host_ipv4_first(): + """Test orb_host returns first IPv4 address.""" + from tinyagentos.orb.bridges import orb_host + + orb = {"addresses": ["10.0.0.5", "fe80::1"], "name": "orb-1"} + assert orb_host(orb) == "10.0.0.5" + + +def test_orb_host_ipv6_wrapped(): + """Test orb_host returns IPv6 address wrapped in brackets.""" + from tinyagentos.orb.bridges import orb_host + + orb = {"addresses": ["fe80::1"], "name": "orb-1"} + assert orb_host(orb) == "[fe80::1]" + + +def test_orb_host_fallback_to_name_local(): + """Test orb_host falls back to name.local when no addresses.""" + from tinyagentos.orb.bridges import orb_host + + orb = {"addresses": [], "name": "orb-1"} + assert orb_host(orb) == "orb-1.local" + + +def test_orb_host_empty_name_returns_none(): + """Test orb_host returns None when name is empty string.""" + from tinyagentos.orb.bridges import orb_host + + orb = {"addresses": [], "name": ""} + assert orb_host(orb) is None + + +def test_orb_host_no_addresses_key_no_name_returns_none(): + """Test orb_host returns None when dict has no addresses key and no name.""" + from tinyagentos.orb.bridges import orb_host + + orb = {} + assert orb_host(orb) is None + + +def test_target_builds_dict(): + """Test target builds the expected dict shape.""" + from tinyagentos.orb.bridges import target + + result = target("orb-1", "192.168.1.20", "Connect to ws://...") + assert result == {"orb": "orb-1", "address": "192.168.1.20", "instruction": "Connect to ws://..."} + + +def test_bridge_result_to_dict_round_trip(): + """Test BridgeResult.to_dict returns correct shape with targets as list of dicts.""" + from tinyagentos.orb.bridges import BridgeResult + + br = BridgeResult( + bridge_id="test", + title="Test Bridge", + installed=True, + targets=({"orb": "orb-1", "address": "1.2.3.4", "instruction": "connect"},), + instruction="Top instruction", + ) + + d = br.to_dict() + assert d["bridge_id"] == "test" + assert d["title"] == "Test Bridge" + assert d["installed"] is True + assert d["targets"] == [{"orb": "orb-1", "address": "1.2.3.4", "instruction": "connect"}] + assert d["instruction"] == "Top instruction" + + +def test_adapters_includes_mimiclaw_and_sorted(): + """Test adapters() includes mimiclaw module and all have BRIDGE_ID/TITLE/evaluate, sorted by BRIDGE_ID.""" + from tinyagentos.orb.bridges import adapters + + modules = adapters() + assert len(modules) >= 1 + bridge_ids = [m.BRIDGE_ID for m in modules] + assert "mimiclaw" in bridge_ids + # Check sorted by BRIDGE_ID + assert bridge_ids == sorted(bridge_ids) + for m in modules: + assert hasattr(m, "BRIDGE_ID") + assert hasattr(m, "TITLE") + assert hasattr(m, "evaluate") + assert callable(m.evaluate) + + +def test_adapters_raises_type_error_for_missing_evaluate(monkeypatch): + """Test adapters() raises TypeError for a module missing evaluate.""" + import pkgutil + import importlib + import tinyagentos.orb.bridges as bridges_mod + + # Create a fake module without evaluate + fake_module = ModuleType("tinyagentos.orb.bridges.broken") + fake_module.BRIDGE_ID = "broken" + fake_module.TITLE = "Broken Bridge" + # No evaluate attribute + + def fake_iter_modules(path): + return [SimpleNamespace(name="broken")] + + def fake_import_module(name): + return fake_module + + monkeypatch.setattr(pkgutil, "iter_modules", fake_iter_modules) + monkeypatch.setattr(importlib, "import_module", fake_import_module) + + with pytest.raises(TypeError, match="bridge adapter broken missing BRIDGE_ID/TITLE/evaluate"): + bridges_mod.adapters() + + +def test_evaluate_all_monkeypatch_adapters(monkeypatch): + """Test evaluate_all with fake adapters, one raising RuntimeError.""" + import tinyagentos.orb.bridges as bridges_mod + + # Create good adapter module + good_module = ModuleType("tinyagentos.orb.bridges.good") + good_module.BRIDGE_ID = "good" + good_module.TITLE = "Good Bridge" + good_result = bridges_mod.BridgeResult( + bridge_id="good", + title="Good Bridge", + installed=False, + targets=(bridges_mod.target("orb-1", "1.2.3.4", "connect"),), + instruction="Good instruction", + ) + + def good_evaluate(lan_orbs, agent_buddy): + return good_result + + good_module.evaluate = good_evaluate + + # Create bad adapter module + bad_module = ModuleType("tinyagentos.orb.bridges.bad") + bad_module.BRIDGE_ID = "bad" + bad_module.TITLE = "Bad Bridge" + + def bad_evaluate(lan_orbs, agent_buddy): + raise RuntimeError("adapter failed") + + bad_module.evaluate = bad_evaluate + + def fake_adapters(): + return [bad_module, good_module] # unsorted, evaluate_all doesn't sort + + monkeypatch.setattr(bridges_mod, "adapters", fake_adapters) + + result = bridges_mod.evaluate_all([], []) + + assert len(result) == 2 + # Find good and bad results + good_found = False + bad_found = False + for r in result: + if r.get("bridge_id") == "good": + good_found = True + assert r["title"] == "Good Bridge" + assert r["installed"] is False + assert len(r["targets"]) == 1 + assert r["instruction"] == "Good instruction" + elif r.get("bridge_id") == "bad": + bad_found = True + assert r["title"] == "Bad Bridge" + assert r["error"] == "adapter_failed" + assert good_found + assert bad_found + + +def test_mimiclaw_evaluate_two_orbs_one_matching(): + """Test mimiclaw.evaluate with two LAN orbs, one matching.""" + from tinyagentos.orb.bridges import mimiclaw + + lan_orbs = [ + {"name": "orb-1", "app": "mimiclaw", "addresses": ["192.168.1.20"], "port": 80}, + {"name": "orb-2", "app": "esphome", "addresses": ["192.168.1.21"], "port": 80}, + ] + agent_buddy = [] + + result = mimiclaw.evaluate(lan_orbs, agent_buddy) + + assert result.bridge_id == "mimiclaw" + assert result.title == "MimiClaw" + assert result.installed is None + assert len(result.targets) == 1 + assert result.targets[0]["orb"] == "orb-1" + assert result.targets[0]["address"] == "192.168.1.20" + assert result.targets[0]["instruction"] == "Connect a WebSocket client to ws://192.168.1.20:18789/" + assert result.instruction == "Connect a WebSocket client to the URL shown for your Orb." + + +def test_mimiclaw_evaluate_empty_input(): + """Test mimiclaw.evaluate with empty input returns no targets and correct instruction.""" + from tinyagentos.orb.bridges import mimiclaw + + result = mimiclaw.evaluate([], []) + + assert result.bridge_id == "mimiclaw" + assert result.title == "MimiClaw" + assert result.installed is None + assert len(result.targets) == 0 + assert result.instruction == "No Orb running MimiClaw found on the LAN." + + +def test_mimiclaw_installed_returns_none(): + """Test mimiclaw.installed returns None.""" + from tinyagentos.orb.bridges import mimiclaw + + assert mimiclaw.installed() is None + + +def test_mimiclaw_constants(): + """Test mimiclaw module constants.""" + from tinyagentos.orb.bridges import mimiclaw + + assert mimiclaw.BRIDGE_ID == "mimiclaw" + assert mimiclaw.TITLE == "MimiClaw" + assert mimiclaw.APP_ID == "mimiclaw" + assert mimiclaw.WS_PORT == 18789 \ No newline at end of file diff --git a/tinyagentos/orb/bridges/__init__.py b/tinyagentos/orb/bridges/__init__.py new file mode 100644 index 000000000..684fec539 --- /dev/null +++ b/tinyagentos/orb/bridges/__init__.py @@ -0,0 +1,110 @@ +"""Orb bridge adapters. + +Adapters are PURE. They take the lists that +`tinyagentos.orb.discovery.browse_lan_orbs()` and `probe_agent_buddy()` +already return and never call discovery, the network or a subprocess. + +Shapes (from tinyagentos/orb/discovery.py on dev): +- LAN orb: {"name": str, "addresses": list[str], "port": int, "board": str|None, "app": str|None, "version": str|None} +- Agent Buddy reply: {"device_id": str, "hostname": str, "address": str, "port": int, "boot": int} +""" + +import importlib +import logging +import pkgutil +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class BridgeResult: + bridge_id: str + title: str + installed: bool | None + targets: tuple[dict, ...] + instruction: str + + def to_dict(self) -> dict: + return { + "bridge_id": self.bridge_id, + "title": self.title, + "installed": self.installed, + "targets": [dict(t) for t in self.targets], + "instruction": self.instruction, + } + + +def orbs_running(lan_orbs: list[dict], app_id: str) -> list[dict]: + """Return orbs whose .get('app') == app_id, keeping input order.""" + return [orb for orb in lan_orbs if orb.get("app") == app_id] + + +def orb_host(orb: dict) -> str | None: + """Extract a host address from an orb dict. + + Priority: + 1. First IPv4 address (no colon) from addresses + 2. First IPv6 address (contains colon) wrapped in brackets + 3. Fallback to {name}.local if name is non-empty + 4. None otherwise + """ + addresses = orb.get("addresses") or [] + for addr in addresses: + if ":" not in addr: + return addr + for addr in addresses: + if ":" in addr: + return f"[{addr}]" + name = orb.get("name") + if isinstance(name, str) and name: + return f"{name}.local" + return None + + +def target(orb_name: str, address: str, instruction: str) -> dict: + """Build a target dict with fixed shape.""" + return {"orb": orb_name, "address": address, "instruction": instruction} + + +def adapters() -> list: + """Discover and load bridge adapter modules. + + Iterates pkgutil.iter_modules(__path__), skips names starting with '_', + imports each module, and validates it has BRIDGE_ID (str), TITLE (str), + and a callable evaluate. Returns modules sorted by BRIDGE_ID. + """ + modules = [] + for module_info in pkgutil.iter_modules(__path__): + name = module_info.name + if name.startswith("_"): + continue + mod = importlib.import_module(f"{__name__}.{name}") + if not hasattr(mod, "BRIDGE_ID") or not hasattr(mod, "TITLE") or not hasattr(mod, "evaluate"): + raise TypeError(f"bridge adapter {name} missing BRIDGE_ID/TITLE/evaluate") + if not callable(mod.evaluate): + raise TypeError(f"bridge adapter {name} missing BRIDGE_ID/TITLE/evaluate") + modules.append(mod) + modules.sort(key=lambda m: m.BRIDGE_ID) + return modules + + +def evaluate_all(lan_orbs: list[dict], agent_buddy: list[dict]) -> list[dict]: + """Evaluate all bridge adapters. + + For each adapter, calls mod.evaluate(lan_orbs, agent_buddy).to_dict(). + If an exception is raised, logs it and returns an error dict instead. + One broken adapter must not hide the others. + """ + results = [] + for mod in adapters(): + try: + results.append(mod.evaluate(lan_orbs, agent_buddy).to_dict()) + except Exception: + logger.exception("bridge adapter %s failed", mod.BRIDGE_ID) + results.append({ + "bridge_id": mod.BRIDGE_ID, + "title": mod.TITLE, + "error": "adapter_failed", + }) + return results \ No newline at end of file diff --git a/tinyagentos/orb/bridges/mimiclaw.py b/tinyagentos/orb/bridges/mimiclaw.py new file mode 100644 index 000000000..f0fabf156 --- /dev/null +++ b/tinyagentos/orb/bridges/mimiclaw.py @@ -0,0 +1,48 @@ +"""MimiClaw bridge adapter. + +Reference adapter for the MimiClaw WebSocket interface. +MimiClaw has no host-side tool; installed() returns None. +""" + +from . import BridgeResult, orbs_running, orb_host, target + +BRIDGE_ID = "mimiclaw" +TITLE = "MimiClaw" +APP_ID = "mimiclaw" +WS_PORT = 18789 # upstream memovai/mimiclaw main/mimi_config.h MIMI_WS_PORT + + +def installed() -> bool | None: + """MimiClaw has no host-side tool.""" + return None + + +def evaluate(lan_orbs: list[dict], agent_buddy: list[dict]) -> BridgeResult: + """Evaluate MimiClaw bridges from discovered LAN orbs. + + agent_buddy is unused (kept for adapter signature compatibility). + """ + matching_orbs = orbs_running(lan_orbs, APP_ID) + targets = [] + for orb in matching_orbs: + host = orb_host(orb) + if host is not None: + targets.append( + target( + orb["name"], + host, + f"Connect a WebSocket client to ws://{host}:{WS_PORT}/", + ) + ) + targets_tuple = tuple(targets) + if targets: + instruction = "Connect a WebSocket client to the URL shown for your Orb." + else: + instruction = "No Orb running MimiClaw found on the LAN." + return BridgeResult( + bridge_id=BRIDGE_ID, + title=TITLE, + installed=installed(), + targets=targets_tuple, + instruction=instruction, + ) \ No newline at end of file