"""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"], "2c9607e3dc1fde3ddf4e4469fbe48cf0e6a3c8114ef0c6b5f3c82f8ed4f78e3e") 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"], "2f018ab0e308a8465898e75a867e18d8620c03b4a49109edc1d2d590bf0d1121") 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") self.assertEqual(versions, ["2.0.0", "2.0.1"]) if __name__ == "__main__": unittest.main()