prose-contracts/tests/test_mumuni_monitor_removal.py

"""Regression tests for the 2026-09-10 retirement of the Mumuni monitoring leg.

WHY THIS FILE EXISTS: captain ruling 2026-09-10 — Mumuni moved off this host
onto her own container (kagentz CT 105 on minipve, 192.168.68.14, dedicated
`hermes` user) and is monitored from her side. The monitor nevertheless kept
ssh'ing to root@192.168.68.24 for `~/.hermes/gateway_state.json` on the
decommissioned deployment, read "unknown" on every run, and posted a false 🔴
"Mumuni (Hermes) Zulip state: unknown" DM + #agent-hub stream alert to the
captain. The daily infra digest published a matching `mumuni:unknown` row.

CONTRACT UNDER TEST:
  * `scripts/zulip-monitor.sh` carries NO Mumuni probe and NO 192.168.68.24
    reference; it never ssh'es .24, and even on a failing run it emits no Mumuni
    notify (stdout alert, Zulip payload, or log line).
  * The Abiba (pi — the Zulip bridge), Tanko (DSH) and Agent Zero (kagentz) legs
    still work: deleting the Mumuni leg must not have gutted the rest.
  * `scripts/daily-infra-report.py` no longer probes .24 for a Hermes gateway
    state and no longer emits a `mumuni` agent entry.
  * `scripts/agent-health-check.py`'s AGENTS roster has no mumuni entry. This is
    a pin, not a behavior change — verify the probe was already gone.
  * `zulip-health.prose.md` retires the Mumuni-only steps and says explicitly
    that Mumuni is not monitored from this host.

HOW: structural greps plus a behavioral sandbox. The sandbox copies the shipped
script verbatim and rewrites only its LOG constant, then runs it with stub
ssh/curl on PATH. The ssh stub records every host it is asked to reach, so
"never probes .24" is asserted from observed behavior, not from source text.

Usage: python3 -m pytest tests/test_mumuni_monitor_removal.py
"""
from __future__ import annotations

import importlib.util
import os
import pathlib
import stat
import subprocess

import pytest

ROOT = pathlib.Path(__file__).resolve().parents[1]
ZULIP_MONITOR = ROOT / "scripts" / "zulip-monitor.sh"
DAILY_REPORT = ROOT / "scripts" / "daily-infra-report.py"
AHC = ROOT / "scripts" / "agent-health-check.py"
HEALTH_CONTRACT = ROOT / "zulip-health.prose.md"
CONNECTED_FIXTURE = ROOT / "tests" / "fixtures" / "zulip-health-connected.json"

MUMUNI_IP = "192.168.68.24"       # Mumuni's old (decommissioned) deployment
TANKO_VANTAGE = "192.168.68.15"   # amdpve — Tanko CT 112 via pct exec
AGENT_ZERO_HOST = "192.168.68.14"  # kagentz host, Agent Zero docker


# ── scripts/zulip-monitor.sh: structural guards ──────────────────────

def test_zulip_monitor_has_no_mumuni_reference():
    text = ZULIP_MONITOR.read_text()
    assert "mumuni" not in text.lower()
    assert "gateway_state.json" not in text


def test_zulip_monitor_has_no_24_reference():
    assert MUMUNI_IP not in ZULIP_MONITOR.read_text()


def test_zulip_monitor_keeps_bridge_tanko_and_agent_zero_legs():
    text = ZULIP_MONITOR.read_text()
    assert "Platform A: pi (Abiba)" in text          # the Zulip bridge
    assert "Platform B: Tanko" in text
    assert "Platform C: Agent Zero" in text
    assert TANKO_VANTAGE in text
    assert AGENT_ZERO_HOST in text
    assert "kagentz" in text


# ── scripts/zulip-monitor.sh: behavioral sandbox ─────────────────────

SSH_STUB = r"""#!/usr/bin/env bash
# Stub ssh: record the target host, then answer by host + remote command.
printf '%s\n' "$*" >> "$RECORD_DIR/ssh.calls"
host=""
for a in "$@"; do
  case "$a" in
    *@192.168.*) host="${a##*@}" ;;
  esac
done
printf '%s\n' "$host" >> "$RECORD_DIR/ssh.hosts"
cmd="${*: -1}"
case "$host" in
  192.168.68.15)
    case "$cmd" in
      *"systemctl is-active"*) printf '%s' "$TANKO_SVC" ;;
      *curl*)                  printf '%s' "$TANKO_HTTP" ;;
    esac ;;
  192.168.68.14)
    case "$cmd" in
      *agent.json*) printf '%s' "$AZ_A2A" ;;
      *"ps aux"*)   printf '%s\n' "$AZ_PS" ;;
    esac ;;
  *)
    printf 'UNEXPECTED-SSH-HOST %s\n' "$host" >> "$RECORD_DIR/unexpected-ssh" ;;
esac
exit 0
"""

CURL_STUB = r"""#!/usr/bin/env bash
# Stub curl: serve the Abiba health fixture and the Zulip server 200, and
# record every call (including notify) payloads.
printf '%s\n' "$*" >> "$RECORD_DIR/curl.calls"
case "$*" in
  *:9200/health*)
    case " $* " in
      *" -w "*) printf '%s' "$PI_HTTP" ;;   # -w '%{http_code}' probe
      *)        printf '%s' "$PI_BODY" ;;   # body probe
    esac ;;
  *server_settings*)
    printf '%s' "$SERVER_HTTP" ;;
esac
exit 0
"""


def _write_exec(path: pathlib.Path, body: str) -> None:
    path.write_text(body)
    path.chmod(path.stat().st_mode
               | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)


def _run_monitor(tmp_path, *, tanko_svc="active", tanko_http="200",
                 az_a2a='{"name":"kagentz"}',
                 az_ps="root 111 0.1 0.2 /opt/venv-a0/bin/python3 -u adapter.py"):
    """Run the shipped monitor in a sandbox; return (proc, record_dir, log_path).

    Only the LOG constant is rewritten (to keep the run inside the worktree).
    Everything else — legs, labels, notify logic — is the shipped script.
    """
    sandbox = tmp_path / "sandbox"
    bindir = sandbox / "bin"
    record = sandbox / "record"
    bindir.mkdir(parents=True)
    record.mkdir()

    _write_exec(bindir / "ssh", SSH_STUB)
    _write_exec(bindir / "curl", CURL_STUB)

    source = ZULIP_MONITOR.read_text()
    log_line = 'LOG="/root/zulip-health-monitor.log"'
    assert log_line in source, "LOG constant moved — update the sandbox harness"
    log_path = sandbox / "zulip-health-monitor.log"
    script = sandbox / "zulip-monitor.sh"
    script.write_text(source.replace(log_line, f'LOG="{log_path}"'))

    env = dict(os.environ)
    env.update({
        "PATH": f"{bindir}:{env['PATH']}",
        "RECORD_DIR": str(record),
        "TANKO_SVC": tanko_svc,
        "TANKO_HTTP": tanko_http,
        "AZ_A2A": az_a2a,
        "AZ_PS": az_ps,
        "PI_HTTP": "200",
        "PI_BODY": CONNECTED_FIXTURE.read_text(),
        "SERVER_HTTP": "200",
    })
    proc = subprocess.run(["bash", str(script)], cwd=sandbox, env=env,
                          capture_output=True, text=True)
    return proc, record, log_path


def test_healthy_run_is_quiet_and_never_reaches_mumuni(tmp_path):
    proc, record, log_path = _run_monitor(tmp_path)
    assert proc.returncode == 0, proc.stderr
    log = log_path.read_text()

    # Every retained leg actually ran and passed.
    assert "Server: ✅ HTTP 200" in log
    assert "Abiba: ✅ Connected" in log
    assert "Tanko: ✅ service=active http=200" in log
    assert "kagentz: ✅ A2A alive" in log
    assert "kagentz: ✅ Adapter running" in log
    assert "Result: ✅ All healthy" in log

    # A healthy run emits no notify at all — and certainly no Mumuni one.
    assert proc.stdout == ""
    assert "Mumuni" not in log
    assert "🔴" not in log

    # Observed behavior: .24 is never resolved, only Tanko's vantage and the
    # Agent Zero host are contacted.
    hosts = record.joinpath("ssh.hosts").read_text().split()
    assert MUMUNI_IP not in hosts
    assert set(hosts) == {TANKO_VANTAGE, AGENT_ZERO_HOST}
    assert not record.joinpath("unexpected-ssh").exists()


def test_failing_run_alerts_on_tanko_but_never_on_mumuni(tmp_path):
    # Failure path: exercises notify() end to end so "no Mumuni notify" is
    # proven on the alert path, not only on the quiet healthy path.
    proc, record, log_path = _run_monitor(tmp_path, tanko_svc="inactive",
                                          tanko_http="000")
    assert proc.returncode == 0, proc.stderr

    alerts = proc.stdout
    assert "Tanko (DSH dsh-web) service state: inactive" in alerts
    assert "1 issue(s) found" in alerts

    # No Mumuni text in stdout, the log, or any Zulip DM/stream payload.
    assert "Mumuni" not in alerts
    assert "Mumuni" not in log_path.read_text()
    assert MUMUNI_IP not in alerts + log_path.read_text()
    payloads = record.joinpath("curl.calls").read_text()
    assert "Mumuni" not in payloads
    assert MUMUNI_IP not in payloads

    # The rest of the monitor still ran alongside the failing Tanko leg.
    log = log_path.read_text()
    assert "Abiba: ✅ Connected" in log
    assert "kagentz: ✅ A2A alive" in log
    assert "Result: 🔴 1 issue(s) found" in log


# ── scripts/daily-infra-report.py: no Mumuni agent probe ─────────────

def test_daily_report_has_no_mumuni_agent_probe():
    text = DAILY_REPORT.read_text()
    assert 'report["agents"]["mumuni"]' not in text
    assert f'ssh("{MUMUNI_IP}"' not in text
    assert "hermes --version" not in text


def test_daily_report_keeps_abiba_and_tanko_agent_legs():
    text = DAILY_REPORT.read_text()
    assert 'report["agents"]["abiba"]' in text
    assert 'report["agents"]["tanko"]' in text


def test_daily_report_still_describes_abiba_at_its_own_ct100_ip():
    # .24 is abiba's own CT 100 address — the IP itself is legitimate; only a
    # Mumuni gateway probe against it was the fault.
    assert '"platform": "pi", "ct": 100, "ip": "192.168.68.24"' in \
        DAILY_REPORT.read_text()


# ── scripts/agent-health-check.py: roster pin ───────────────────────

@pytest.fixture(scope="module")
def ahc():
    spec = importlib.util.spec_from_file_location("agent_health_check_roster", AHC)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_agent_health_roster_has_no_mumuni_entry(ahc):
    assert "mumuni" not in ahc.AGENTS


def test_agent_health_roster_comment_no_longer_places_mumuni_at_24():
    text = AHC.read_text()
    assert "mumuni (.24" not in text
    assert "mumuni (.24, inside abiba CT100)" not in text


# ── zulip-health.prose.md: contract reconciliation ──────────────────

def test_health_contract_retires_mumuni_only_steps():
    text = HEALTH_CONTRACT.read_text()
    assert MUMUNI_IP not in text
    for step in ("**B4:", "**B5:", "**B6:"):
        assert step not in text


def test_health_contract_states_mumuni_is_not_monitored_from_this_host():
    text = HEALTH_CONTRACT.read_text()
    assert "Mumuni is NOT monitored from this host" in text
    assert "monitored on her side" in text
    assert "her own container" in text


def test_health_contract_keeps_tanko_agent_zero_and_bridge_steps():
    text = HEALTH_CONTRACT.read_text()
    for marker in ("**B1:", "**B2:", "**B3:", "Step 4: Platform C",
                   "Step 2: Platform A", "Step 1: Zulip Server Liveness"):
        assert marker in text, marker