Agentswarm/scripts/test-task-competition.py

r"""Hermetic tests for the agent task-competition protocol (issue #8).

Module-level only: NO Redis, NO WebSocket, NO model calls. `orchestrator.task_competition`
is side-effect free (pure data models + a pure `arbitrate`), so this exercises the real
arbitration math directly and asserts it is explainable and deterministic.

Run from agent_swarm_v6 (install deps first — pydantic is the only requirement here):
    pip install -r orchestrator/requirements.txt
    ..\.venv\Scripts\python.exe scripts/test-task-competition.py
"""
import os
import sys
from pathlib import Path

# Belt-and-braces: this test never reaches Redis, but mirror the other scripts' hermetic
# guard so an accidental import that DOES touch redis_client stays in-memory.
os.environ.setdefault("REDIS_FAKE", "1")
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from orchestrator.task_competition import (
    ArbitrationPolicy,
    TaskBid,
    TaskTakeoverRequest,
    TaskYield,
    arbitrate,
    arbitrated_event,
    bid_submitted_event,
    takeover_requested_event,
    yielded_event,
)

failures = []


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


def main():
    task_id = "task-42"
    required = ["python", "testing"]
    # τ map: the strong agent has earned reputation, the weak one has not (→ neutral 0.5).
    historical = {"agent-strong": 0.9, "agent-weak": 0.3}

    # --- two agents bid for the SAME task ------------------------------------
    strong = TaskBid(
        task_id=task_id,
        agent_id="agent-strong",
        confidence=0.9,
        estimated_cost=0.2,
        estimated_time=120.0,
        risk_score=0.1,
        reason="Specialist; has done this many times.",
        capabilities=["python", "testing", "docs"],
        current_load=0,
    )
    weak = TaskBid(
        task_id=task_id,
        agent_id="agent-weak",
        confidence=0.6,
        estimated_cost=0.7,
        estimated_time=400.0,
        risk_score=0.5,
        reason="Available but unproven on this kind of task.",
        capabilities=["python"],          # missing 'testing' → lower capability fit
        current_load=2,
    )

    result = arbitrate(
        [strong, weak],
        required_capabilities=required,
        historical_success=historical,
    )

    check("better agent wins", result.winner_agent_id == "agent-strong")
    check("loser recorded", result.losers == ["agent-weak"])
    check("decision is decisive", result.decisive is True)
    # Auditable: a non-empty reason naming the winner, plus a full per-bid score breakdown.
    check("reason names winner", "agent-strong" in result.reason)
    check("reason is explainable", "dominant factor" in result.reason)
    check("scores cover every bid", {s.agent_id for s in result.scores} == {"agent-strong", "agent-weak"})
    winner_score = next(s for s in result.scores if s.agent_id == "agent-strong")
    check("winner score broken down into components", set(winner_score.components.keys()) == {
        "confidence", "capability", "historical_success", "budget", "time", "risk", "load",
    })
    check("component sum equals total", round(sum(winner_score.components.values()), 6) == winner_score.total)

    # --- determinism: input order must not change the outcome ----------------
    reversed_result = arbitrate(
        [weak, strong],
        required_capabilities=required,
        historical_success=historical,
    )
    check("winner deterministic under input reorder", reversed_result.winner_agent_id == "agent-strong")
    check("scores deterministic under input reorder",
          [(s.agent_id, s.total) for s in reversed_result.scores]
          == [(s.agent_id, s.total) for s in result.scores])

    # --- determinism: repeated calls are byte-identical ----------------------
    again = arbitrate([strong, weak], required_capabilities=required, historical_success=historical)
    check("repeat call identical winner", again.winner_agent_id == result.winner_agent_id)
    check("repeat call identical scores",
          [(s.agent_id, s.total) for s in again.scores] == [(s.agent_id, s.total) for s in result.scores])

    # --- τ as a real arbitration input ---------------------------------------
    # Make the two bids identical EXCEPT reputation; the higher-τ agent must win, proving
    # historical_success actually moves the result (not a decorative field).
    twin_a = TaskBid(task_id=task_id, agent_id="agent-a", confidence=0.7,
                     capabilities=required, estimated_cost=0.3, risk_score=0.2)
    twin_b = TaskBid(task_id=task_id, agent_id="agent-b", confidence=0.7,
                     capabilities=required, estimated_cost=0.3, risk_score=0.2)
    tau_result = arbitrate(
        [twin_a, twin_b],
        required_capabilities=required,
        historical_success={"agent-a": 0.95, "agent-b": 0.10},
    )
    check("higher historical_success (tau) wins all-else-equal", tau_result.winner_agent_id == "agent-a")

    # --- contested (narrow) ties are flagged, not silently auto-assigned -----
    # Fully identical inputs incl. τ → zero score gap → not decisive; winner falls back to
    # the stable agent_id tiebreak so the call is still deterministic.
    tie = arbitrate(
        [TaskBid(task_id=task_id, agent_id="agent-z", capabilities=required),
         TaskBid(task_id=task_id, agent_id="agent-a", capabilities=required)],
        required_capabilities=required,
        historical_success={},
    )
    check("dead tie is not decisive", tie.decisive is False)
    check("dead tie still deterministic (agent_id tiebreak)", tie.winner_agent_id == "agent-a")
    check("contested reason recommends Manager review", "Manager review" in tie.reason)

    # --- empty bid set is handled ---------------------------------------------
    empty = arbitrate([], required_capabilities=required)
    check("no bids → no winner", empty.winner_agent_id is None)
    check("no bids → not decisive", empty.decisive is False)

    # --- yield-with-reason ----------------------------------------------------
    yield_msg = TaskYield(
        task_id=task_id,
        agent_id="agent-weak",
        release_with_reason="Blocked on missing 'testing' capability.",
        recommend_agent="agent-strong",
    )
    y_type, y_payload = yielded_event(yield_msg)
    check("yield event type", y_type == "task.yielded")
    check("yield carries reason", y_payload["reason"] == "Blocked on missing 'testing' capability.")
    check("yield carries recommendation", y_payload["recommend_agent"] == "agent-strong")
    check("yield summary explainable", "yielded task" in y_payload["summary"])

    # --- takeover request -----------------------------------------------------
    takeover = TaskTakeoverRequest(
        task_id=task_id,
        requesting_agent_id="agent-strong",
        current_agent_id="agent-weak",
        reason="Incumbent stalled; I am a better fit.",
        bid=strong,
    )
    t_type, t_payload = takeover_requested_event(takeover)
    check("takeover event type", t_type == "task.takeover_requested")
    check("takeover names requester", t_payload["requesting_agent_id"] == "agent-strong")
    check("takeover names incumbent", t_payload["current_agent_id"] == "agent-weak")
    # The takeover bid can be arbitrated against the incumbent's bid on identical terms.
    contest = arbitrate(
        [takeover.bid, weak],
        required_capabilities=required,
        historical_success=historical,
    )
    check("takeover bid wins arbitration vs incumbent", contest.winner_agent_id == "agent-strong")

    # --- remaining event builders ---------------------------------------------
    b_type, b_payload = bid_submitted_event(strong)
    check("bid_submitted event type", b_type == "task.bid_submitted")
    check("bid_submitted carries confidence", b_payload["confidence"] == 0.9)
    a_type, a_payload = arbitrated_event(result)
    check("arbitrated event type", a_type == "task.arbitrated")
    check("arbitrated event carries winner", a_payload["winner_agent_id"] == "agent-strong")
    check("arbitrated event carries full scores", len(a_payload["scores"]) == 2)
    check("arbitrated event carries losers", a_payload["losers"] == ["agent-weak"])

    # --- custom policy changes the weighting deterministically ----------------
    # Zero out everything except load; the LESS-loaded agent must then win regardless of τ.
    load_policy = ArbitrationPolicy(
        w_confidence=0.0, w_capability=0.0, w_historical_success=0.0,
        w_budget=0.0, w_time=0.0, w_risk=0.0, w_load=1.0,
    )
    loaded = TaskBid(task_id=task_id, agent_id="agent-busy", current_load=4, capabilities=required)
    free = TaskBid(task_id=task_id, agent_id="agent-free", current_load=0, capabilities=required)
    load_result = arbitrate(
        [loaded, free], policy=load_policy,
        required_capabilities=required, historical_success={"agent-busy": 0.99},
    )
    check("policy override: least-loaded wins under load-only policy",
          load_result.winner_agent_id == "agent-free")

    print()
    if failures:
        print(f"{len(failures)} FAILED: {failures}")
        sys.exit(1)
    print("ALL PASSED")


if __name__ == "__main__":
    main()