Agentswarm/scripts/test-dispatch-scored.py

"""Integration test for issue #9: task-centric SCORED dispatch picks among AGENTS.

Exercises orchestrator.main.scored_matchmake against the real task_queue + agent_registry +
decision_engine (REDIS_FAKE, no WS/model): two capable agents differ in historical success (τ)
and load; the higher-scored agent is chosen; an incapable agent is excluded; the explainable
dispatch.decision_made record (candidate breakdown + exclusion reasons + ≥4 non-capability
dimensions with real signals) is produced and stored on the run.

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-dispatch-scored.py
"""
import asyncio
import os
import sys
from pathlib import Path

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

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.decision_engine import decision_engine
from orchestrator import main as orch

failures = []


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


async def _noop(self, *a, **k):
    return None


async def main():
    await redis_client.connect()
    sr_mod.SwarmRuntime._post_callback = _noop

    body = {
        "mode": "swarm",
        "requirement": {"objective": "scored dispatch test"},
        "orchestration_plan": {"budget": {"max_cost_usd": 10}},
        "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
        "metadata": {"manager_deployment_id": "m-disp"},
    }
    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cd")

    # One ready task requiring python.
    task = await task_queue.create_task(task_id="t-implementation", description="impl",
                                        agent_role="implementation",
                                        required_capabilities=["python"], enqueue=True)
    await swarm_runtime.attach_task(run, task.task_id)

    # Three agents: A & B both capable (python); C incapable (java only).
    await agent_registry.register_agent("agent-A", ["python"])
    await agent_registry.register_agent("agent-B", ["python"])
    await agent_registry.register_agent("agent-C", ["java"])
    orch.AGENT_SLOTS.update({"agent-A": 1, "agent-B": 1, "agent-C": 1})

    # Historical success (τ) differs: A strong, B weak — same capability & load otherwise.
    for _ in range(8):
        await decision_engine.deposit(agent_role="implementation", agent_id="agent-A", success=True)
        await decision_engine.deposit(agent_role="implementation", agent_id="agent-B", success=False)

    idle = [a for a in await agent_registry.get_idle_agents()
            if a.agent_id in {"agent-A", "agent-B", "agent-C"}]
    assignments = await orch.scored_matchmake(idle)

    check("exactly one assignment produced", len(assignments) == 1)
    agent, chosen_task, event = assignments[0]
    check("higher-τ agent A chosen over B (historical_success decides)", agent.agent_id == "agent-A")
    check("chosen task is the python task", chosen_task.task_id == "t-implementation")

    # Explainable event payload
    check("event names chosen agent/task",
          event["chosen_agent_id"] == "agent-A" and event["chosen_task_id"] == "t-implementation")
    check("incapable agent C excluded as capability_mismatch",
          event["excluded"].get("agent-C") == "capability_mismatch")
    check("loser B excluded as lower_score", event["excluded"].get("agent-B") == "lower_score")
    a_cand = next(c for c in event["candidates"] if c["agent_id"] == "agent-A")
    present = {k for k, v in a_cand["score"]["breakdown"].items() if v is not None}
    noncap_present = present - {"capability_match"}
    check("≥4 non-capability dimensions have real signals",
          {"historical_success", "load", "permission_fit", "budget_pressure"} <= noncap_present
          and len(noncap_present) >= 4)
    check("uncollected dims disclosed (no fabrication)",
          set(event["uncollected_dimensions"]) >= {"estimated_cost", "estimated_time", "risk_score"})
    check("A's historical_success > B's in breakdown",
          a_cand["score"]["breakdown"]["historical_success"]
          > next(c for c in event["candidates"] if c["agent_id"] == "agent-B")["score"]["breakdown"]["historical_success"])

    # finalize_dispatch stores the decision on the run (audit/benchmark-replayable, internal)
    await task_queue.remove_pending_task(chosen_task.task_id)
    ok = await orch.finalize_dispatch(agent, chosen_task, dispatch_event=event)
    check("finalize_dispatch assigned the task", ok is True)
    refreshed = await swarm_runtime.get_run(run.swarm_id)
    check("dispatch decision recorded on run (replayable)",
          len(refreshed.dispatch_decisions) == 1
          and refreshed.dispatch_decisions[0]["chosen_agent_id"] == "agent-A")

    print()
    if failures:
        print(f"{len(failures)} scored-dispatch check(s) FAILED: {failures}")
        sys.exit(1)
    print("all scored-dispatch (#9) checks passed")


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