diff --git a/contracts/slim-l9-wire.json b/contracts/slim-l9-wire.json index 9dd76cbc..5147859f 100644 --- a/contracts/slim-l9-wire.json +++ b/contracts/slim-l9-wire.json @@ -23,6 +23,15 @@ "node_endpoint": "http://127.0.0.1:46357", "node_port": 46357, + "valid_subkinds": { + "_comment": "Kind -> allowed subkinds. Backend: app.services.l9.VALID_SUBKINDS (keyed by the Kind enum). CLI: mycelium.slim.l9.VALID_SUBKINDS (keyed by the plain string, since the hidden `l9 send`/`slim send` plumbing takes a raw --kind string). An empty/None subkind is always valid for any kind.", + "knowledge": ["distillation", "extraction", "feedback", "query"], + "commit": ["converged", "rejected", "resolved"], + "intent": ["coordinator-assignment", "mission"], + "exchange": ["team-formation"], + "contingency": ["negotiation"] + }, + "shared_secret": { "_comment": "mint_shared_secret is keyed on workspace/room only (agent ignored). HMAC-SHA256(master_secret, 'workspace/room').hexdigest().", "workspace": "acme", diff --git a/fastapi-backend/tests/test_slim_l9_wire.py b/fastapi-backend/tests/test_slim_l9_wire.py index 56fba31b..f128b22a 100644 --- a/fastapi-backend/tests/test_slim_l9_wire.py +++ b/fastapi-backend/tests/test_slim_l9_wire.py @@ -127,6 +127,14 @@ def test_knowledge_envelope_serializes_to_contract(): assert produced == g["expected_envelope"] +def test_valid_subkinds_match_contract(): + """The backend's subkind table matches the frozen contract the CLI mirrors.""" + g = {k: v for k, v in _contract()["valid_subkinds"].items() if k != "_comment"} + assert {k.value for k in l9.VALID_SUBKINDS} == set(g) + for kind, allowed in g.items(): + assert l9.VALID_SUBKINDS[Kind(kind)] == frozenset(allowed) + + def test_channel_name_topic_matches_contract(): """A room channel's app segment is the frozen default topic.""" pytest.importorskip("slim_bindings") diff --git a/mycelium-cli/src/mycelium/cli.py b/mycelium-cli/src/mycelium/cli.py index 09fca3ed..564393f9 100644 --- a/mycelium-cli/src/mycelium/cli.py +++ b/mycelium-cli/src/mycelium/cli.py @@ -31,6 +31,7 @@ room, ui, user, + wire, ) app = typer.Typer( @@ -137,6 +138,10 @@ def skill() -> None: app.add_typer(demo.app, name="demo") app.add_typer(hub.app, name="hub") +# Hidden dev/testing plumbing — inject raw L9/SLIM traffic (see commands/wire.py). +app.add_typer(wire.l9_app, name="l9", hidden=True) +app.add_typer(wire.slim_app, name="slim", hidden=True) + if __name__ == "__main__": app() diff --git a/mycelium-cli/src/mycelium/commands/wire.py b/mycelium-cli/src/mycelium/commands/wire.py new file mode 100644 index 00000000..befa127b --- /dev/null +++ b/mycelium-cli/src/mycelium/commands/wire.py @@ -0,0 +1,205 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Mycelium Contributors + +"""``l9 send`` / ``slim send`` — hidden dev/testing plumbing. + +There's no way to put **L9 wire traffic** (or arbitrary SLIM messages) into a +room without running a full aligner-mediated negotiation. These two commands +are ``git cat-file``-style escape hatches for exercising the real path (SLIM +channel -> backend persister -> bus -> SSE) directly — QA'ing the frontend L9 +Inspector, demoing the AOP layer, reproducing protocol edge cases (odd +subkinds, deep ``parents`` chains, missing metrics). + +Deliberately **undocumented**: registered ``hidden=True`` and never +``@doc_ref``'d, so they never show up in ``mycelium --help`` or the generated +docs site. They bypass the aligner entirely and are for testing/demo only — +never a coordination shortcut. +""" + +from __future__ import annotations + +import asyncio +import json as json_module + +import typer + +from mycelium.commands.room import _resolve_room +from mycelium.config import MyceliumConfig +from mycelium.error_handler import print_error +from mycelium.slim import l9 +from mycelium.slim.client import SlimError +from mycelium.slim.member import DEFAULT_WORKSPACE, publish_once + +_BANNER = ( + "bypasses the aligner — real SLIM wire traffic for testing/demo only, " + "never a coordination shortcut" +) + +l9_app = typer.Typer(hidden=True, help="Inject L9 wire traffic into a room (dev/testing).") +slim_app = typer.Typer(hidden=True, help="Inject raw SLIM messages into a room (dev/testing).") + + +def _split_csv(raw: str | None) -> list[str]: + return [part.strip().lstrip("@") for part in raw.split(",") if part.strip()] if raw else [] + + +def _parse_json_object(raw: str | None, *, label: str) -> dict: + if raw is None: + return {} + try: + parsed = json_module.loads(raw) + except json_module.JSONDecodeError as exc: + typer.secho(f" ⟫ --{label} is not valid JSON: {exc}", fg=typer.colors.RED) + raise typer.Exit(2) from exc + if not isinstance(parsed, dict): + typer.secho(f" ⟫ --{label} must be a JSON object", fg=typer.colors.RED) + raise typer.Exit(2) + return parsed + + +def _run_publish( + config: MyceliumConfig, room: str, handle: str, payload: bytes, workspace: str | None +) -> None: + asyncio.run( + publish_once( + api_url=config.server.api_url, + node_endpoint=config.slim.node_endpoint, + room=room, + handle=handle, + payload=payload, + workspace=workspace or DEFAULT_WORKSPACE, + ) + ) + + +@l9_app.command("send") +def l9_send( + ctx: typer.Context, + room: str | None = typer.Option(None, "--room", "-r", help="Room (default: active room)"), + as_handle: str = typer.Option(..., "--as", "--handle", help="Sender handle to publish as"), + kind: str = typer.Option(..., "--kind", help=f"L9 kind ({', '.join(sorted(l9.VALID_KINDS))})"), + subkind: str | None = typer.Option(None, "--subkind", help="L9 subkind (kind-specific)"), + data: str | None = typer.Option(None, "--data", help="Payload data as a JSON object"), + text: str = typer.Option("", "--text", help="Human-facing text body"), + recipients: str | None = typer.Option(None, "--to", help="Comma-separated recipient handles"), + episode: str | None = typer.Option( + None, "--episode", help="Episode URN (default: the room's live episode)" + ), + parents: str | None = typer.Option( + None, "--parents", help="Comma-separated parent message ids" + ), + payload_type: str = typer.Option("data", "--payload-type", help="L9 payload.type"), + message_id: str | None = typer.Option(None, "--message-id", help="Explicit L9 message id"), + workspace: str | None = typer.Option( + None, "--workspace", help="SLIM workspace (default: the shared dev workspace)" + ), +) -> None: + """Publish a hand-crafted L9 envelope into a room as ``--as``, over the real SLIM wire. + + Built with the same envelope primitives every connector uses + (``mycelium.slim.l9``), so the wire shape matches + ``contracts/slim-l9-wire.json`` exactly. Kind/subkind are validated before + anything touches the wire. + + Example: + mycelium l9 send --room design --as @julia --kind commit --subkind resolved \\ + --data '{"assignments": {"cap": "30"}}' + """ + try: + l9.validate_kind(kind) + l9.validate_subkind(kind, subkind) + except l9.L9ValidationError as e: + typer.secho(f" ⟫ {e}", fg=typer.colors.RED) + raise typer.Exit(2) from e + + payload_data = _parse_json_object(data, label="data") + sender = as_handle.lstrip("@") + + try: + config = MyceliumConfig.load() + room_name = _resolve_room(config, room) + episode_urn = episode or l9.room_episode(room_name) + + content = l9.build_envelope_content( + kind=kind, + subkind=subkind, + sender=sender, + recipients=_split_csv(recipients), + episode=episode_urn, + parents=_split_csv(parents), + topic=l9.room_topic(room_name), + text=text, + message_id=message_id, + payload_type=payload_type, + payload_data=payload_data, + ) + + typer.secho(f" ⚠ {_BANNER}", fg=typer.colors.YELLOW) + _run_publish(config, room_name, sender, l9.serialize(content), workspace) + label = f"{kind}:{subkind}" if subkind else kind + typer.secho(f" ⟫ @{sender} → {room_name}: {label}", fg=typer.colors.GREEN) + except (typer.Exit, typer.Abort): + raise + except SlimError as e: + typer.secho(f" ⟫ {e}", fg=typer.colors.RED) + raise typer.Exit(1) from e + except Exception as e: + verbose = ctx.obj.get("verbose", False) if ctx.obj else False + print_error(e, verbose=verbose) + raise typer.Exit(1) from e + + +@slim_app.command("send") +def slim_send( + ctx: typer.Context, + room: str | None = typer.Option(None, "--room", "-r", help="Room (default: active room)"), + as_handle: str = typer.Option(..., "--as", "--handle", help="Sender handle to publish as"), + text: str | None = typer.Option(None, "--text", help="Raw text payload"), + json_payload: str | None = typer.Option(None, "--json", help="Raw JSON payload"), + workspace: str | None = typer.Option( + None, "--workspace", help="SLIM workspace (default: the shared dev workspace)" + ), +) -> None: + """Publish an arbitrary raw message onto a room's SLIM channel as ``--as``. + + No L9 semantics — the lowest-level escape hatch. Exercises the real channel: + other SLIM members and the moderator see it, and the persister decides + how/whether it surfaces. + + Example: + mycelium slim send --room design --as @julia --text "hello channel" + """ + if (text is None) == (json_payload is None): + typer.secho(" ⟫ pass exactly one of --text or --json", fg=typer.colors.RED) + raise typer.Exit(2) + + if json_payload is not None: + try: + json_module.loads(json_payload) + except json_module.JSONDecodeError as e: + typer.secho(f" ⟫ --json is not valid JSON: {e}", fg=typer.colors.RED) + raise typer.Exit(2) from e + payload = json_payload.encode("utf-8") + else: + payload = (text or "").encode("utf-8") + + sender = as_handle.lstrip("@") + + try: + config = MyceliumConfig.load() + room_name = _resolve_room(config, room) + + typer.secho(f" ⚠ {_BANNER}", fg=typer.colors.YELLOW) + _run_publish(config, room_name, sender, payload, workspace) + typer.secho( + f" ⟫ @{sender} → {room_name}: raw SLIM message published", fg=typer.colors.GREEN + ) + except (typer.Exit, typer.Abort): + raise + except SlimError as e: + typer.secho(f" ⟫ {e}", fg=typer.colors.RED) + raise typer.Exit(1) from e + except Exception as e: + verbose = ctx.obj.get("verbose", False) if ctx.obj else False + print_error(e, verbose=verbose) + raise typer.Exit(1) from e diff --git a/mycelium-cli/src/mycelium/slim/l9.py b/mycelium-cli/src/mycelium/slim/l9.py index bf3b2e69..0e2d1d20 100644 --- a/mycelium-cli/src/mycelium/slim/l9.py +++ b/mycelium-cli/src/mycelium/slim/l9.py @@ -51,6 +51,42 @@ KNOWLEDGE_KIND = "knowledge" +class L9ValidationError(ValueError): + """A hand-crafted envelope's kind/subkind falls outside the wire vocabulary.""" + + +# Kind -> allowed subkinds. Mirrors the backend's authoritative table +# (``app.services.l9.VALID_SUBKINDS``) byte-for-byte — see +# ``contracts/slim-l9-wire.json`` for the drift guard both suites assert +# against. An empty/None subkind is always valid, whatever the kind. +VALID_SUBKINDS: dict[str, frozenset[str]] = { + "knowledge": frozenset({"query", "distillation", "extraction", "feedback"}), + "commit": frozenset({"converged", "resolved", "rejected"}), + "intent": frozenset({"coordinator-assignment", "mission"}), + "exchange": frozenset({"team-formation"}), + "contingency": frozenset({"negotiation"}), +} + +VALID_KINDS: frozenset[str] = frozenset(VALID_SUBKINDS) + + +def validate_kind(kind: str) -> None: + """Reject a kind outside the L9 vocabulary.""" + if kind not in VALID_KINDS: + raise L9ValidationError(f"invalid kind {kind!r} (allowed: {sorted(VALID_KINDS)})") + + +def validate_subkind(kind: str, subkind: str | None) -> None: + """Reject a subkind outside the allowed table for ``kind`` (mirrors the backend).""" + if not subkind: + return + allowed = VALID_SUBKINDS.get(kind, frozenset()) + if subkind not in allowed: + raise L9ValidationError( + f"invalid subkind {subkind!r} for kind={kind} (allowed: {sorted(allowed)})" + ) + + def room_episode(room: str) -> str: """The room's live-episode URN — must match the backend's ``l9.episode_urn``.""" return f"urn:ioc:mycelium:episode:{room}:live" @@ -61,42 +97,49 @@ def room_topic(room: str) -> str: return f"urn:concept:mycelium:{room}" -def build_reply_content( +def build_envelope_content( *, + kind: str, + subkind: str | None = None, sender: str, - recipients: list[str], + recipients: list[str] | None = None, episode: str, - parents: list[str], + parents: list[str] | None = None, topic: str | None = None, text: str = "", message_id: str | None = None, - payload_type: str = "reply", + payload_type: str = "data", payload_data: dict[str, Any] | None = None, ) -> dict[str, Any]: - """Build a full content dict for an agent reply: ``{content, l9: }``. + """Build a full content dict for a hand-crafted envelope of any kind/subkind. - The envelope is an ``exchange`` (no subkind — always valid), with ``sender`` - as the first actor and ``recipients`` after it, parented on ``parents`` (the - message that woke the agent) so the backend's causal ordering + transcript - stay correct. + The general form: :func:`build_reply_content` is the ``exchange``-reply + specialization every connector uses; this is the escape hatch for crafting + anything else (a ``commit``, a ``knowledge`` push, an odd subkind) — the CLI's + ``mycelium l9 send`` plumbing. Raises :class:`L9ValidationError` before + touching the wire if ``subkind`` isn't valid for ``kind``. """ + validate_subkind(kind, subkind) + actors: list[dict[str, str]] = [{"id": sender, "role": "agent"}] - actors += [{"id": r, "role": "agent"} for r in recipients] + actors += [{"id": r, "role": "agent"} for r in (recipients or [])] header: dict[str, Any] = { "protocol": PROTOCOL, "subprotocol": SUBPROTOCOL, "version": VERSION, - "kind": EXCHANGE_KIND, - # ``participants.groups`` is required-but-nullable in the schema; the - # backend restores an explicit null after ``exclude_none``, so we mirror - # that here for a clean re-validation. - "participants": {"actors": actors, "groups": None}, - "message": { - "id": message_id or str(uuid.uuid4()), - "parents": list(parents), - "episode": episode, - }, + "kind": kind, + } + if subkind: + header["subkind"] = subkind + # ``participants.groups`` is required-but-nullable in the schema; the + # backend restores an explicit null after ``exclude_none``, so we mirror + # that here for a clean re-validation. + header["participants"] = {"actors": actors, "groups": None} + header["message"] = { + "id": message_id or str(uuid.uuid4()), + "parents": list(parents or []), + "episode": episode, } if topic: header["context"] = {"topic": topic} @@ -108,6 +151,39 @@ def build_reply_content( return {CONTENT_TEXT_KEY: text, CONTENT_L9_KEY: envelope} +def build_reply_content( + *, + sender: str, + recipients: list[str], + episode: str, + parents: list[str], + topic: str | None = None, + text: str = "", + message_id: str | None = None, + payload_type: str = "reply", + payload_data: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Build a full content dict for an agent reply: ``{content, l9: }``. + + The envelope is an ``exchange`` (no subkind — always valid), with ``sender`` + as the first actor and ``recipients`` after it, parented on ``parents`` (the + message that woke the agent) so the backend's causal ordering + transcript + stay correct. + """ + return build_envelope_content( + kind=EXCHANGE_KIND, + sender=sender, + recipients=recipients, + episode=episode, + parents=parents, + topic=topic, + text=text, + message_id=message_id, + payload_type=payload_type, + payload_data=payload_data, + ) + + def serialize(content: dict[str, Any]) -> bytes: """Encode a content dict to publishable bytes.""" return json.dumps(content).encode("utf-8") diff --git a/mycelium-cli/src/mycelium/slim/member.py b/mycelium-cli/src/mycelium/slim/member.py new file mode 100644 index 00000000..4c8e177a --- /dev/null +++ b/mycelium-cli/src/mycelium/slim/member.py @@ -0,0 +1,100 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Mycelium Contributors + +"""Ephemeral SLIM room membership — join, publish once, leave (dev/testing). + +The CLI's ``l9 send`` / ``slim send`` plumbing (see ``mycelium.commands.wire``) +needs to put real bytes on a room's SLIM channel so the backend's own +already-running persister sees them, records them to the durable transcript, and +republishes to its in-process bus (which feeds SSE). A separate process writing +straight to that bus can't reach it — ``bus`` is per-process, so a ``docker exec +python`` script publishing to its own copy of it never reaches the running +server's subscribers. Riding the real channel is the only way in from outside +the backend's process. + +The backend is every room's sole moderator (``app/services/room_channels.py``); +it never gets invited, it invites. So a short-lived CLI process has to ask the +backend to invite it before it can publish — the same join call a resident +connector makes on every reconnect (``POST /rooms/{room}/sessions``, +``app/routes/sessions.py::join_room`` → ``room_channels.manager.invite_in_background``). +That call is fire-and-forget on the backend side, with its own retrying +handshake, so asking before this process has even connected to the node is fine. +""" + +from __future__ import annotations + +import asyncio + +import httpx + +from mycelium.slim.client import SlimClient, SlimError +from mycelium.slim.naming import DEFAULT_WORKSPACE, SlimIdentity, node_reachable + +# How long to wait, after asking the backend to invite us, for the moderator's +# handshake to land and admit us into the encrypted group. Comfortably inside the +# backend's own invite retry budget (5 retries * 5s interval = 25s, see +# ``SlimClient._group_session_config``). +DEFAULT_JOIN_TIMEOUT_S = 30.0 + +# The join request is a same-host call to the backend, not the SLIM handshake +# itself, so it gets a short budget. +_HTTP_TIMEOUT_S = 10.0 + + +class SlimSendError(SlimError): + """A join/connect/publish attempt as an ephemeral room member failed.""" + + +async def announce_presence(api_url: str, room: str, handle: str) -> None: + """Ask the backend to (re)invite ``handle`` into ``room``'s SLIM group. + + Mirrors ``ParticipantCreate`` — the same request body a resident connector + sends on every join/reconnect. + """ + async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT_S) as client: + resp = await client.post( + f"{api_url}/api/rooms/{room}/sessions", json={"agent_handle": handle} + ) + resp.raise_for_status() + + +async def publish_once( + *, + api_url: str, + node_endpoint: str, + room: str, + handle: str, + payload: bytes, + workspace: str = DEFAULT_WORKSPACE, + join_timeout_s: float = DEFAULT_JOIN_TIMEOUT_S, +) -> None: + """Join ``room`` as ``handle`` over a live SLIM connection, publish once, leave. + + Connects a new SLIM app under ``workspace/room/handle``, asks the backend to + invite it in, waits to be admitted into the moderator's encrypted group, + publishes ``payload`` verbatim, and disconnects. Raises :class:`SlimSendError` + if the node is unreachable or the invite never lands within + ``join_timeout_s``; raises :class:`~mycelium.slim.client.SlimUnavailableError` + (via ``SlimClient.connect``) if ``slim_bindings`` has no wheel for this + platform — both are :class:`~mycelium.slim.client.SlimError`. + """ + if not node_reachable(node_endpoint): + raise SlimSendError( + f"SLIM node unreachable at {node_endpoint} — is `mycelium hub host` " + "(or the node's compose service) running?" + ) + + identity = SlimIdentity(workspace, room, handle) + client = await SlimClient(identity).connect(node_endpoint) + try: + await announce_presence(api_url, room, handle) + try: + session = await asyncio.wait_for(client.listen_for_session(), timeout=join_timeout_s) + except TimeoutError as exc: + raise SlimSendError( + f"timed out waiting to be invited into room {room!r} as @{handle} " + "— is the backend up and moderating this room?" + ) from exc + await SlimClient.publish(session, payload) + finally: + await client.close() diff --git a/mycelium-cli/tests/README.md b/mycelium-cli/tests/README.md index 173a0bc9..5c2ddbea 100644 --- a/mycelium-cli/tests/README.md +++ b/mycelium-cli/tests/README.md @@ -10,8 +10,8 @@ exactly one of each fake — don't re-declare them per file. | You're testing… | Use | It stands in for | | --- | --- | --- | | A command that calls the backend (`room`, `memory`, `plan`, …) | the `backend` fixture + patch the one generated `…​.sync` | the typed `mycelium_backend_client` plumbing | -| Connector / daemon HTTP (`announce_presence`, `reindex_after_knowledge`, briefing fetch) | the `fake_httpx` fixture, `FakeResp` | `httpx.AsyncClient` / `httpx.Client` | -| The member message stream / wake decision | `FakeSlimClient` (or the `fake_slim_client` fixture) | `slim.client.SlimClient` | +| Connector HTTP (`slim.member.announce_presence`, briefing fetch) | the `fake_httpx` fixture, `FakeResp` | `httpx.AsyncClient` / `httpx.Client` | +| Joining/publishing over SLIM (`slim.member.publish_once`, the `l9 send`/`slim send` plumbing) | `FakeSlimClient` (or the `fake_slim_client` fixture) | `slim.client.SlimClient` | | Anything that touches `~/.mycelium` | the `isolated_home` fixture | points `Path.home()` at a temp dir | ### Example — a command test (typed backend client) @@ -35,20 +35,23 @@ def test_list_rooms(backend, monkeypatch): from tests.conftest import FakeHTTPX, FakeResp async def test_announce(fake_httpx: FakeHTTPX): - await connector.announce_presence(cfg, "myroom", "agent-a") + await member.announce_presence("http://localhost:8000", "myroom", "agent-a") assert fake_httpx.calls == [("POST", ".../api/rooms/myroom/sessions", {"agent_handle": "agent-a"})] fake_httpx.respond_with(lambda *_: FakeResp(boom=True)) # exercise the error branch ``` -### Example — the member stream (`FakeSlimClient`) +### Example — joining + publishing over SLIM (`FakeSlimClient`) ```python from tests.conftest import FakeSlimClient -monkeypatch.setattr(member, "SlimClient", FakeSlimClient) -FakeSlimClient.inbox = [l9.serialize(tick), ...] # scripted inbound messages -content = await member.await_addressed(cfg, "r", "agent-a", timeout_s=2) +monkeypatch.setattr(member, "SlimClient", fake_slim_client) # the fixture, reset per test +await member.publish_once( + api_url="http://localhost:8000", node_endpoint="http://127.0.0.1:46357", + room="r", handle="agent-a", payload=l9.serialize(content), +) +assert FakeSlimClient.published == [l9.serialize(content)] ``` ## Commands diff --git a/mycelium-cli/tests/test_slim_l9.py b/mycelium-cli/tests/test_slim_l9.py index 28cb82b8..3b398883 100644 --- a/mycelium-cli/tests/test_slim_l9.py +++ b/mycelium-cli/tests/test_slim_l9.py @@ -12,6 +12,8 @@ from __future__ import annotations +import pytest + from mycelium.slim import l9 @@ -81,3 +83,89 @@ def test_no_topic_omits_context() -> None: ) assert "context" not in content["l9"]["header"] assert l9.topic_of(content) is None + + +# ── build_envelope_content — the generic builder behind `l9 send` ─────────── + + +def test_build_envelope_content_sets_kind_and_subkind() -> None: + content = l9.build_envelope_content( + kind="commit", + subkind="resolved", + sender="julia", + recipients=["bob"], + episode="ep-1", + parents=["p-1"], + payload_data={"assignments": {"cap": "30"}}, + ) + assert content["l9"]["header"]["kind"] == "commit" + assert content["l9"]["header"]["subkind"] == "resolved" + assert content["l9"]["header"]["message"]["parents"] == ["p-1"] + assert content["l9"]["payload"]["data"] == {"assignments": {"cap": "30"}} + + +def test_build_envelope_content_omits_subkind_when_absent() -> None: + content = l9.build_envelope_content(kind="exchange", sender="julia", episode="ep-1") + assert "subkind" not in content["l9"]["header"] + + +def test_build_envelope_content_rejects_invalid_subkind() -> None: + with pytest.raises(l9.L9ValidationError): + l9.build_envelope_content( + kind="exchange", subkind="not-a-real-subkind", sender="julia", episode="ep-1" + ) + + +def test_build_reply_content_matches_generic_builder_for_exchange() -> None: + """build_reply_content is exactly the exchange-kind specialization.""" + via_generic = l9.build_envelope_content( + kind=l9.EXCHANGE_KIND, + sender="a", + recipients=["b"], + episode="ep", + parents=["p"], + topic="t", + text="hi", + message_id="m-1", + payload_type="reply", + ) + via_reply = l9.build_reply_content( + sender="a", + recipients=["b"], + episode="ep", + parents=["p"], + topic="t", + text="hi", + message_id="m-1", + payload_type="reply", + ) + assert via_generic == via_reply + + +# ── kind/subkind validation ────────────────────────────────────────────────── + + +def test_validate_kind_accepts_known_kinds() -> None: + for kind in l9.VALID_KINDS: + l9.validate_kind(kind) # must not raise + + +def test_validate_kind_rejects_unknown_kind() -> None: + with pytest.raises(l9.L9ValidationError): + l9.validate_kind("not-a-kind") + + +def test_validate_subkind_none_always_valid() -> None: + for kind in l9.VALID_KINDS: + l9.validate_subkind(kind, None) # must not raise + + +def test_validate_subkind_accepts_allowed_pairs() -> None: + for kind, subkinds in l9.VALID_SUBKINDS.items(): + for subkind in subkinds: + l9.validate_subkind(kind, subkind) # must not raise + + +def test_validate_subkind_rejects_mismatched_pair() -> None: + with pytest.raises(l9.L9ValidationError): + l9.validate_subkind("exchange", "converged") # "converged" belongs to commit diff --git a/mycelium-cli/tests/test_slim_l9_wire.py b/mycelium-cli/tests/test_slim_l9_wire.py index 6b4ddc2d..a737793e 100644 --- a/mycelium-cli/tests/test_slim_l9_wire.py +++ b/mycelium-cli/tests/test_slim_l9_wire.py @@ -117,6 +117,15 @@ def test_knowledge_envelope_parses_to_contract_write(): assert data["updated_at"] == w["updated_at"] +def test_valid_subkinds_match_contract(): + """The CLI's kind/subkind vocabulary (used by the hidden `l9 send` plumbing) + is byte-for-byte the backend's ``app.services.l9.VALID_SUBKINDS`` table.""" + g = {k: v for k, v in _contract()["valid_subkinds"].items() if k != "_comment"} + assert set(l9.VALID_KINDS) == set(g) + for kind, allowed in g.items(): + assert l9.VALID_SUBKINDS[kind] == frozenset(allowed) + + def test_channel_name_topic_matches_contract(): """A room channel's app segment is the frozen default topic.""" pytest.importorskip("slim_bindings") diff --git a/mycelium-cli/tests/test_slim_member.py b/mycelium-cli/tests/test_slim_member.py new file mode 100644 index 00000000..542aef21 --- /dev/null +++ b/mycelium-cli/tests/test_slim_member.py @@ -0,0 +1,86 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Mycelium Contributors + +"""Unit tests for ``mycelium.slim.member`` — ephemeral join/publish/leave. + +Node-free: ``SlimClient`` is replaced with :class:`~tests.conftest.FakeSlimClient` +and ``httpx`` with the recording fakes, so these exercise the join → connect → +listen → publish → close sequence without a live SLIM node or backend. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from mycelium.slim import member +from tests.conftest import FakeHTTPX, FakeResp, FakeSlimClient + + +@pytest.fixture(autouse=True) +def _patch_slim_client(monkeypatch: pytest.MonkeyPatch, fake_slim_client: type[FakeSlimClient]): + monkeypatch.setattr(member, "SlimClient", fake_slim_client) + monkeypatch.setattr(member, "node_reachable", lambda _endpoint: True) + return fake_slim_client + + +async def test_publish_once_announces_then_publishes(fake_httpx: FakeHTTPX) -> None: + await member.publish_once( + api_url="http://localhost:8000", + node_endpoint="http://127.0.0.1:46357", + room="demo", + handle="julia", + payload=b"hello wire", + ) + + assert fake_httpx.calls == [ + ("POST", "http://localhost:8000/api/rooms/demo/sessions", {"agent_handle": "julia"}) + ] + assert FakeSlimClient.published == [b"hello wire"] + + +def test_publish_once_raises_when_node_unreachable( + monkeypatch: pytest.MonkeyPatch, fake_httpx: FakeHTTPX +) -> None: + monkeypatch.setattr(member, "node_reachable", lambda _endpoint: False) + + with pytest.raises(member.SlimSendError, match="unreachable"): + asyncio.run( + member.publish_once( + api_url="http://localhost:8000", + node_endpoint="http://127.0.0.1:46357", + room="demo", + handle="julia", + payload=b"x", + ) + ) + assert fake_httpx.calls == [] # never got as far as announcing + + +def test_publish_once_raises_on_join_timeout( + monkeypatch: pytest.MonkeyPatch, fake_httpx: FakeHTTPX +) -> None: + async def _hang(self): # noqa: ANN001 - matches FakeSlimClient.listen_for_session + await asyncio.sleep(10) + return "unreachable-session" + + monkeypatch.setattr(FakeSlimClient, "listen_for_session", _hang) + + with pytest.raises(member.SlimSendError, match="timed out"): + asyncio.run( + member.publish_once( + api_url="http://localhost:8000", + node_endpoint="http://127.0.0.1:46357", + room="demo", + handle="julia", + payload=b"x", + join_timeout_s=0.05, + ) + ) + + +async def test_announce_presence_raises_on_http_error(fake_httpx: FakeHTTPX) -> None: + fake_httpx.respond_with(lambda *_a: FakeResp(boom=True, status_code=500)) + with pytest.raises(RuntimeError): + await member.announce_presence("http://localhost:8000", "demo", "julia") diff --git a/mycelium-cli/tests/test_wire_cli.py b/mycelium-cli/tests/test_wire_cli.py new file mode 100644 index 00000000..ca67a2bf --- /dev/null +++ b/mycelium-cli/tests/test_wire_cli.py @@ -0,0 +1,172 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Mycelium Contributors + +"""``mycelium l9 send`` / ``mycelium slim send`` — hidden dev/testing plumbing. + +Node-free: ``wire.publish_once`` (the SLIM join/publish primitive) is stubbed so +these exercise the CLI plumbing — hidden-from-help registration, kind/subkind +validation *before* anything is published, envelope construction, and the +``slim send`` text/json exclusivity check. +""" + +from __future__ import annotations + +import json + +import pytest +from typer.testing import CliRunner + +from mycelium.cli import app +from mycelium.commands import wire +from mycelium.config import MyceliumConfig +from mycelium.slim import l9 +from mycelium.slim.member import SlimSendError + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def _stub_config(monkeypatch: pytest.MonkeyPatch) -> None: + fake_config = MyceliumConfig() + fake_config.rooms.active = "demo" + monkeypatch.setattr(wire.MyceliumConfig, "load", classmethod(lambda _cls: fake_config)) + + +def test_l9_and_slim_are_hidden_from_top_level_help() -> None: + result = runner.invoke(app, ["--help"]) + assert result.exit_code == 0 + assert " l9 " not in result.output + assert " slim " not in result.output + + +def test_l9_send_rejects_invalid_kind_before_publishing(monkeypatch: pytest.MonkeyPatch) -> None: + called = False + + def _fake_publish(*_a, **_k): + nonlocal called + called = True + + monkeypatch.setattr(wire, "publish_once", _fake_publish) + + result = runner.invoke(app, ["l9", "send", "--as", "julia", "--kind", "not-a-kind"]) + assert result.exit_code == 2 + assert not called + + +def test_l9_send_rejects_mismatched_subkind_before_publishing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + called = False + + def _fake_publish(*_a, **_k): + nonlocal called + called = True + + monkeypatch.setattr(wire, "publish_once", _fake_publish) + + result = runner.invoke( + app, + ["l9", "send", "--as", "julia", "--kind", "exchange", "--subkind", "converged"], + ) + assert result.exit_code == 2 + assert not called + + +def test_l9_send_builds_envelope_and_publishes(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + async def _fake_publish(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr(wire, "publish_once", _fake_publish) + + result = runner.invoke( + app, + [ + "l9", + "send", + "--as", + "@julia", + "--kind", + "commit", + "--subkind", + "resolved", + "--to", + "bob, @selina", + "--parents", + "p1,p2", + "--data", + '{"assignments": {"cap": "30"}}', + ], + ) + assert result.exit_code == 0, result.output + assert captured["room"] == "demo" + assert captured["handle"] == "julia" + + content = l9.parse(captured["payload"]) + assert content is not None + assert l9.kind_of(content) == "commit" + assert content["l9"]["header"]["subkind"] == "resolved" + assert content["l9"]["header"]["message"]["parents"] == ["p1", "p2"] + assert l9.recipients_of(content) == ["bob", "selina"] + assert l9.payload_data_of(content) == {"assignments": {"cap": "30"}} + + +def test_l9_send_rejects_non_object_data(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wire, "publish_once", lambda **_k: None) + result = runner.invoke( + app, + ["l9", "send", "--as", "julia", "--kind", "exchange", "--data", "[1, 2, 3]"], + ) + assert result.exit_code == 2 + + +def test_l9_send_surfaces_slim_send_error(monkeypatch: pytest.MonkeyPatch) -> None: + async def _boom(**_k): + raise SlimSendError("SLIM node unreachable at http://x") + + monkeypatch.setattr(wire, "publish_once", _boom) + result = runner.invoke(app, ["l9", "send", "--as", "julia", "--kind", "exchange"]) + assert result.exit_code == 1 + assert "unreachable" in result.output + + +def test_slim_send_requires_exactly_one_of_text_or_json() -> None: + result = runner.invoke(app, ["slim", "send", "--as", "julia"]) + assert result.exit_code == 2 + + result = runner.invoke(app, ["slim", "send", "--as", "julia", "--text", "hi", "--json", "{}"]) + assert result.exit_code == 2 + + +def test_slim_send_publishes_raw_text(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + async def _fake_publish(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr(wire, "publish_once", _fake_publish) + + result = runner.invoke(app, ["slim", "send", "--as", "@julia", "--text", "hello channel"]) + assert result.exit_code == 0, result.output + assert captured["handle"] == "julia" + assert captured["room"] == "demo" + assert captured["payload"] == b"hello channel" + + +def test_slim_send_publishes_raw_json(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + async def _fake_publish(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr(wire, "publish_once", _fake_publish) + + result = runner.invoke(app, ["slim", "send", "--as", "julia", "--json", '{"anything": true}']) + assert result.exit_code == 0, result.output + assert json.loads(captured["payload"]) == {"anything": True} + + +def test_slim_send_rejects_invalid_json() -> None: + result = runner.invoke(app, ["slim", "send", "--as", "julia", "--json", "{not json"]) + assert result.exit_code == 2