phase 0.9: multi-AI council pattern — test + demo

tests/test_council.py: e2e — 3 panel AIs (alpha/beta/gamma), coordinator
publishes a handler that connects to each panel and synthesizes replies,
client asks coordinator, asserts synthesized text cites each panel by
signature. proves bidirectional substrate supports arbitrary multi-AI
orchestration without bespoke routing.

examples/council_demo.py: user-facing version of the same pattern with
named stubs (claude-stub, gpt-stub, gemini-stub) so the printed output
reads like a real council. run alongside zhub.server, see synthesized
multi-AI reply.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Zawwarsami16

examples/council_demo.py

@@ -0,0 +1,103 @@
+"""Council demo — three published AIs, one coordinator, all wired through zhub.
+
+Run order:
+    1. python -m zhub.server --port 8080
+    2. python examples/council_demo.py
+
+The script:
+    - publishes 3 stub AIs with distinct voices
+    - publishes a coordinator AI whose chat-handler queries all 3
+    - connects a 'user' that talks to the coordinator
+    - prints the synthesized reply
+
+This is the bidirectional substrate's killer demo: any operator can wire
+multiple AIs into a council pattern without bespoke router code.
+"""
+
+import asyncio
+import logging
+
+from zhub import publish, connect
+
+
+HUB_URL = "ws://localhost:8080"
+
+
+def make_panel_handler(signature: str):
+    def handler(messages, options):
+        last = messages[-1].get("content", "")
+        return f"{signature} thinks: {last}"
+    return handler
+
+
+async def main():
+    logging.basicConfig(level=logging.INFO)
+
+    # --- 3 panel members ---
+    panel = []
+    for name, signature in [
+        ("claude-stub", "[claude]"),
+        ("gpt-stub",    "[gpt]"),
+        ("gemini-stub", "[gemini]"),
+    ]:
+        p = publish(
+            name=name,
+            description=f"panel: {name}",
+            chat_handler=make_panel_handler(signature),
+            hub_url=HUB_URL,
+            public=True,
+        )
+        panel.append(p)
+
+    # Wait for all panel members to register
+    for p in panel:
+        while not p.api_key:
+            await asyncio.sleep(0.1)
+    print(f"panel registered: {[p.name for p in panel]}")
+
+    # --- coordinator ---
+    panel_creds = [(p.name, p.api_key) for p in panel]
+
+    async def coordinator(messages, options):
+        question = messages[-1].get("content", "")
+        votes = []
+        for name, key in panel_creds:
+            sub = connect(
+                ai_name=name, api_key=key, hub_url=HUB_URL,
+                capabilities={},
+            )
+            await asyncio.sleep(0.15)
+            r = await sub.chat(messages=[{"role": "user", "content": question}])
+            votes.append(r.get("text", ""))
+        return "Council synthesis:\n  " + "\n  ".join(votes)
+
+    coord = publish(
+        name="council",
+        description="multi-AI council coordinator",
+        chat_handler=coordinator,
+        hub_url=HUB_URL,
+        public=True,
+    )
+    while not coord.api_key:
+        await asyncio.sleep(0.1)
+    print(f"coordinator registered: name={coord.name} key={coord.api_key[:14]}...")
+
+    # --- user side ---
+    user = connect(
+        ai_name=coord.name, api_key=coord.api_key, hub_url=HUB_URL,
+        capabilities={},
+    )
+    await asyncio.sleep(0.4)
+
+    for question in [
+        "what's the meaning of life?",
+        "should I deploy on Friday?",
+    ]:
+        print(f"\n>>> {question}")
+        resp = await user.chat(messages=[{"role": "user", "content": question}])
+        print(resp.get("text", "(no text)"))
+        await asyncio.sleep(0.2)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_council.py

@@ -0,0 +1,116 @@
+"""End-to-end council pattern — coordinator AI orchestrates 3 panel AIs through the hub."""
+
+import asyncio
+import socket
+import threading
+import time
+
+import pytest
+
+try:
+    import fastapi  # noqa
+    import uvicorn  # noqa
+    SERVER_AVAILABLE = True
+except ImportError:
+    SERVER_AVAILABLE = False
+
+if SERVER_AVAILABLE:
+    from zhub.server import create_app
+from zhub import publish, connect
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="module")
+def hub_port():
+    if not SERVER_AVAILABLE:
+        pytest.skip("fastapi/uvicorn not installed")
+    port = _free_port()
+
+    def run():
+        config = uvicorn.Config(create_app(), host="127.0.0.1", port=port, log_level="warning")
+        asyncio.run(uvicorn.Server(config).serve())
+
+    threading.Thread(target=run, daemon=True).start()
+    for _ in range(30):
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                break
+        except OSError:
+            time.sleep(0.1)
+    yield port
+
+
+@pytest.mark.asyncio
+async def test_coordinator_calls_all_panel_members(hub_port):
+    """The coordinator AI publishes itself, also connects to three other
+    published AIs, and synthesizes their replies."""
+    hub_url = f"ws://127.0.0.1:{hub_port}"
+
+    # Three "panel" AIs — each returns a distinct signature.
+    pubs = []
+    for name, signature in [("alpha", "[A]"), ("beta", "[B]"), ("gamma", "[C]")]:
+        p = publish(
+            name=name,
+            description=f"panel member {name}",
+            chat_handler=(lambda sig: (lambda m, o: f"{sig} {m[-1]['content']}"))(signature),
+            hub_url=hub_url,
+        )
+        pubs.append(p)
+
+    for p in pubs:
+        for _ in range(50):
+            if p.api_key:
+                break
+            await asyncio.sleep(0.05)
+        assert p.api_key, f"panel member {p.name} never registered"
+
+    # Coordinator: publishes a chat handler that internally connects to each
+    # panel member and aggregates.
+    panel_creds = [(p.name, p.api_key) for p in pubs]
+
+    async def coordinator_handler(messages, options):
+        question = messages[-1]["content"]
+        replies = []
+        for name, key in panel_creds:
+            sub = connect(
+                ai_name=name, api_key=key, hub_url=hub_url,
+                capabilities={},
+            )
+            await asyncio.sleep(0.1)
+            r = await sub.chat(messages=[{"role": "user", "content": question}])
+            replies.append(r.get("text", ""))
+        return "council: " + " | ".join(replies)
+
+    coord = publish(
+        name="coordinator",
+        description="multi-AI council",
+        chat_handler=coordinator_handler,
+        hub_url=hub_url,
+    )
+    for _ in range(50):
+        if coord.api_key:
+            break
+        await asyncio.sleep(0.05)
+    assert coord.api_key, "coordinator never registered"
+
+    # Connect a client that asks the coordinator a question.
+    client = connect(
+        ai_name=coord.name, api_key=coord.api_key, hub_url=hub_url,
+        capabilities={},
+    )
+    await asyncio.sleep(0.4)
+
+    resp = await asyncio.wait_for(
+        client.chat(messages=[{"role": "user", "content": "ping"}]),
+        timeout=15.0,
+    )
+    text = resp.get("text", "")
+    assert text.startswith("council:"), f"expected council prefix, got: {text!r}"
+    assert "[A] ping" in text, f"missing alpha signature: {text!r}"
+    assert "[B] ping" in text, f"missing beta signature: {text!r}"
+    assert "[C] ping" in text, f"missing gamma signature: {text!r}"