"""Golden and transport tests for pool-selection.v2.""" from __future__ import annotations import json import subprocess import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "lib")) from selector_policy import select # noqa: E402 from selector_primitives import canonical_sha256 # noqa: E402 SCHEMA = "claude-multiacc/pool-selection.v2" NOW = "2026-08-25T09:00:00.000000Z" def _candidate(**patch) -> dict: row = { "runner_id": 3, "runner_generation": 7, "engine": "claude", "account_id": "acct-a", "status": "active", "weekly_pct": 20, "session_pct": 40, "resets_at": "2026-08-26T09:00:00Z", "limited_until": None, "seen_at": "2026-08-25T08:59:00Z", "provider_capable": True, "reservation_history": {"active_expires_at": None, "last_selected_at": None}} row.update(patch) return row def _request(candidates: list[dict], **patch) -> dict: request = { "schema": SCHEMA, "database_now": NOW, "policy": "default_both", "required_engine": None, "producer_identity": None, "excluded_identities": [], "candidates": candidates, "reservation_key": "proof"} request.update(patch) return request class SelectorTests(unittest.TestCase): def test_union_proof_vector_is_byte_stable(self): request = _request([ _candidate(engine="claude", account_id="claude-a", weekly_pct=10, session_pct=20, seen_at="2026-08-25T08:54:59Z"), _candidate(engine="codex", account_id="codex-b", weekly_pct=30, session_pct=40), _candidate(engine="both", account_id=" ", weekly_pct=0, session_pct=0), ], reservation_key="launch-1") response = select(request) self.assertEqual(response["engine"], "codex") self.assertEqual(response["account_id"], "codex-b") self.assertEqual( response["candidate_snapshot_digest"], "e8ef0b4bc7b95e76231ee6da79616d92bd9251c1c991845c8c0524fcbfb6593a") self.assertEqual( response["selection_digest"], "8b68d69b31b2d2252a16f7b83ac46860987647aa360402caa815b2a1bd468def") def test_unicode_and_active_reservation_proof(self): busy = {"active_expires_at": "2026-08-25T09:00:00.000001Z", "last_selected_at": None} response = select(_request([ _candidate(engine="claude", account_id="claude-active", reservation_history=busy), _candidate(engine="codex", account_id="café-🚀", weekly_pct=25, session_pct=50), ], policy="explicit", required_engine="codex", reservation_key="launch-unicode")) self.assertEqual((response["engine"], response["account_id"]), ("codex", "café-🚀")) self.assertEqual( response["candidate_snapshot_digest"], "29e8e7d1a611f38ee0be43622102b33875e7ff75ff9024aa43d16151575071d4") self.assertEqual( response["selection_digest"], "72e7b31a00496fff0f6281f4dacd656d3a2a1fc438038e2c20c49274e1f61068") def test_both_chooses_each_provider_and_survives_provider_loss(self): codex = select(_request([ _candidate(engine="claude", account_id="claude-a", weekly_pct=70, session_pct=60), _candidate(engine="codex", account_id="codex-a", weekly_pct=20, session_pct=30)])) claude = select(_request([ _candidate(engine="claude", account_id="claude-a", weekly_pct=10, session_pct=30), _candidate(engine="codex", account_id="codex-a", weekly_pct=20, session_pct=80)])) only = select(_request([_candidate(engine="codex", account_id="codex-only")])) self.assertEqual((codex["engine"], claude["engine"], only["engine"]), ("codex", "claude", "codex")) def test_known_quota_and_least_recent_selection_win(self): known = select(_request([ _candidate(engine="claude", account_id="unknown", weekly_pct=None, session_pct=None), _candidate(engine="codex", account_id="known", weekly_pct=99, session_pct=99)])) old = {"active_expires_at": None, "last_selected_at": "2026-08-25T08:00:00Z"} recent = {"active_expires_at": None, "last_selected_at": "2026-08-25T08:50:00Z"} fair = select(_request([ _candidate(engine="claude", account_id="recent", weekly_pct=20, session_pct=20, reservation_history=recent), _candidate(engine="codex", account_id="old", weekly_pct=20, session_pct=20, reservation_history=old)])) self.assertEqual((known["account_id"], fair["account_id"]), ("known", "old")) utc = select(_request([_candidate()])) offset = select(_request( [_candidate()], database_now="2026-08-25T12:00:00+03:00")) self.assertEqual(utc["selection_digest"], offset["selection_digest"]) def test_explicit_and_reviewer_policies_never_cross_provider(self): explicit = select(_request([ _candidate(engine="claude", account_id="free", weekly_pct=0, session_pct=0), _candidate(engine="codex", account_id="busy", weekly_pct=90, session_pct=90), ], policy="explicit", required_engine="codex")) producer = {"runner_id": 3, "runner_generation": 7, "engine": "claude", "account_id": "producer"} review = select(_request([ _candidate(engine="claude", account_id="producer", weekly_pct=0, session_pct=0), _candidate(engine="claude", account_id="reviewer", weekly_pct=90, session_pct=90), ], policy="reviewer", required_engine="claude", producer_identity=producer, excluded_identities=[producer])) sole = select(_request([ _candidate(engine="claude", account_id="producer")], policy="reviewer", required_engine="claude", producer_identity=producer)) alternative = {"runner_id": 4, "runner_generation": 7, "engine": "claude", "account_id": "alternative"} fallback = select(_request([ _candidate(account_id="producer"), _candidate(runner_id=4, account_id="alternative")], policy="reviewer", required_engine="claude", producer_identity=producer, excluded_identities=[alternative])) self.assertEqual((explicit["engine"], review["account_id"]), ("codex", "reviewer")) self.assertEqual( (sole["fallback_reason"], sole["eligible_count"], sole["eligible_alternative_count"]), ("sole_eligible_account", 1, 0)) self.assertEqual( (review["eligible_count"], review["eligible_alternative_count"]), (1, 1)) self.assertEqual((fallback["account_id"], fallback["fallback_reason"]), ("producer", "sole_eligible_account")) def test_stable_errors_fail_closed(self): collisions = [(" Acct-A ", "acct-a"), ("CAFÉ", "cafe\u0301")] duplicates = [select(_request([ _candidate(engine=" CLAUDE ", account_id=left), _candidate(engine="claude", account_id=right)])) for left, right in collisions] no_candidate = select(_request([ _candidate(engine="claude")], policy="default_codex")) invalid = select({**_request([]), "unexpected": True}) preserved = select(_request([_candidate(account_id=" Acct-A ")])) overflow_now = select(_request( [], database_now="9999-12-31T23:59:59-01:00")) overflow_seen = select(_request([ _candidate(seen_at="0001-01-01T00:00:00+01:00")])) overflow_history = select(_request([_candidate(reservation_history={ "active_expires_at": "9999-12-31T23:59:59-01:00", "last_selected_at": None})])) self.assertTrue(all(item["error_code"] == "duplicate_candidate_identity" for item in duplicates)) self.assertTrue(all(item["error_detail"] == {"input_ordinals": [0, 1]} for item in duplicates)) self.assertEqual(no_candidate["error_code"], "no_candidate") self.assertEqual(preserved["account_id"], "Acct-A") self.assertEqual(overflow_now["error_detail"], {"field": "database_now"}) self.assertEqual(overflow_seen["error_code"], "no_candidate") self.assertEqual(overflow_history["error_detail"], { "field": "candidates[0].reservation_history.active_expires_at"}) self.assertEqual(invalid, {"schema": SCHEMA, "selector_version": "2.0.1", "ok": False, "error_code": "invalid_request", "error_detail": {"field": "$"}}) def test_rfc8785_digest_vectors(self): self.assertEqual( canonical_sha256({"account_id": "café-🚀", "metadata": {"": "bmp", "😀": "astral"}}), "2f33747d98d7c2f3cfc0a660eba492a7652a66bc46b147edeccf4cc941154981") with self.assertRaises(ValueError): canonical_sha256({"unsafe": 9_007_199_254_740_992}) def test_cli_transport_and_version_contract(self): binary = ROOT / "bin" / "multiacc-select" args = [str(binary), "--request-json", "-", "--response-json", "-"] good_payload = json.dumps(_request([_candidate(engine="codex")]), separators=(",", ":")) good = subprocess.run(args, input=good_payload, text=True, capture_output=True, check=True) bad_payloads = [ good_payload.replace('"weekly_pct":20', '"weekly_pct":NaN', 1), good_payload.replace( '"weekly_pct":20', '"weekly_pct":1e999999999999999999999999999999', 1), good_payload.replace('"policy":"default_both"', '"policy":"explicit","policy":"default_both"', 1), good_payload.replace('"active_expires_at":null', '"active_expires_at":null,"active_expires_at":null', 1), good_payload + "{}", ] bad = [subprocess.run(args, input=payload, text=True, capture_output=True, check=True) for payload in bad_payloads] decimal_request = _request([ _candidate(account_id="z-lower", weekly_pct="LOW"), _candidate(runner_id=4, account_id="a-higher", weekly_pct="HIGH")]) decimal_payload = json.dumps(decimal_request, separators=(",", ":")) decimal_payload = decimal_payload.replace( '"LOW"', "50.00000000000000000000000000001") decimal_payload = decimal_payload.replace( '"HIGH"', "50.00000000000000000000000000002") decimal_response = subprocess.run( args, input=decimal_payload, text=True, capture_output=True, check=True) version = subprocess.run([str(binary), "--version"], text=True, capture_output=True, check=True).stdout.strip() version_script = ( "import {resolveVersion} from './scripts/auto-version.mjs';" "console.log(resolveVersion('2.0.0','1.0.21'));" "console.log(resolveVersion('2.0.0','2.0.0'));" ) versions = subprocess.run( ["node", "--input-type=module", "-e", version_script], cwd=ROOT, text=True, capture_output=True, check=True).stdout.splitlines() self.assertTrue(json.loads(good.stdout)["ok"]) self.assertTrue(all(json.loads(item.stdout)["error_code"] == "invalid_request" for item in bad)) self.assertEqual(json.loads(decimal_response.stdout)["account_id"], "z-lower") self.assertEqual(version, "2.0.1") # npm package versioning, unrelated to SELECTOR_VERSION: auto-version bumps the # next patch above what is published. self.assertEqual(versions, ["2.0.0", "2.0.1"]) class HeadroomBandTests(unittest.TestCase): """Accounts within `headroom_band` points of the leader are interchangeable. Ranking strictly by headroom sends every task to whichever account is on top until it is spent below the runner-up, which is how one account's weekly limit was burned to zero while three others sat idle. """ def _pool(self, *specs) -> list[dict]: # spec: (account_id, weekly_pct, last_selected_at) return [_candidate(account_id=name, weekly_pct=pct, session_pct=0, reservation_history={"active_expires_at": None, "last_selected_at": last}) for name, pct, last in specs] def test_band_spreads_launches_instead_of_stacking_on_the_leader(self): # headroom 100 / 80 / 70 / 60 — the first three are within 30 of the leader. pool = self._pool(("a", 0, None), ("b", 20, None), ("c", 30, None), ("d", 40, None)) picked = {} for index in range(120): response = select(_request(pool, reservation_key=f"launch-{index}")) self.assertTrue(response["ok"]) picked[response["account_id"]] = picked.get(response["account_id"], 0) + 1 self.assertEqual(response["band_floor"], "70") self.assertEqual(response["band_count"], 3) self.assertEqual(set(picked), {"a", "b", "c"}, "every banded account must take work; 'd' is out of band") # No account may take more than half: the point is spreading the burn. self.assertLess(max(picked.values()), 60, picked) self.assertGreater(min(picked.values()), 20, picked) def test_same_request_always_selects_the_same_account(self): pool = self._pool(("a", 0, None), ("b", 20, None), ("c", 30, None)) first = select(_request(pool, reservation_key="stable")) for _ in range(5): repeat = select(_request(pool, reservation_key="stable")) self.assertEqual(repeat["account_id"], first["account_id"]) self.assertEqual(repeat["selection_digest"], first["selection_digest"]) def test_least_recently_used_wins_inside_the_band(self): # The emptiest account (a) was used most recently; the band rotates past it. pool = self._pool(("a", 0, "2026-08-25T08:59:00.000000Z"), ("b", 20, "2026-08-25T08:00:00.000000Z"), ("c", 30, "2026-08-25T07:00:00.000000Z")) self.assertEqual(select(_request(pool))["account_id"], "c") # A never-used account outranks every used one. pool = self._pool(("a", 0, "2026-08-25T08:59:00.000000Z"), ("b", 20, None)) self.assertEqual(select(_request(pool))["account_id"], "b") def test_out_of_band_account_never_wins_however_idle(self): pool = self._pool(("a", 0, "2026-08-25T08:59:59.000000Z"), ("d", 40, "2020-01-01T00:00:00.000000Z")) response = select(_request(pool)) self.assertEqual(response["account_id"], "a") self.assertEqual(response["band_count"], 1) def test_zero_band_restores_strict_most_headroom(self): pool = self._pool(("a", 0, "2026-08-25T08:59:00.000000Z"), ("b", 20, None)) self.assertEqual(select(_request(pool, headroom_band=0))["account_id"], "a") def test_band_is_configurable_and_validated(self): pool = self._pool(("a", 0, None), ("d", 40, None)) wide = select(_request(pool, headroom_band="45")) self.assertEqual(wide["band_floor"], "55") self.assertEqual(wide["band_count"], 2) for bad in ("abc", True, [], "1e5", None): refusal = select(_request(pool, headroom_band=bad)) self.assertFalse(refusal["ok"], bad) self.assertEqual(refusal["error_detail"], {"field": "headroom_band"}, bad) def test_unknown_quota_never_enters_the_band(self): # A row with no telemetry has no headroom to compare; it must not become # "as good as the leader" just because the band is generous. pool = self._pool(("a", 0, None)) pool.append(_candidate(account_id="blind", weekly_pct=None, session_pct=None, reservation_history={"active_expires_at": None, "last_selected_at": None})) response = select(_request(pool)) self.assertEqual(response["account_id"], "a") self.assertEqual(response["band_count"], 1) # ...and with NO usable telemetry anywhere, the band cannot apply at all. blind = [_candidate(account_id=name, weekly_pct=None, session_pct=None) for name in ("x", "y")] response = select(_request(blind)) self.assertTrue(response["ok"]) self.assertIsNone(response["band_floor"]) def test_band_participates_in_the_selection_proof(self): pool = self._pool(("a", 0, None), ("b", 20, None)) narrow = select(_request(pool, headroom_band="5")) wide = select(_request(pool, headroom_band="50")) self.assertNotEqual(narrow["selection_digest"], wide["selection_digest"]) # An omitted band is the documented default, and says so in the response. self.assertEqual(select(_request(pool))["headroom_band"], "30") if __name__ == "__main__": unittest.main()