arborist/tests/test_mesh_cli.py

"""CLI integration: `arborist mesh serve` and `arborist 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 arborist.cli import build_parser
from arborist.document import Document
from arborist.ingest import ingest_source
from arborist.mesh import init_identity, set_enabled
from arborist.mesh.members import add_member
from arborist.mesh.wire import MeshWireServer
from arborist.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",
            "--kindergarten-seconds", "0",   # immediate broadcast for the test
        ])
        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_roots"] == 2
    assert payload["announced_falsifications"] == 0  # bob has none
    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_announces_falsifications(tmp_path, capsys):
    """Bob falsifies a local cache record; mesh sync to alice broadcasts
    the falsification. Alice's audit chain gains one mesh_received event
    whose body carries an ANNOUNCE_FALSIFICATION envelope."""
    import time as _time
    from arborist.store import append_audit, transaction

    alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)

    # Seed a falsification on bob's side.
    b_conn = connect(bob_db)
    try:
        with transaction(b_conn):
            event_hash = append_audit(
                b_conn,
                event_type="falsify",
                subject_root="ab" * 32,
                body={"cache_key": "ab" * 32, "to_state": "failed", "reason": "wrong"},
            )
            b_conn.execute(
                "INSERT INTO falsifications "
                "(cache_key, state, reason, by_actor, at, audit_event_hash) "
                "VALUES (?, 'failed', 'wrong', 'bob', ?, ?)",
                ("ab" * 32, int(_time.time()), event_hash),
            )
    finally:
        b_conn.close()

    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
    srv.start_in_thread()
    try:
        parser = build_parser()
        # --no-roots so we isolate the falsification path.
        args = parser.parse_args([
            "--db", str(bob_db), "mesh", "sync",
            "--peer", srv.url,
            "--limit", "10",
            "--no-roots",
            "--kindergarten-seconds", "0",
        ])
        rc = args.func(args)
        assert rc == 0
    finally:
        srv.stop()

    payload = json.loads(capsys.readouterr().out)
    assert payload["announced_roots"] == 0
    assert payload["announced_falsifications"] == 1
    assert payload["errors"] == 0

    # Alice's chain gained one mesh_received event with the falsification body.
    import json as _json
    a_conn = connect(alice_db)
    try:
        last = a_conn.execute(
            "SELECT event_type, body FROM audit_events "
            "WHERE event_type = 'mesh_received' "
            "ORDER BY seq DESC LIMIT 1"
        ).fetchone()
    finally:
        a_conn.close()
    body = _json.loads(last["body"])
    assert body["wire_type"] == "ANNOUNCE_FALSIFICATION"
    assert body["envelope_body"]["cache_key"] == "ab" * 32
    assert body["envelope_body"]["reason"] == "wrong"


def test_sync_no_falsifications_flag_skips_them(tmp_path, capsys):
    """--no-falsifications skips the falsification broadcast. Bob's
    falsification is NOT announced. Alice gets only the root broadcasts."""
    import time as _time
    from arborist.store import append_audit, transaction

    alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
    b_conn = connect(bob_db)
    try:
        with transaction(b_conn):
            event_hash = append_audit(
                b_conn, event_type="falsify",
                subject_root="cd" * 32,
                body={"cache_key": "cd" * 32, "to_state": "failed", "reason": "x"},
            )
            b_conn.execute(
                "INSERT INTO falsifications "
                "(cache_key, state, reason, by_actor, at, audit_event_hash) "
                "VALUES (?, 'failed', 'x', 'bob', ?, ?)",
                ("cd" * 32, int(_time.time()), event_hash),
            )
    finally:
        b_conn.close()

    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", "10",
            "--no-falsifications",
            "--kindergarten-seconds", "0",
        ])
        rc = args.func(args)
        assert rc == 0
    finally:
        srv.stop()

    payload = json.loads(capsys.readouterr().out)
    assert payload["announced_falsifications"] == 0


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"


def test_sync_kindergarten_holds_fresh_records(tmp_path, capsys):
    """Fox 2026-04-29: gossip should default-delay so an operator can
    burn or falsify within a window before peers see fresh records.
    A document ingested 30 seconds ago must NOT broadcast under the
    default 1-hour kindergarten window."""
    import time as _time
    from arborist.store import transaction

    alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
    _ingest(bob_db, uri="https://example.com/old", content="Old prose.", title="Old")
    _ingest(bob_db, uri="https://example.com/fresh", content="Fresh prose.", title="Fresh")

    # Backdate the "old" doc so it's >>> kindergarten window. Leave the
    # "fresh" one at its just-set ingest_ts (within the last second).
    long_ago = int(_time.time()) - 7200  # 2 hours ago
    b_conn = connect(bob_db)
    try:
        with transaction(b_conn):
            b_conn.execute(
                "UPDATE documents SET ingest_ts = ? "
                "WHERE document_uri = 'https://example.com/old'",
                (long_ago,),
            )
    finally:
        b_conn.close()

    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", "10",
            # Default kindergarten = 3600s. Fresh doc held back; old doc
            # broadcast.
        ])
        rc = args.func(args)
        assert rc == 0
    finally:
        srv.stop()

    payload = json.loads(capsys.readouterr().out)
    assert payload["kindergarten_seconds"] == 3600
    assert payload["announced_roots"] == 1   # only "old" broadcast
    assert payload["kindergarten_held_roots"] >= 1   # "fresh" held
    # Confirm exactly which root was sent.
    sent_uris = [r["ack"]["wire_type"] for r in payload["sent_roots"]]
    assert sent_uris == ["ANNOUNCE_ROOT"]


def test_sync_kindergarten_zero_broadcasts_everything(tmp_path, capsys):
    """--kindergarten-seconds 0 is the explicit opt-out: cron-friendly
    immediate broadcast. Verifies the existing "no kindergarten" tests
    in this file aren't accidentally protected by the default window."""
    alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
    _ingest(bob_db, uri="https://example.com/x", content="Just-ingested.", title="X")

    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", "10",
            "--kindergarten-seconds", "0",
        ])
        rc = args.func(args)
        assert rc == 0
    finally:
        srv.stop()

    payload = json.loads(capsys.readouterr().out)
    assert payload["announced_roots"] == 1
    assert payload["kindergarten_held_roots"] == 0