Agentswarm/scripts/test-swarm-guard.py

"""Test the swarm health guard (P-guard): detect 'swarm can't run' + reasons.

Covers the pure diagnostic (orchestrator/guard.diagnose) across blocker types, and the
orchestrator wrapper (assess_swarm_health) that records/emits the report. Hermetic, no model key.

Run from agent_swarm_v6 (install deps first — needs fakeredis):
    pip install -r orchestrator/requirements.txt -r agent/requirements.txt
    REDIS_FAKE=1 python scripts/test-swarm-guard.py
"""
import asyncio
import os
import sys
from pathlib import Path

os.environ["REDIS_FAKE"] = "1"
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from orchestrator import guard
from orchestrator.redis_client import redis_client
from orchestrator import swarm_runtime as sr_mod
from orchestrator.swarm_runtime import swarm_runtime
from orchestrator.task_queue import task_queue, TaskStatus
from orchestrator.agent_registry import agent_registry
from orchestrator import main as orch

failures = []


def check(name, cond):
    print(("PASS" if cond else "FAIL"), "-", name)
    if not cond:
        failures.append(name)


def reasons(report):
    return {b["reason"] for b in report.blockers}


# ---- pure diagnose ----
def test_pure():
    # healthy: a ready task and a capable agent
    r = guard.diagnose({
        "tasks": [{"task_id": "t1", "status": "pending", "required_capabilities": ["python"], "depends_on": []}],
        "connected_agent_caps": [["python", "testing"]],
    })
    check("healthy when a capable agent exists", r.healthy and not r.blockers)

    # no agents connected
    r = guard.diagnose({
        "tasks": [{"task_id": "t1", "status": "pending", "required_capabilities": [], "depends_on": []}],
        "connected_agent_caps": [],
    })
    check("NO_AGENTS_CONNECTED when nobody is connected",
          not r.healthy and guard.Blocker.NO_AGENTS_CONNECTED.value in reasons(r))

    # no capable agent
    r = guard.diagnose({
        "tasks": [{"task_id": "t1", "status": "pending", "required_capabilities": ["rust"], "depends_on": []}],
        "connected_agent_caps": [["python"]],
    })
    check("NO_CAPABLE_AGENT when caps uncovered",
          not r.healthy and guard.Blocker.NO_CAPABLE_AGENT.value in reasons(r))

    # dependency deadlock (dep failed)
    r = guard.diagnose({
        "tasks": [
            {"task_id": "dep", "status": "failed", "required_capabilities": [], "depends_on": []},
            {"task_id": "t1", "status": "pending", "required_capabilities": ["python"], "depends_on": ["dep"]},
        ],
        "connected_agent_caps": [["python"]],
    })
    check("DEPENDENCY_DEADLOCK when a dependency FAILED",
          not r.healthy and guard.Blocker.DEPENDENCY_DEADLOCK.value in reasons(r))

    # budget exhausted while active
    r = guard.diagnose({
        "tasks": [{"task_id": "t1", "status": "pending", "required_capabilities": ["python"], "depends_on": []}],
        "connected_agent_caps": [["python"]],
        "budget_state": {"exhausted": True},
    })
    check("BUDGET_EXHAUSTED while work remains",
          not r.healthy and guard.Blocker.BUDGET_EXHAUSTED.value in reasons(r))

    # seed completed but nothing proposed
    r = guard.diagnose({
        "tasks": [{"task_id": "seed", "status": "completed", "required_capabilities": [], "depends_on": [], "source": "seed"}],
        "connected_agent_caps": [["python"]],
    })
    check("SEED_UNDECOMPOSED when only a terminal seed exists",
          not r.healthy and guard.Blocker.SEED_UNDECOMPOSED.value in reasons(r))

    # blockers carry a human-readable detail
    check("blockers include a detail string", all(b.get("detail") for b in r.blockers))


# ---- orchestrator wrapper records the report INTERNALLY (not a Manager event) ----
async def test_wrapper():
    await redis_client.connect()

    async def _noop_cb(self, *a, **k):
        return None
    sr_mod.SwarmRuntime._post_callback = _noop_cb
    emitted = []
    orig_emit = sr_mod.SwarmRuntime.emit_event

    async def spy_emit(self, run, event_type, **k):
        emitted.append(event_type)
        return await orig_emit(self, run, event_type, **k)
    sr_mod.SwarmRuntime.emit_event = spy_emit

    body = {"mode": "swarm", "requirement": {"objective": "guard test"},
            "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
            "metadata": {"manager_deployment_id": "m-guard"}}
    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cg")
    # a pending task requiring caps nobody connected has
    t = await task_queue.create_task(task_id=f"{run.swarm_id}-t", description="x",
                                     agent_role="impl", required_capabilities=["rust"], enqueue=True)
    await swarm_runtime.attach_task(run, t.task_id)
    # no connected agents → NO_AGENTS_CONNECTED (+ would be NO_CAPABLE if any)
    report = await orch.assess_swarm_health(run, connected_agent_ids=[])
    check("wrapper reports unhealthy", report.healthy is False)
    refreshed = await swarm_runtime.get_run(run.swarm_id)
    check("health report stored on run", bool(refreshed.metadata.get("health"))
          and refreshed.metadata["health"]["healthy"] is False)
    check("unhealthy report appended to internal health_log", bool(refreshed.metadata.get("health_log")))
    # Contract hygiene: health is INTERNAL — no unregistered Manager event is emitted.
    check("no swarm.health Manager event emitted", "swarm.health" not in emitted)

    # now register a capable agent → healthy
    await agent_registry.register_agent("rust-agent", ["rust"])
    report2 = await orch.assess_swarm_health(run, connected_agent_ids=["rust-agent"])
    check("healthy once a capable agent is connected", report2.healthy is True)

    sr_mod.SwarmRuntime.emit_event = orig_emit


# ---- #56: a stuck run explains itself (events + /result + launch note) ----
async def test_surface_56():
    await redis_client.connect()

    async def _noop_cb(self, *a, **k):
        return None
    sr_mod.SwarmRuntime._post_callback = _noop_cb
    emitted = []
    orig_emit = sr_mod.SwarmRuntime.emit_event

    async def spy_emit(self, run, event_type, **k):
        emitted.append(event_type)
        return await orig_emit(self, run, event_type, **k)
    sr_mod.SwarmRuntime.emit_event = spy_emit

    body = {"mode": "swarm", "requirement": {"objective": "surface test"},
            "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
            "metadata": {"manager_deployment_id": "m-surface"}}
    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cs")
    seed = await task_queue.create_task(task_id=f"{run.swarm_id}-seed", description="x",
                                        agent_role="general", required_capabilities=[], enqueue=True)
    await swarm_runtime.attach_task(run, seed.task_id)

    # pending seed + no agents → NO_AGENTS_CONNECTED; first tick emits timeline.updated, second dedups.
    await orch.assess_swarm_health(run, connected_agent_ids=[])
    await orch.assess_swarm_health(run, connected_agent_ids=[])
    check("#56 blocked run emits timeline.updated exactly once (dedup)",
          emitted.count("timeline.updated") == 1)
    check("#56 still no unregistered swarm.health event", "swarm.health" not in emitted)

    refreshed = await swarm_runtime.get_run(run.swarm_id)
    result = await orch.build_run_result(refreshed)
    health = result.get("health") or {}
    check("#56 /result carries health (unhealthy)", health.get("healthy") is False)
    check("#56 /result health names no_agents_connected",
          any(b["reason"] == guard.Blocker.NO_AGENTS_CONNECTED.value for b in health.get("blockers", [])))
    diag = await orch.build_runtime_diagnostics(refreshed)
    check("#56 /diagnostics carries health", bool(diag.get("health")))

    sr_mod.SwarmRuntime.emit_event = orig_emit


async def test_launch_outcome_56():
    await redis_client.connect()
    os.environ.pop("AGENT_LAUNCH_BACKEND", None)  # default 'none'
    body = {"mode": "swarm", "requirement": {"objective": "launch note"},
            "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
            "metadata": {"manager_deployment_id": "m-launch"}}
    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cl")
    await orch.launch_swarm_agents(run, body)
    info = (await swarm_runtime.get_run(run.swarm_id)).metadata.get("agent_launch") or {}
    check("#56 agent_launch recorded (backend none)", info.get("backend") == "none")
    check("#56 agent_launch launched 0", info.get("launched") == 0)
    check("#56 agent_launch note explains backend=none", "AGENT_LAUNCH_BACKEND=none" in (info.get("note") or ""))


async def main():
    test_pure()
    await test_wrapper()
    await test_surface_56()
    await test_launch_outcome_56()
    print()
    if failures:
        print(f"{len(failures)} guard check(s) FAILED: {failures}")
        sys.exit(1)
    print("all swarm guard (P-guard) checks passed")


if __name__ == "__main__":
    asyncio.run(main())