Add MatrixTransport (A1) — step-2 PR B
- matrix.py: Transport implementation over the Matrix client-server HTTP API directly (token auth, /sync long-poll, /send) using the existing httpx dependency — no Matrix SDK; four endpoints total. Deviation from the presented plan (matrix-nio) flagged in the PR - First sync learns position and skips backlog, so room history is never delivered as context (A1); own messages and non-message events filtered; sender MXID passed through for the orchestrator's operator allowlist - Sync loop backs off (capped at 60s) and retries indefinitely on homeserver errors, mirroring B2's polling posture - Encrypted rooms unsupported: bot rooms are created unencrypted until E2E lands (operator decision; room recreated accordingly) - NISSE_MATRIX_TEST_ROOM_ID added: the live smoke test binds only to the dedicated test room, never the production room - 8 unit tests over a mocked HTTP layer; live integration test verified against the real homeserver Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Fable 5 committed
1 parent
5dad5cf90f
commit
5db6d9d96e
4 files changed
+380
No files matched your search
@@ -16,4 +16,6 @@ NISSE_MATRIX_HOMESERVER=https://matrix.example
|
||||
NISSE_MATRIX_USER_ID=@nisse-dev:example
|
||||
NISSE_MATRIX_TOKEN=syt_your_matrix_access_token
|
||||
NISSE_MATRIX_ROOM_ID=!roomid:example
|
||||
# Dedicated room for the live integration test; never the production room.
|
||||
NISSE_MATRIX_TEST_ROOM_ID=!testroomid:example
|
||||
NISSE_MATRIX_OPERATORS=@you:example
|
||||
@@ -0,0 +1,167 @@
|
||||
"""MatrixTransport (spec A1): Matrix as strictly a UI channel.
|
||||
|
||||
Implements the Transport interface directly over the Matrix client-server HTTP
|
||||
API (token auth, /sync long-poll, /send) with the httpx client already in the
|
||||
dependency tree — no Matrix SDK. The surface used is four endpoints; an SDK
|
||||
earns its keep when E2E arrives, and bot rooms are created unencrypted until
|
||||
then, so encrypted rooms are not supported here.
|
||||
|
||||
Inbound handling follows B2's polling posture: on homeserver errors the sync
|
||||
loop backs off (capped) and retries indefinitely — an outage delays reactions
|
||||
but never fails anything.
|
||||
"""
|
||||
|
||||
import os
|
||||
import threading
|
||||
import uuid
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass
|
||||
from typing import ClassVar
|
||||
|
||||
import httpx
|
||||
|
||||
from nisse_dev.transport import InboundMessage
|
||||
|
||||
|
||||
class TransportError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MatrixConfig:
|
||||
homeserver: str
|
||||
user_id: str
|
||||
token: str
|
||||
room_id: str
|
||||
|
||||
_ENV: ClassVar[tuple[str, ...]] = (
|
||||
"NISSE_MATRIX_HOMESERVER",
|
||||
"NISSE_MATRIX_USER_ID",
|
||||
"NISSE_MATRIX_TOKEN",
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_env(
|
||||
cls,
|
||||
env: Mapping[str, str] = os.environ,
|
||||
room_var: str = "NISSE_MATRIX_ROOM_ID",
|
||||
) -> "MatrixConfig":
|
||||
needed = (*cls._ENV, room_var)
|
||||
missing = [var for var in needed if not env.get(var)]
|
||||
if missing:
|
||||
raise TransportError(f"matrix config incomplete; set: {', '.join(missing)}")
|
||||
return cls(
|
||||
homeserver=env["NISSE_MATRIX_HOMESERVER"].rstrip("/"),
|
||||
user_id=env["NISSE_MATRIX_USER_ID"],
|
||||
token=env["NISSE_MATRIX_TOKEN"],
|
||||
room_id=env[room_var],
|
||||
)
|
||||
|
||||
|
||||
class MatrixTransport:
|
||||
def __init__(
|
||||
self,
|
||||
config: MatrixConfig,
|
||||
*,
|
||||
poll_timeout_ms: int = 30000,
|
||||
http_transport: httpx.BaseTransport | None = None,
|
||||
) -> None:
|
||||
self._config = config
|
||||
self._poll_timeout_ms = poll_timeout_ms
|
||||
self._http = httpx.Client(
|
||||
base_url=f"{config.homeserver}/_matrix/client/v3",
|
||||
headers={"Authorization": f"Bearer {config.token}"},
|
||||
timeout=(poll_timeout_ms / 1000) + 30,
|
||||
transport=http_transport,
|
||||
)
|
||||
self._callback: Callable[[InboundMessage], None] | None = None
|
||||
self._stop = threading.Event()
|
||||
self._thread: threading.Thread | None = None
|
||||
|
||||
# -- outbound ---------------------------------------------------------
|
||||
|
||||
def send_message(self, text: str) -> None:
|
||||
self._send(text)
|
||||
|
||||
def send_approval_request(self, text: str) -> None:
|
||||
self._send(text)
|
||||
|
||||
def ask_question(self, text: str) -> None:
|
||||
self._send(text)
|
||||
|
||||
def _send(self, text: str) -> None:
|
||||
room = self._config.room_id
|
||||
txn = uuid.uuid4().hex
|
||||
try:
|
||||
response = self._http.put(
|
||||
f"/rooms/{room}/send/m.room.message/{txn}",
|
||||
json={"msgtype": "m.text", "body": text},
|
||||
)
|
||||
except httpx.HTTPError as exc:
|
||||
raise TransportError(f"matrix send failed: {exc}") from exc
|
||||
if response.status_code != 200:
|
||||
raise TransportError(
|
||||
f"matrix send returned {response.status_code}: {response.text[:300]}"
|
||||
)
|
||||
|
||||
# -- inbound ----------------------------------------------------------
|
||||
|
||||
def on_message(self, callback: Callable[[InboundMessage], None]) -> None:
|
||||
self._callback = callback
|
||||
|
||||
def start(self) -> None:
|
||||
"""Begin delivering new room messages to the registered callback.
|
||||
History from before start() is never delivered (Matrix room history
|
||||
must not become load-bearing context, A1)."""
|
||||
if self._callback is None:
|
||||
raise TransportError("no message handler registered; call on_message first")
|
||||
self._stop.clear()
|
||||
self._thread = threading.Thread(target=self._sync_loop, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop.set()
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=(self._poll_timeout_ms / 1000) + 5)
|
||||
self._thread = None
|
||||
|
||||
def _sync_loop(self) -> None:
|
||||
since: str | None = None
|
||||
backoff = 1.0
|
||||
while not self._stop.is_set():
|
||||
params: dict[str, str] = {}
|
||||
if since is None:
|
||||
params["timeout"] = "0" # first sync: learn position, skip backlog
|
||||
else:
|
||||
params["timeout"] = str(self._poll_timeout_ms)
|
||||
params["since"] = since
|
||||
try:
|
||||
response = self._http.get("/sync", params=params)
|
||||
response.raise_for_status()
|
||||
except httpx.HTTPError:
|
||||
self._stop.wait(backoff)
|
||||
backoff = min(backoff * 2, 60.0)
|
||||
continue
|
||||
backoff = 1.0
|
||||
data = response.json()
|
||||
if since is not None:
|
||||
self._deliver(data)
|
||||
since = data.get("next_batch", since)
|
||||
|
||||
def _deliver(self, sync_data: dict) -> None:
|
||||
room = sync_data.get("rooms", {}).get("join", {}).get(self._config.room_id, {})
|
||||
for event in room.get("timeline", {}).get("events", []):
|
||||
if event.get("type") != "m.room.message":
|
||||
continue
|
||||
sender = event.get("sender", "")
|
||||
if sender == self._config.user_id:
|
||||
continue
|
||||
body = event.get("content", {}).get("body")
|
||||
if not body:
|
||||
continue
|
||||
assert self._callback is not None
|
||||
self._callback(InboundMessage(sender=sender, text=body))
|
||||
|
||||
def close(self) -> None:
|
||||
self.stop()
|
||||
self._http.close()
|
||||
@@ -0,0 +1,162 @@
|
||||
import json
|
||||
import threading
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from nisse_dev.matrix import MatrixConfig, MatrixTransport, TransportError
|
||||
from nisse_dev.transport import InboundMessage
|
||||
|
||||
CONFIG = MatrixConfig(
|
||||
homeserver="https://matrix.test",
|
||||
user_id="@nisse-dev:test",
|
||||
token="syt_secret",
|
||||
room_id="!room:test",
|
||||
)
|
||||
|
||||
|
||||
def sync_body(events: list[dict], next_batch: str) -> dict:
|
||||
return {
|
||||
"next_batch": next_batch,
|
||||
"rooms": {"join": {"!room:test": {"timeline": {"events": events}}}},
|
||||
}
|
||||
|
||||
|
||||
def message_event(sender: str, body: str) -> dict:
|
||||
return {
|
||||
"type": "m.room.message",
|
||||
"sender": sender,
|
||||
"content": {"msgtype": "m.text", "body": body},
|
||||
}
|
||||
|
||||
|
||||
def test_send_message_puts_event_with_auth_and_unique_txn() -> None:
|
||||
seen: list[httpx.Request] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request)
|
||||
return httpx.Response(200, json={"event_id": "$e"})
|
||||
|
||||
transport = MatrixTransport(CONFIG, http_transport=httpx.MockTransport(handler))
|
||||
transport.send_message("hello")
|
||||
transport.send_approval_request("plan rev-1")
|
||||
|
||||
first, second = seen
|
||||
assert first.headers["authorization"] == "Bearer syt_secret"
|
||||
assert "/rooms/!room:test/send/m.room.message/" in str(first.url)
|
||||
assert json.loads(first.content) == {"msgtype": "m.text", "body": "hello"}
|
||||
assert str(first.url) != str(second.url) # txn ids must be unique
|
||||
|
||||
|
||||
def test_send_failure_raises_transport_error() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(403, text="forbidden")
|
||||
|
||||
transport = MatrixTransport(CONFIG, http_transport=httpx.MockTransport(handler))
|
||||
with pytest.raises(TransportError, match="403"):
|
||||
transport.send_message("hello")
|
||||
|
||||
|
||||
class SyncScript:
|
||||
"""Serves scripted /sync responses, then endless empty ones."""
|
||||
|
||||
def __init__(self, responses: list[dict]) -> None:
|
||||
self._responses = responses
|
||||
self._i = 0
|
||||
self.batches_seen: list[str | None] = []
|
||||
|
||||
def __call__(self, request: httpx.Request) -> httpx.Response:
|
||||
assert request.url.path.endswith("/sync")
|
||||
self.batches_seen.append(request.url.params.get("since"))
|
||||
if self._i < len(self._responses):
|
||||
body = self._responses[self._i]
|
||||
self._i += 1
|
||||
return httpx.Response(200, json=body)
|
||||
return httpx.Response(200, json=sync_body([], f"b{self._i}"))
|
||||
|
||||
|
||||
def collect_messages(
|
||||
responses: list[dict], expected: int, timeout_s: float = 5.0
|
||||
) -> list[InboundMessage]:
|
||||
received: list[InboundMessage] = []
|
||||
done = threading.Event()
|
||||
|
||||
def on_message(msg: InboundMessage) -> None:
|
||||
received.append(msg)
|
||||
if len(received) >= expected:
|
||||
done.set()
|
||||
|
||||
transport = MatrixTransport(
|
||||
CONFIG,
|
||||
poll_timeout_ms=10,
|
||||
http_transport=httpx.MockTransport(SyncScript(responses)),
|
||||
)
|
||||
transport.on_message(on_message)
|
||||
transport.start()
|
||||
done.wait(timeout_s)
|
||||
transport.close()
|
||||
return received
|
||||
|
||||
|
||||
def test_listener_delivers_operator_messages() -> None:
|
||||
responses = [
|
||||
sync_body([message_event("@don:test", "backlog: never delivered")], "b1"),
|
||||
sync_body([message_event("@don:test", "approve")], "b2"),
|
||||
]
|
||||
received = collect_messages(responses, expected=1)
|
||||
assert received == [InboundMessage(sender="@don:test", text="approve")]
|
||||
|
||||
|
||||
def test_own_and_non_message_events_are_filtered() -> None:
|
||||
responses = [
|
||||
sync_body([], "b1"),
|
||||
sync_body(
|
||||
[
|
||||
message_event("@nisse-dev:test", "my own echo"),
|
||||
{"type": "m.room.member", "sender": "@don:test", "content": {}},
|
||||
message_event("@don:test", "status"),
|
||||
],
|
||||
"b2",
|
||||
),
|
||||
]
|
||||
received = collect_messages(responses, expected=1)
|
||||
assert received == [InboundMessage(sender="@don:test", text="status")]
|
||||
|
||||
|
||||
def test_first_sync_learns_position_without_delivering() -> None:
|
||||
script = SyncScript([sync_body([message_event("@don:test", "old")], "b1")])
|
||||
transport = MatrixTransport(
|
||||
CONFIG, poll_timeout_ms=10, http_transport=httpx.MockTransport(script)
|
||||
)
|
||||
got = threading.Event()
|
||||
transport.on_message(lambda msg: got.set())
|
||||
transport.start()
|
||||
assert not got.wait(0.3)
|
||||
transport.close()
|
||||
assert script.batches_seen[0] is None # initial sync carries no since
|
||||
assert "b1" in script.batches_seen # later polls resume from its batch
|
||||
|
||||
|
||||
def test_start_without_handler_raises() -> None:
|
||||
transport = MatrixTransport(
|
||||
CONFIG, http_transport=httpx.MockTransport(lambda r: httpx.Response(200))
|
||||
)
|
||||
with pytest.raises(TransportError, match="no message handler"):
|
||||
transport.start()
|
||||
|
||||
|
||||
def test_config_from_env_reports_missing() -> None:
|
||||
with pytest.raises(TransportError, match="NISSE_MATRIX_TOKEN"):
|
||||
MatrixConfig.from_env({"NISSE_MATRIX_HOMESERVER": "https://m"})
|
||||
|
||||
|
||||
def test_config_from_env_custom_room_var() -> None:
|
||||
env = {
|
||||
"NISSE_MATRIX_HOMESERVER": "https://m/",
|
||||
"NISSE_MATRIX_USER_ID": "@nisse-dev:test",
|
||||
"NISSE_MATRIX_TOKEN": "t",
|
||||
"NISSE_MATRIX_TEST_ROOM_ID": "!t:test",
|
||||
}
|
||||
config = MatrixConfig.from_env(env, room_var="NISSE_MATRIX_TEST_ROOM_ID")
|
||||
assert config.room_id == "!t:test"
|
||||
assert config.homeserver == "https://m" # trailing slash stripped
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Live smoke test against the real homeserver, bound strictly to the
|
||||
dedicated test room (never the production room). Skips unless the Matrix env
|
||||
vars, including NISSE_MATRIX_TEST_ROOM_ID, are present."""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from nisse_dev.envfile import load_env_file
|
||||
from nisse_dev.matrix import MatrixConfig, MatrixTransport
|
||||
|
||||
load_env_file(Path(__file__).resolve().parents[2] / ".env")
|
||||
|
||||
REQUIRED = (
|
||||
"NISSE_MATRIX_HOMESERVER",
|
||||
"NISSE_MATRIX_USER_ID",
|
||||
"NISSE_MATRIX_TOKEN",
|
||||
"NISSE_MATRIX_TEST_ROOM_ID",
|
||||
)
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.integration,
|
||||
pytest.mark.skipif(
|
||||
not all(os.environ.get(var) for var in REQUIRED),
|
||||
reason=f"live matrix config absent; set {', '.join(REQUIRED)}",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def test_send_lands_in_test_room() -> None:
|
||||
config = MatrixConfig.from_env(room_var="NISSE_MATRIX_TEST_ROOM_ID")
|
||||
marker = f"transport smoke test {uuid.uuid4().hex}"
|
||||
transport = MatrixTransport(config)
|
||||
try:
|
||||
transport.send_message(marker)
|
||||
response = transport._http.get(
|
||||
f"/rooms/{config.room_id}/messages",
|
||||
params={"dir": "b", "limit": "10"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
bodies = [
|
||||
event.get("content", {}).get("body")
|
||||
for event in response.json().get("chunk", [])
|
||||
]
|
||||
assert marker in bodies
|
||||
finally:
|
||||
transport.close()
|
||||
Reference in new issue
Block a user