Agentswarm/scripts/test-dispatch-score.py

"""Tests for the explainable dispatch scoring layer (issue #9).

Hermetic and deterministic: this exercises orchestrator/dispatch_score.py, which
is pure computation (no RNG, no Redis, no WebSocket, no model key). It does not
touch the dispatch loop or the decision engine's sampling — it only verifies the
explainable score INPUTS and the `dispatch.decision_made` event payload.

Run from agent_swarm_v6 (install deps first to match the other test scripts;
this test itself imports nothing beyond the stdlib + dispatch_score, so it also
runs with no deps installed):
    pip install -r orchestrator/requirements.txt -r agent/requirements.txt
    ..\\.venv\\Scripts\\python.exe scripts/test-dispatch-score.py
"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from orchestrator.dispatch_score import (
    DISPATCH_DECISION_EVENT_TYPE,
    DispatchCandidate,
    build_dispatch_decision_event,
    normalize_capability_match,
    normalize_load,
    normalize_tau,
    rank_candidates,
    score_candidate,
)

failures = []


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


def main():
    # --- normalizers map real signals onto [0,1] and preserve "not collected" ---
    check("capability Jaccard rewards focus",
          normalize_capability_match(["python"], ["python"])
          > normalize_capability_match(["python"], ["python", "go", "rust", "java"]))
    check("no requirements -> perfect capability fit",
          normalize_capability_match([], ["python"]) == 1.0)
    check("tau normalized into [0,1]", 0.0 <= normalize_tau(0.5) <= 1.0)
    check("tau None stays None (not collected, not zero)", normalize_tau(None) is None)
    check("more free slots -> higher load score", normalize_load(2) > normalize_load(0))
    check("load None stays None", normalize_load(None) is None)

    # --- SAME capability, DIFFERENT cost/load/historical_success -> different scores ---
    # All three share capability_match=1.0 and permission_fit=1.0; only the
    # non-capability dimensions vary. A correct scorer must separate them.
    base = dict(task_id="t1", agent_role="implementation",
                capability_match=1.0, permission_fit=1.0)
    strong = DispatchCandidate(agent_id="A", historical_success=0.95, load=1.0,
                               budget_pressure=0.1, **base)
    weak = DispatchCandidate(agent_id="B", historical_success=0.10, load=0.2,
                             budget_pressure=0.9, **base)
    s_strong = score_candidate(strong)
    s_weak = score_candidate(weak)
    check("same capability+permission -> different total by other dims",
          abs(s_strong.total - s_weak.total) > 1e-6)
    check("better history/load/budget scores higher", s_strong.total > s_weak.total)
    check("score stays on [0,1]", 0.0 <= s_strong.total <= 1.0 and 0.0 <= s_weak.total <= 1.0)

    # --- the chosen one is EXPLAINABLE: ranking + breakdown justify it ---
    ranked = rank_candidates([weak, strong])
    chosen = ranked[0]
    check("ranking picks the stronger candidate", chosen.agent_id == "A")
    check("breakdown carries every dimension",
          {"capability_match", "historical_success", "load", "permission_fit",
           "budget_pressure", "risk_score", "estimated_cost", "estimated_time"}
          <= set(chosen.score.breakdown))
    # cost dimension is inverted internally: lower budget_pressure must help, and
    # we can see it in the raw breakdown.
    check("chosen has lower budget_pressure recorded",
          chosen.score.breakdown["budget_pressure"] < s_weak.breakdown["budget_pressure"])

    # --- non-capability dimensions REALLY affect the score (>=4 of them) ---
    # Flip one non-capability dimension at a time from a neutral baseline and
    # confirm each independently moves the total. capability_match is held fixed.
    def neutral():
        return DispatchCandidate(
            agent_id="N", task_id="t1", capability_match=1.0, permission_fit=0.5,
            historical_success=0.5, load=0.5, budget_pressure=0.5,
            risk_score=0.5, estimated_cost=0.5, estimated_time=0.5,
        )

    baseline_total = score_candidate(neutral()).total
    movers = []
    for dim, better_value in [
        ("historical_success", 1.0),  # benefit up -> score up
        ("load", 1.0),                # benefit up -> score up
        ("permission_fit", 1.0),      # benefit up -> score up
        ("budget_pressure", 0.0),     # cost down -> score up
        ("risk_score", 0.0),          # cost down -> score up
        ("estimated_cost", 0.0),      # cost down -> score up
        ("estimated_time", 0.0),      # cost down -> score up
    ]:
        cand = neutral()
        setattr(cand, dim, better_value)
        moved = score_candidate(cand).total
        if moved > baseline_total + 1e-9:
            movers.append(dim)
    check("at least 4 non-capability dimensions independently raise the score",
          len(movers) >= 4)
    check("each of >=4 movers is a distinct non-capability dimension",
          len(set(movers)) >= 4 and "capability_match" not in movers)

    # --- honesty: None dimensions are masked, not fabricated to 0 ---
    # A candidate with several signals uncollected must NOT be punished for them:
    # masking out a None dimension should give the SAME total as a candidate that
    # only has the collected dimensions.
    partial = DispatchCandidate(agent_id="P", task_id="t1", capability_match=1.0,
                                permission_fit=1.0, historical_success=0.8)
    sp = score_candidate(partial)
    check("uncollected dims listed in `missing`",
          {"load", "budget_pressure", "risk_score",
           "estimated_cost", "estimated_time"} <= set(sp.missing))
    check("masked None dims do not drag score toward 0", sp.total > 0.5)
    check("weights_used only counts present dims",
          set(sp.weights_used) == {"capability_match", "historical_success", "permission_fit"})

    # --- event payload carries candidate breakdown + exclusion reasons ---
    excluded = {"B": "lower_score", "C": "capability_mismatch"}
    mismatch = DispatchCandidate(agent_id="C", task_id="t1", capability_match=0.0,
                                 permission_fit=0.0)
    score_candidate(mismatch)
    event = build_dispatch_decision_event(
        candidates=[strong, weak, mismatch],
        chosen=strong,
        excluded_reasons=excluded,
        task_id="t1",
    )
    check("event type constant matches schema", DISPATCH_DECISION_EVENT_TYPE == "dispatch.decision_made")
    check("event records chosen agent + task",
          event["chosen_agent_id"] == "A" and event["chosen_task_id"] == "t1")
    check("event lists all candidates", event["candidate_count"] == 3 and len(event["candidates"]) == 3)
    check("each event candidate carries a full score breakdown",
          all("breakdown" in c["score"] and "missing" in c["score"] and "weights_used" in c["score"]
              for c in event["candidates"]))
    check("event carries exclusion reasons", event["excluded"] == excluded)
    check("event discloses uncollected dimensions honestly",
          set(event["uncollected_dimensions"]) == {"estimated_cost", "estimated_time", "risk_score"})

    # --- no dispatch case: chosen=None is representable ---
    empty_event = build_dispatch_decision_event(candidates=[mismatch], chosen=None,
                                                excluded_reasons={"C": "capability_mismatch"})
    check("no-dispatch event has chosen None but still explains the excluded candidate",
          empty_event["chosen_agent_id"] is None
          and empty_event["excluded"] == {"C": "capability_mismatch"}
          and empty_event["candidate_count"] == 1)

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


if __name__ == "__main__":
    main()