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()