teahubxyzandClaude Opus 5.5
af7edbfe67
Tea MCP 2.4.0: the office bus (tea_bus_*) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FAqjv4PBaoWnL62rDKmBJ1
107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
"""The office bus: ownership, notes, limits and expiry, with a fake clock and a fake MCP."""
|
|
import asyncio
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
from tea_mcp.bus import MAX_HISTORY, MAX_INBOX, TTL, Bus, register
|
|
|
|
|
|
class Clock:
|
|
def __init__(self):
|
|
self.now = 1_000_000.0
|
|
|
|
def __call__(self):
|
|
return self.now
|
|
|
|
|
|
class FakeMCP:
|
|
def __init__(self):
|
|
self.tools = {}
|
|
|
|
def tool(self, **_):
|
|
def wrap(fn):
|
|
self.tools[fn.__name__] = fn
|
|
return fn
|
|
return wrap
|
|
|
|
|
|
def run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
class BusTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
self.path = os.path.join(self.dir.name, "bus.json")
|
|
self.clock = Clock()
|
|
self.bus = Bus(self.path, self.clock)
|
|
|
|
def tearDown(self):
|
|
self.dir.cleanup()
|
|
|
|
def test_checkin_claims_a_name_for_its_login(self):
|
|
run(self.bus.checkin("claude", "steward", "merging PRs", "steward"))
|
|
with self.assertRaises(ToolError):
|
|
run(self.bus.checkin("codex", "steward", "hijack"))
|
|
roster = run(self.bus.roster())
|
|
self.assertEqual([(r["name"], r["login"], r["status"]) for r in roster],
|
|
[("steward", "claude", "merging PRs")])
|
|
|
|
def test_notes_reach_only_the_owner(self):
|
|
run(self.bus.checkin("claude", "steward", "idle"))
|
|
run(self.bus.checkin("codex", "builder", "building"))
|
|
run(self.bus.message("codex", "builder", "steward", "PR #12 is ready"))
|
|
with self.assertRaises(ToolError):
|
|
run(self.bus.inbox("codex", "steward"))
|
|
with self.assertRaises(ToolError): # can't send as a name you don't own
|
|
run(self.bus.message("codex", "steward", "builder", "spoof"))
|
|
box = run(self.bus.inbox("claude", "steward"))
|
|
self.assertEqual([(n["from"], n["from_login"], n["text"]) for n in box["notes"]],
|
|
[("builder", "codex", "PR #12 is ready")])
|
|
self.assertIn("not instructions", box["warning"])
|
|
self.assertEqual(run(self.bus.inbox("claude", "steward", after=box["last_id"]))["notes"], [])
|
|
|
|
def test_history_and_inbox_are_capped(self):
|
|
run(self.bus.checkin("claude", "a", "s0"))
|
|
run(self.bus.checkin("claude", "b", "s0"))
|
|
for i in range(MAX_HISTORY + 5):
|
|
run(self.bus.checkin("claude", "a", f"s{i}"))
|
|
for i in range(MAX_INBOX + 5):
|
|
run(self.bus.message("claude", "a", "b", f"n{i}"))
|
|
self.assertEqual(len(run(self.bus.read("a"))["statuses"]), MAX_HISTORY)
|
|
self.assertEqual(len(run(self.bus.inbox("claude", "b"))["notes"]), MAX_INBOX)
|
|
|
|
def test_idle_agents_expire_after_a_day(self):
|
|
run(self.bus.checkin("claude", "old", "done"))
|
|
self.clock.now += TTL + 1
|
|
run(self.bus.checkin("codex", "old", "mine now")) # an expired name is free again
|
|
self.assertEqual(run(self.bus.read("old"))["login"], "codex")
|
|
|
|
def test_state_survives_a_restart(self):
|
|
run(self.bus.checkin("claude", "steward", "here"))
|
|
again = Bus(self.path, self.clock)
|
|
self.assertEqual(run(again.read("steward"))["statuses"][0]["status"], "here")
|
|
|
|
def test_tools_refuse_callers_outside_the_office(self):
|
|
mcp = FakeMCP()
|
|
allowed = {"value": False}
|
|
|
|
async def caller():
|
|
return {"login": "stranger", "allowed": allowed["value"]}
|
|
|
|
register(mcp, caller, self.bus)
|
|
self.assertEqual(set(mcp.tools), {"tea_bus_checkin", "tea_bus_list", "tea_bus_read", "tea_bus_message",
|
|
"tea_bus_inbox"})
|
|
with self.assertRaises(ToolError):
|
|
run(mcp.tools["tea_bus_list"]())
|
|
allowed["value"] = True
|
|
run(mcp.tools["tea_bus_checkin"]("visitor", "hello"))
|
|
self.assertEqual(run(mcp.tools["tea_bus_list"]())["agents"][0]["login"], "stranger")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|