Agentswarm/scripts/test-workflow-e2e.py

"""End-to-end workflow test: proves the program follows the expected workflow.

Boots the REAL orchestrator (uvicorn, in-process) on the in-memory store, connects a keyless
stub agent over WebSocket, submits one objective, and asserts the full loop:

    decompose -> dispatch to experts -> execute -> master review (with one forced reopen)
              -> iterate -> synthesize -> deliver

No OPENAI_API_KEY required: the planner is forced offline (static decomposition + heuristic
review + concatenated synthesis), and the stub agent returns canned results that make the
critic reject exactly once before accepting.

Run from the agent_swarm_v5 directory:
    python scripts/test-workflow-e2e.py
"""
import asyncio
import logging
import os
import sys
import threading
from pathlib import Path

# Configure the runtime for a deterministic, hermetic run BEFORE importing the app.
# No mode flags: this repo IS the swarm runtime — seed + self-organize is the only flow.
os.environ["REDIS_FAKE"] = "1"
os.environ["MAX_REVIEW_CYCLES"] = "2"

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts"))

import httpx
import uvicorn
from orchestrator import main as orch
from stub_agent import StubAgent

# Force the planner/critic/synthesis offline so the workflow is deterministic regardless of any
# .env key (load_dotenv runs at import). Quiet the dummy-callback warnings.
orch.planner.client = None
logging.getLogger("orchestrator.swarm_runtime").setLevel(logging.ERROR)

PORT = 8123
BASE = f"http://127.0.0.1:{PORT}"
WS = f"ws://127.0.0.1:{PORT}"

failures = []


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


async def wait_health(client):
    for _ in range(150):
        try:
            if (await client.get(f"{BASE}/health")).status_code == 200:
                return True
        except Exception:
            pass
        await asyncio.sleep(0.1)
    return False


async def main():
    config = uvicorn.Config(orch.app, host="127.0.0.1", port=PORT, log_level="warning")
    server = uvicorn.Server(config)
    server.install_signal_handlers = lambda: None  # required when not on the main thread
    server_thread = threading.Thread(target=server.run, daemon=True)
    server_thread.start()

    agent = StubAgent(WS, "stub-1", [
        "python", "code_generation", "testing", "pytest", "technical-writing", "general",
    ])
    agent_task = None
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            if not await wait_health(client):
                check("orchestrator started", False)
                return 1

            agent_task = asyncio.create_task(agent.run())
            await asyncio.sleep(0.5)  # let the agent register

            body = {
                "mode": "swarm",
                "requirement": {"objective": "Write a Python add(a,b) function with tests and docs"},
                "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
                "metadata": {"manager_deployment_id": "e2e-1"},
            }
            created = (await client.post(f"{BASE}/api/swarms", json=body)).json()
            dep = created["data"]["deployment_id"]

            # Poll until the run finishes (or times out).
            status, wf = None, {}
            for _ in range(200):
                wf = (await client.get(f"{BASE}/api/swarms/{dep}/workflow")).json()["data"]
                status = wf.get("status")
                if status in ("completed", "failed"):
                    break
                await asyncio.sleep(0.25)

            tasks = (await client.get(f"{BASE}/api/swarms/{dep}/tasks")).json()["data"]["tasks"]
            events = (await client.get(f"{BASE}/api/swarms/{dep}/logs")).json()["data"]["events"]
            roles = sorted(t.get("agent_role") for t in tasks)
            sources = [t.get("source") for t in tasks]
            payloads = [(e.get("payload") or {}) for e in events]
            termination_seen = any(p.get("termination_reason") for p in payloads)
            ev_types = [e.get("event_type") for e in events]
            seqs = [e.get("sequence") for e in events]

            # ---- swarm-flow assertions (seed → agent-decompose → self-select → converge) ----
            check("seed: a single objective seed task was injected (no Master plan)",
                  sources.count("seed") == 1)
            check("decompose: agents grew the graph bottom-up (>=2 agent-proposed subtasks)",
                  sources.count("agent_proposed") >= 2)
            check("decompose: proposed roles include specialist roles",
                  {"implementation", "testing", "documentation"} & set(roles))
            check("execute: every task (seed + proposed) completed",
                  bool(tasks) and all(t["status"] == "completed" for t in tasks))
            check("converge: run reached completed", status == "completed")
            check("converge: a termination_reason was emitted (convergence report)", termination_seen)
            check("synthesize: a final unified summary is present", bool(wf.get("summary")))
            # ---- frozen Manager/client contract (agent_swarm#14/#15) ----
            check("contract: terminal swarm.completed event emitted", "swarm.completed" in ev_types)
            check("contract: every event carries a per-swarm sequence",
                  bool(seqs) and all(isinstance(s, int) for s in seqs))
            check("contract: sequence is strictly increasing and gap-free",
                  seqs == list(range(1, len(seqs) + 1)))
    finally:
        agent.running = False
        if agent_task:
            agent_task.cancel()
        server.should_exit = True
        server_thread.join(timeout=5)  # let uvicorn stop so no daemon thread lingers at exit

    print()
    if failures:
        print(f"{len(failures)} workflow check(s) FAILED: {failures}")
        return 1
    print("swarm flow verified: seed -> agent self-select -> bottom-up decompose -> execute -> converge")
    return 0


if __name__ == "__main__":
    # os._exit after flushing avoids a Fatal Python error if the uvicorn daemon thread is still
    # finalizing (writing to stderr) at interpreter shutdown — guarantees a deterministic exit code.
    _code = asyncio.run(main())
    sys.stdout.flush()
    sys.stderr.flush()
    os._exit(_code or 0)