From 5427a6176824e769e59f081bdc99b844f3eefe02 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 28 Apr 2026 17:05:31 -0400 Subject: [PATCH] mesh: 'serve' and 'sync' CLI verbs over the wire layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the gossip-wire core (commit f141bab) into the user-facing CLI: aborist mesh serve --host 127.0.0.1 --port 8400 Spawns MeshWireServer, blocks until SIGINT. Refuses if mesh isn't initialized or .enabled is off — same gate every other mesh verb enforces. Stdout is unbuffered JSON status lines. aborist mesh sync --peer http://other.example:8400 [--limit N] Enumerates local documents (most-recent N, default 100), fires one ANNOUNCE_ROOT per document via MeshWireClient, reports counts of acknowledged vs. errored. Pulling missing roots back from the peer is a v2 addition — this verb pushes only. Tests (6): gating refusals on uninitialized + disabled mesh; sync round-trip with two peers proving Alice's audit chain gains one 'mesh_received' per Bob announce; unreachable-peer reporting; serve smoke-test via start_in_thread + GET /mesh/info. --- aborist/cli.py | 106 +++++++++++++++++++++ tests/test_mesh_cli.py | 203 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 309 insertions(+) create mode 100644 tests/test_mesh_cli.py diff --git a/aborist/cli.py b/aborist/cli.py index 81d87c6..1a383db 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -1592,6 +1592,95 @@ def _cmd_mesh_rotate(args: argparse.Namespace) -> int: return 0 +def _cmd_mesh_serve(args: argparse.Namespace) -> int: + """Run the HTTP gossip server until SIGINT.""" + from aborist.mesh import is_enabled, load_identity + from aborist.mesh.wire import MeshWireServer + + conn = connect(args.db) + try: + if load_identity(conn) is None: + print("error: mesh not initialized; run 'aborist mesh init' first", file=sys.stderr) + return 2 + if not is_enabled(conn): + print("error: mesh.enabled is off; run 'aborist mesh enable' first", file=sys.stderr) + return 2 + finally: + conn.close() + + srv = MeshWireServer(args.db, host=args.host, port=args.port) + print(json.dumps({"status": "serving", "url": srv.url, "db": str(args.db)})) + sys.stdout.flush() + try: + srv.serve() + except KeyboardInterrupt: + print(json.dumps({"status": "stopped", "reason": "SIGINT"})) + finally: + srv.stop() + return 0 + + +def _cmd_mesh_sync(args: argparse.Namespace) -> int: + """Send ANNOUNCE_ROOT for local documents to a peer. + + v1: enumerates local document_roots (limited by --limit) and announces + each. Receives one ack per announce. Pulling missing roots back from the + peer is a future addition; this verb currently only pushes. + """ + from aborist.mesh import is_enabled, load_identity + from aborist.mesh.wire import MeshWireClient + + conn = connect(args.db) + try: + if load_identity(conn) is None: + print("error: mesh not initialized", file=sys.stderr) + return 2 + if not is_enabled(conn): + print("error: mesh.enabled is off", file=sys.stderr) + return 2 + rows = conn.execute( + "SELECT document_root, document_uri, chunking_version, " + " canonicalization_version, schema_version " + "FROM documents ORDER BY rowid DESC LIMIT ?", + (args.limit,), + ).fetchall() + finally: + conn.close() + + sent: list[dict] = [] + errors: list[dict] = [] + with MeshWireClient(args.db, args.peer) as client: + try: + peer_info = client.info() + except Exception as e: + print(json.dumps({"status": "peer_unreachable", "peer": args.peer, "error": str(e)}, indent=2)) + return 2 + for r in rows: + try: + resp = client.announce_root( + document_root=r["document_root"], + source_uri=r["document_uri"], + chunking_version=r["chunking_version"], + canonicalization_version=r["canonicalization_version"], + schema_version=r["schema_version"], + ) + sent.append({"document_root": r["document_root"], "ack": resp}) + except Exception as e: + errors.append({"document_root": r["document_root"], "error": str(e)}) + + print(json.dumps({ + "status": "synced", + "peer": args.peer, + "peer_member_id": peer_info.get("member_id"), + "peer_epoch": peer_info.get("current_epoch"), + "announced": len(sent), + "errors": len(errors), + "sent": sent[: args.verbose], + "error_samples": errors[:5], + }, indent=2)) + return 0 if not errors else 1 + + def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="aborist", @@ -2104,6 +2193,23 @@ def build_parser() -> argparse.ArgumentParser: mesh_rotate.add_argument("--reason", default="scheduled") mesh_rotate.set_defaults(func=_cmd_mesh_rotate) + mesh_serve = mesh_sub.add_parser( + "serve", + help="run the HTTP gossip server (blocks until SIGINT)", + ) + mesh_serve.add_argument("--host", default="127.0.0.1", help="bind host (default: 127.0.0.1)") + mesh_serve.add_argument("--port", type=int, default=8400, help="bind port (default: 8400)") + mesh_serve.set_defaults(func=_cmd_mesh_serve) + + mesh_sync = mesh_sub.add_parser( + "sync", + help="announce local document_roots to a peer's gossip server", + ) + mesh_sync.add_argument("--peer", required=True, help="peer URL, e.g. http://other.example.com:8400") + mesh_sync.add_argument("--limit", type=int, default=100, help="announce at most N most-recent docs (default: 100)") + mesh_sync.add_argument("--verbose", type=int, default=10, help="include this many ack details in output (default: 10)") + mesh_sync.set_defaults(func=_cmd_mesh_sync) + return p diff --git a/tests/test_mesh_cli.py b/tests/test_mesh_cli.py new file mode 100644 index 0000000..1e4c3d5 --- /dev/null +++ b/tests/test_mesh_cli.py @@ -0,0 +1,203 @@ +"""CLI integration: `aborist mesh serve` and `aborist mesh sync`. + +Wire-layer protocol behavior is covered in test_mesh_wire.py and +test_mesh_wire_e2e.py — these tests just exercise the argparse plumbing, +the gating checks (mesh initialized + enabled), and the round-trip +through MeshWireClient that the sync verb performs. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from aborist.cli import build_parser +from aborist.document import Document +from aborist.ingest import ingest_source +from aborist.mesh import init_identity, set_enabled +from aborist.mesh.members import add_member +from aborist.mesh.wire import MeshWireServer +from aborist.store import connect + + +class _OneDocSource: + source_type = "mesh_cli_test" + + def __init__(self, uri: str, content: str, title: str): + self._doc = Document( + uri=uri, content=content, source_type=self.source_type, title=title + ) + + def iter_documents(self): + yield self._doc + + +def _ingest(db_path, *, uri: str, content: str, title: str) -> str: + conn = connect(db_path) + try: + ingest_source(conn, _OneDocSource(uri, content, title)) + row = conn.execute( + "SELECT document_root FROM documents WHERE document_uri=?", (uri,) + ).fetchone() + finally: + conn.close() + return row["document_root"] + + +def _bootstrap_two_peers(tmp_path: Path): + alice_db = tmp_path / "alice.db" + bob_db = tmp_path / "bob.db" + a_conn = connect(alice_db) + try: + alice = init_identity(a_conn, group_name="t", member_id="alice") + set_enabled(a_conn, True) + finally: + a_conn.close() + b_conn = connect(bob_db) + try: + bob = init_identity(b_conn, group_name="t", member_id="bob") + set_enabled(b_conn, True) + finally: + b_conn.close() + a_conn = connect(alice_db) + try: + add_member(a_conn, member_id="bob", sign_pub=bob.sign_pub, dh_pub=bob.dh_pub) + finally: + a_conn.close() + b_conn = connect(bob_db) + try: + add_member(b_conn, member_id="alice", sign_pub=alice.sign_pub, dh_pub=alice.dh_pub) + finally: + b_conn.close() + return alice_db, bob_db, alice, bob + + +# --------------------------------------------------------------------------- +# Gating: mesh must be initialized AND enabled +# --------------------------------------------------------------------------- + + +def test_serve_refuses_when_mesh_not_initialized(tmp_path, capsys): + db = tmp_path / "x.db" + parser = build_parser() + args = parser.parse_args(["--db", str(db), "mesh", "serve", "--port", "0"]) + rc = args.func(args) + assert rc == 2 + assert "not initialized" in capsys.readouterr().err + + +def test_serve_refuses_when_mesh_disabled(tmp_path, capsys): + db = tmp_path / "x.db" + conn = connect(db) + try: + init_identity(conn, group_name="t", member_id="alice") + # Don't call set_enabled — flag stays off by default. + finally: + conn.close() + parser = build_parser() + args = parser.parse_args(["--db", str(db), "mesh", "serve", "--port", "0"]) + rc = args.func(args) + assert rc == 2 + assert "enabled" in capsys.readouterr().err + + +def test_sync_refuses_when_mesh_disabled(tmp_path, capsys): + db = tmp_path / "x.db" + conn = connect(db) + try: + init_identity(conn, group_name="t", member_id="alice") + finally: + conn.close() + parser = build_parser() + args = parser.parse_args(["--db", str(db), "mesh", "sync", "--peer", "http://0:0"]) + rc = args.func(args) + assert rc == 2 + + +# --------------------------------------------------------------------------- +# sync: announces local document_roots to a peer +# --------------------------------------------------------------------------- + + +def test_sync_announces_local_roots(tmp_path, capsys): + """Bob's CLI sync sends ANNOUNCE_ROOT for each local doc to Alice's + server; Alice's audit chain records each receive.""" + alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path) + + # Bob ingests two documents locally — these are what he'll announce. + _ingest(bob_db, uri="https://example.com/a", content="Doc A content here.", title="A") + _ingest(bob_db, uri="https://example.com/b", content="Doc B content here.", title="B") + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + srv.start_in_thread() + try: + parser = build_parser() + args = parser.parse_args([ + "--db", str(bob_db), "mesh", "sync", + "--peer", srv.url, + "--limit", "100", + ]) + rc = args.func(args) + assert rc == 0 + finally: + srv.stop() + + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "synced" + assert payload["peer_member_id"] == "alice" + assert payload["announced"] == 2 + assert payload["errors"] == 0 + + # Alice's chain gained one mesh_received event per announce. + a_conn = connect(alice_db) + try: + rcvd = a_conn.execute( + "SELECT COUNT(*) FROM audit_events WHERE event_type='mesh_received'" + ).fetchone()[0] + finally: + a_conn.close() + assert rcvd == 2 + + +def test_sync_reports_unreachable_peer(tmp_path, capsys): + alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path) + parser = build_parser() + # Bind 127.0.0.1:1 — almost certainly nothing listening here. + args = parser.parse_args([ + "--db", str(bob_db), "mesh", "sync", + "--peer", "http://127.0.0.1:1", + ]) + rc = args.func(args) + assert rc == 2 + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "peer_unreachable" + + +# --------------------------------------------------------------------------- +# serve: smoke-test the server starts and responds via /mesh/info +# --------------------------------------------------------------------------- + + +def test_serve_responds_on_info_when_started_in_thread(tmp_path): + """We don't actually invoke `mesh serve` (which blocks); we exercise + MeshWireServer's start_in_thread path the verb relies on.""" + import httpx + + db = tmp_path / "x.db" + conn = connect(db) + try: + init_identity(conn, group_name="t", member_id="alice") + set_enabled(conn, True) + finally: + conn.close() + + srv = MeshWireServer(db, host="127.0.0.1", port=0) + srv.start_in_thread() + try: + r = httpx.get(srv.url + "/mesh/info", timeout=5.0) + r.raise_for_status() + info = r.json() + finally: + srv.stop() + assert info["member_id"] == "alice" + assert info["group_name"] == "t"