import json from dataclasses import replace from pathlib import Path from core.base_mailbox import MailboxAccount from core.settings import AppSettings from core.cfmail_domain_rotation import DomainHealthTracker from core.cfmail_provisioner import ProvisionResult from main import RegistrationBurstScheduler, RegistrationLoop def _base_settings(**overrides) -> AppSettings: settings = AppSettings( cleanup_enabled=False, register_sleep_min=0, register_sleep_max=0, register_mail_provider="mailtm,mailgw", ) return replace(settings, **overrides) def test_registration_loop_switches_provider_after_configured_failures(monkeypatch) -> None: settings = _base_settings(register_max_consecutive_failures=2) loop = RegistrationLoop(settings) loop._providers = ["mailtm", "mailgw"] calls: list[str] = [] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] calls.append(kwargs["mail_provider"]) if len(calls) >= 3: loop._stop_event.set() return {"success": False, "error_message": "failed"} monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="mailtm") assert calls == ["mailtm", "mailtm", "mailgw"] def test_registration_loop_stops_when_target_count_reached(monkeypatch) -> None: settings = _base_settings(register_target_count=2, register_mail_provider="mailtm") loop = RegistrationLoop(settings) loop._providers = ["mailtm"] calls: list[str] = [] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] calls.append(kwargs["mail_provider"]) return {"success": True, "email": f"user{len(calls)}@example.com"} monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr(loop, "_sync_cpa_from_success", lambda result, thread_id: (True, "", str(result.get("email") or ""))) loop._worker(thread_id=1, initial_provider="mailtm") assert calls == ["mailtm", "mailtm"] assert loop._target_reached.is_set() is True def test_registration_loop_uses_register_proxy_when_proxy_pool_disabled(monkeypatch) -> None: settings = _base_settings(register_proxy="http://127.0.0.1:7890", register_mail_provider="mailtm") loop = RegistrationLoop(settings) loop._providers = ["mailtm"] proxies: list[str | None] = [] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] proxies.append(kwargs["proxy"]) loop._stop_event.set() return {"success": False, "error_message": "failed"} monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="mailtm") assert proxies == ["http://127.0.0.1:7890"] def test_registration_loop_passes_selected_cfmail_profile_into_register_flow(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] recorded: dict[str, object] = {} monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [{"name": "cfmail-tw", "domain": "tw.example.test"}], ) def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] recorded["cfmail_profile_name"] = kwargs["cfmail_profile_name"] loop._stop_event.set() return {"success": True, "email": "demo@example.com", "metadata": {"email_domain": "tw.example.test"}} monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr(loop, "_sync_cpa_from_success", lambda result, thread_id: (True, "", str(result.get("email") or ""))) loop._worker(thread_id=1, initial_provider="cfmail") assert recorded["cfmail_profile_name"] == "cfmail-tw" def test_registration_loop_snapshot_exposes_cfmail_domain_pool(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [ {"name": "cfmail-tw", "domain": "tw.example.test"}, {"name": "cfmail-sg", "domain": "sg.example.test"}, ], ) loop._recent_attempts.extend( [ { "timestamp": "2026-03-30T22:50:00", "success": True, "stage": "completed", "signal": "", "error_message": "", "email_domain": "tw.example.test", "proxy_key": "tw-node", "email": "ok@tw.example.test", }, { "timestamp": "2026-03-30T22:50:10", "success": False, "stage": "create_account", "signal": "mailbox_reused", "error_message": "create account failed", "email_domain": "sg.example.test", "create_account_error_code": "user_already_exists", "proxy_key": "sg-node", "email": "dup@sg.example.test", }, ] ) loop._cfmail_flow_state["inflight_by_thread"] = { 1: {"domain": "tw.example.test", "profile_name": "cfmail-tw"}, } loop._cfmail_flow_state["last_started_by_domain"] = { "tw.example.test": 100.0, "sg.example.test": 90.0, } snapshot = loop.snapshot() domain_pool = snapshot["cfmail_domain_pool"] assert domain_pool["active_count"] == 2 assert [item["domain"] for item in domain_pool["active_domains"]] == [ "tw.example.test", "sg.example.test", ] assert domain_pool["active_domains"][0]["inflight"] == 1 assert domain_pool["active_domains"][0]["recent_success"] == 1 assert domain_pool["active_domains"][1]["recent_failure"] == 1 assert domain_pool["active_domains"][1]["failure_signals"]["mailbox_reused"] == 1 def test_registration_loop_uses_multi_domain_flow_defaults(monkeypatch) -> None: monkeypatch.delenv("ZHUCE6_CFMAIL_START_INTERVAL_SECONDS", raising=False) monkeypatch.delenv("ZHUCE6_CFMAIL_MAX_INFLIGHT", raising=False) monkeypatch.delenv("ZHUCE6_CFMAIL_ACTIVE_DOMAIN_COUNT", raising=False) monkeypatch.delenv("ZHUCE6_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET", raising=False) loop = RegistrationLoop(_base_settings(register_mail_provider="cfmail")) assert loop._cfmail_start_interval_seconds == 8 assert loop._cfmail_max_inflight == 4 assert loop._cfmail_active_domain_count == 3 assert loop._cfmail_fresh_domain_attempt_budget == 2 assert loop._cfmail_add_phone_window == 10 assert loop._cfmail_add_phone_threshold == 3 assert loop._cfmail_wait_otp_window == 6 assert loop._cfmail_wait_otp_threshold == 2 def test_registration_loop_caps_pending_token_retry_delay_to_one_minute(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_PENDING_TOKEN_RETRY_DELAY_SECONDS", "300") loop = RegistrationLoop(_base_settings()) assert loop._pending_token_retry_delay_seconds == 60 def test_registration_loop_uses_proxy_pool_and_releases_lease(monkeypatch) -> None: settings = _base_settings(register_mail_provider="mailtm") loop = RegistrationLoop(settings) loop._providers = ["mailtm"] recorded: dict[str, object] = {} class FakeLease: name = "sg-node" local_port = 17891 proxy_url = "socks5://127.0.0.1:17891" class FakePool: def acquire(self, timeout=5.0, preferred_name=None, preferred_regions=()): # type: ignore[no-untyped-def] recorded["timeout"] = timeout recorded["preferred_name"] = preferred_name recorded["preferred_regions"] = tuple(preferred_regions) return FakeLease() def release(self, lease, *, success, stage=None): # type: ignore[no-untyped-def] recorded["released"] = (lease.proxy_url, success, stage) def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] recorded["proxy"] = kwargs["proxy"] loop._stop_event.set() return {"success": True, "email": "demo@example.com"} loop._proxy_pool = FakePool() monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr(loop, "_sync_cpa_from_success", lambda result, thread_id: (True, "", str(result.get("email") or ""))) loop._worker(thread_id=1, initial_provider="mailtm") assert recorded["proxy"] == "socks5://127.0.0.1:17891" assert recorded["released"] == ("socks5://127.0.0.1:17891", True, "completed") def test_registration_loop_prefers_fresh_proxy_regions(monkeypatch) -> None: settings = _base_settings(register_mail_provider="mailtm", register_fresh_proxy_regions=("tw", "sg")) loop = RegistrationLoop(settings) loop._providers = ["mailtm"] recorded: dict[str, object] = {} class FakeLease: name = "tw-node" local_port = 17891 proxy_url = "socks5://127.0.0.1:17891" class FakePool: def acquire(self, timeout=5.0, preferred_name=None, preferred_regions=()): # type: ignore[no-untyped-def] recorded["timeout"] = timeout recorded["preferred_name"] = preferred_name recorded["preferred_regions"] = tuple(preferred_regions) return FakeLease() def release(self, lease, *, success, stage=None): # type: ignore[no-untyped-def] recorded["released"] = (lease.proxy_url, success, stage) def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] recorded["proxy"] = kwargs["proxy"] loop._stop_event.set() return {"success": True, "email": "demo@example.com"} loop._proxy_pool = FakePool() monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr(loop, "_sync_cpa_from_success", lambda result, thread_id: (True, "", str(result.get("email") or ""))) loop._worker(thread_id=1, initial_provider="mailtm") assert recorded["preferred_regions"] == ("tw", "sg") def test_registration_loop_starts_proxy_pool_for_direct_urls(monkeypatch) -> None: settings = _base_settings( register_mail_provider="mailtm", register_threads=1, proxy_pool_direct_urls="http://5.6.7.8:8080", ) loop = RegistrationLoop(settings) recorded: dict[str, object] = {} class FakePool: def start(self) -> None: recorded["pool_started"] = True class FakeThread: def __init__(self, target=None, args=(), daemon=None, name=None): # type: ignore[no-untyped-def] recorded["thread_args"] = {"target": target, "args": args, "daemon": daemon, "name": name} def start(self) -> None: recorded["thread_started"] = True monkeypatch.setattr("main.load_all", lambda: None) monkeypatch.setattr("core.proxy_pool.ProxyPool.from_settings", lambda settings: FakePool()) monkeypatch.setattr("main.threading.Thread", FakeThread) monkeypatch.setattr("core.registration._maybe_reconcile_cpa_runtime", lambda **kwargs: None) monkeypatch.setattr(loop, "_write_runtime_state", lambda: None) loop.start() assert recorded["pool_started"] is True assert recorded["thread_started"] is True assert loop._proxy_pool is not None def test_registration_loop_start_reconciles_pool_backups_to_cpa(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_mail_provider="mailtm", register_threads=1, backend="cpa", cpa_runtime_reconcile_enabled=True, pool_dir=tmp_path / "pool", ) loop = RegistrationLoop(settings) recorded: dict[str, object] = {} class FakeThread: def __init__(self, target=None, args=(), daemon=None, name=None): # type: ignore[no-untyped-def] recorded["thread_args"] = {"target": target, "args": args, "daemon": daemon, "name": name} def start(self) -> None: recorded["thread_started"] = True monkeypatch.setattr("main.load_all", lambda: None) monkeypatch.setattr("main.threading.Thread", FakeThread) monkeypatch.setattr("core.registration.create_backend_client", lambda settings: "backend-client") monkeypatch.setattr( "core.registration._maybe_reconcile_cpa_runtime", lambda **kwargs: recorded.setdefault("reconcile", kwargs), ) monkeypatch.setattr(loop, "_write_runtime_state", lambda: None) loop.start() assert recorded["thread_started"] is True assert recorded["reconcile"]["pool_dir"] == settings.pool_dir assert recorded["reconcile"]["client"] == "backend-client" def test_registration_loop_start_normalizes_cfmail_to_domain_pool(monkeypatch) -> None: settings = _base_settings( register_mail_provider="cfmail", register_threads=1, backend="cpa", cpa_runtime_reconcile_enabled=False, ) loop = RegistrationLoop(settings) recorded: dict[str, object] = {} class FakeThread: def __init__(self, target=None, args=(), daemon=None, name=None): # type: ignore[no-untyped-def] recorded["thread_args"] = {"target": target, "args": args, "daemon": daemon, "name": name} def start(self) -> None: recorded["thread_started"] = True class FakeProvisioner: def __init__(self, *, proxy_url=None, **_kwargs): # type: ignore[no-untyped-def] recorded["proxy_url"] = proxy_url def normalize_to_domain_pool(self, target_count): # type: ignore[no-untyped-def] recorded["normalized"] = True recorded["target_count"] = target_count return { "active_domains": ["auto-live.example.test", "auto-next.example.test"], "provisioned_domains": ["auto-next.example.test"], "retired_domains": [], } class FakeManager: def reload_if_needed(self, force=False): # type: ignore[no-untyped-def] recorded["reload_force"] = force return True monkeypatch.setattr("main.load_all", lambda: None) monkeypatch.setattr("main.threading.Thread", FakeThread) monkeypatch.setattr("core.cfmail_provisioner.CfmailProvisioner", FakeProvisioner) monkeypatch.setattr("core.cfmail.DEFAULT_CFMAIL_MANAGER", FakeManager()) monkeypatch.setattr(loop, "_ensure_cfmail_active_domain_ready", lambda: True) monkeypatch.setattr(loop, "_write_runtime_state", lambda: None) monkeypatch.setattr(loop, "_log", lambda message: recorded.setdefault("logs", []).append(message)) loop.start() assert recorded["thread_started"] is True assert recorded["proxy_url"] == settings.register_proxy assert recorded["normalized"] is True assert recorded["target_count"] == 3 assert recorded["reload_force"] is True assert any("normalized active domain pool" in line for line in recorded["logs"]) def test_registration_loop_retries_device_id_once_with_new_proxy(monkeypatch) -> None: settings = _base_settings(register_mail_provider="mailtm") loop = RegistrationLoop(settings) loop._providers = ["mailtm"] recorded: dict[str, object] = {"releases": []} class FakeLease: def __init__(self, name: str, local_port: int) -> None: self.name = name self.local_port = local_port self.proxy_url = f"socks5://127.0.0.1:{local_port}" class FakePool: def __init__(self) -> None: self._leases = [FakeLease("tw-bad", 17891), FakeLease("sg-good", 17892)] def acquire(self, timeout=5.0, preferred_name=None, preferred_regions=()): # type: ignore[no-untyped-def] recorded.setdefault("timeouts", []).append(timeout) return self._leases.pop(0) def release(self, lease, *, success, stage=None): # type: ignore[no-untyped-def] recorded["releases"].append((lease.name, success, stage)) attempts: list[str] = [] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] attempts.append(kwargs["proxy"]) if len(attempts) == 1: return { "success": False, "stage": "device_id", "error_message": "device id acquisition failed", } loop._stop_event.set() return {"success": True, "stage": "completed", "email": "demo@example.com"} loop._proxy_pool = FakePool() monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr(loop, "_sync_cpa_from_success", lambda result, thread_id: (True, "", str(result.get("email") or ""))) loop._worker(thread_id=1, initial_provider="mailtm") assert attempts == ["socks5://127.0.0.1:17891", "socks5://127.0.0.1:17892"] assert recorded["releases"] == [ ("tw-bad", False, "device_id"), ("sg-good", True, "completed"), ] snapshot = loop.snapshot() assert snapshot["total_attempts"] == 1 assert snapshot["total_success"] == 1 assert snapshot["total_failure"] == 0 def test_registration_loop_syncs_cpa_immediately_after_success(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_mail_provider="mailtm", ) loop = RegistrationLoop(settings) loop._providers = ["mailtm"] pool_file = tmp_path / "fresh@example.com.json" pool_file.write_text(json.dumps({"email": "fresh@example.com", "access_token": "tok", "account_id": "acct"}), encoding="utf-8") recorded: dict[str, object] = {} def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": True, "email": "fresh@example.com", "pool_file": str(pool_file), "written_to_pool": True, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr("core.registration.get_management_key", lambda: "secret") # type: ignore[no-untyped-def] monkeypatch.setattr("main.classify_token_file", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("readiness probe should not gate direct CPA sync"))) # type: ignore[no-untyped-def] def fake_upload_to_cpa(token_data, api_url=None, api_key=None, proxy=None): # type: ignore[no-untyped-def] recorded["token_data"] = token_data recorded["api_url"] = api_url recorded["api_key"] = api_key recorded["proxy"] = proxy return True, "upload success" monkeypatch.setattr("platforms.chatgpt.cpa_upload.upload_to_cpa", fake_upload_to_cpa) loop._worker(thread_id=1, initial_provider="mailtm") assert recorded["token_data"]["email"] == "fresh@example.com" assert recorded["api_url"] == "http://127.0.0.1:8317" assert recorded["api_key"] == "secret" assert recorded["proxy"] is None payload = json.loads(pool_file.read_text(encoding="utf-8")) assert payload["cpa_sync_status"] == "synced" snapshot = loop.snapshot() assert snapshot["total_success"] == 1 assert snapshot["total_success_registered"] == 1 assert snapshot["total_cpa_sync_success"] == 1 assert snapshot["total_cpa_sync_failure"] == 0 assert snapshot["registered_success_rate"] == 100.0 assert snapshot["cpa_sync_success_rate"] == 100.0 def test_registration_loop_syncs_cpa_before_breaking_on_target_reached(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_mail_provider="mailtm", register_target_count=1, ) loop = RegistrationLoop(settings) loop._providers = ["mailtm"] pool_file = tmp_path / "target@example.com.json" pool_file.write_text(json.dumps({"email": "target@example.com", "access_token": "tok", "account_id": "acct"}), encoding="utf-8") recorded: dict[str, object] = {} def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs return { "success": True, "email": "target@example.com", "pool_file": str(pool_file), "written_to_pool": True, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr("core.registration.get_management_key", lambda: "secret") # type: ignore[no-untyped-def] monkeypatch.setattr("main.classify_token_file", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("readiness probe should not gate direct CPA sync"))) # type: ignore[no-untyped-def] def fake_upload_to_cpa(token_data, api_url=None, api_key=None, proxy=None): # type: ignore[no-untyped-def] recorded["token_data"] = token_data recorded["api_url"] = api_url recorded["api_key"] = api_key recorded["proxy"] = proxy return True, "upload success" monkeypatch.setattr("platforms.chatgpt.cpa_upload.upload_to_cpa", fake_upload_to_cpa) loop._worker(thread_id=1, initial_provider="mailtm") assert loop._target_reached.is_set() is True assert recorded["token_data"]["email"] == "target@example.com" payload = json.loads(pool_file.read_text(encoding="utf-8")) assert payload["cpa_sync_status"] == "synced" snapshot = loop.snapshot() assert snapshot["total_attempts"] == 1 assert snapshot["total_success"] == 1 assert snapshot["total_cpa_sync_success"] == 1 assert snapshot["total_cpa_sync_failure"] == 0 def test_registration_loop_marks_pool_backup_when_direct_cpa_sync_fails(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_mail_provider="mailtm", ) loop = RegistrationLoop(settings) loop._providers = ["mailtm"] pool_file = tmp_path / "fresh@example.com.json" pool_file.write_text(json.dumps({"email": "fresh@example.com", "access_token": "tok", "account_id": "acct"}), encoding="utf-8") def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": True, "email": "fresh@example.com", "pool_file": str(pool_file), "written_to_pool": True, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr("core.registration.get_management_key", lambda: "secret") # type: ignore[no-untyped-def] monkeypatch.setattr("main.classify_token_file", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("readiness probe should not gate direct CPA sync"))) # type: ignore[no-untyped-def] monkeypatch.setattr("platforms.chatgpt.cpa_upload.upload_to_cpa", lambda *args, **kwargs: (False, "unexpected EOF")) # type: ignore[no-untyped-def] loop._worker(thread_id=1, initial_provider="mailtm") payload = json.loads(pool_file.read_text(encoding="utf-8")) assert payload["backup_written"] is True assert payload["cpa_sync_status"] == "failed" assert payload["last_cpa_sync_error"] == "unexpected EOF" snapshot = loop.snapshot() assert snapshot["total_success"] == 0 assert snapshot["total_failure"] == 1 assert snapshot["total_cpa_sync_success"] == 0 assert snapshot["total_cpa_sync_failure"] == 1 assert snapshot["failure_by_stage"]["cpa_sync"] == 1 assert snapshot["failure_signals"]["cpa_sync_failed"] == 1 def test_registration_loop_holds_add_phone_gated_success_in_warmup(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_mail_provider="cfmail", ) loop = RegistrationLoop(settings) loop._providers = ["cfmail"] pool_file = tmp_path / "fresh@example.com.json" pool_file.write_text(json.dumps({"email": "fresh@example.com", "access_token": "tok", "account_id": "acct"}), encoding="utf-8") uploaded: list[str] = [] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": True, "email": "fresh@example.com", "pool_file": str(pool_file), "written_to_pool": True, "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "post_create_gate": "add_phone", }, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr("core.registration.get_management_key", lambda: "secret") # type: ignore[no-untyped-def] monkeypatch.setattr("main.classify_token_file", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("warmup gating should not use legacy readiness probe"))) # type: ignore[no-untyped-def] monkeypatch.setattr( "platforms.chatgpt.cpa_upload.upload_to_cpa", lambda token_data, api_url=None, api_key=None, proxy=None: uploaded.append(token_data["email"]) or (True, "ok"), ) loop._worker(thread_id=1, initial_provider="cfmail") assert uploaded == [] payload = json.loads(pool_file.read_text(encoding="utf-8")) assert payload["warmup_required"] is True assert payload["warmup_state"] == "pending" assert payload["cpa_sync_status"] == "warmup_pending" snapshot = loop.snapshot() assert snapshot["total_success"] == 0 assert snapshot["total_success_registered"] == 0 assert snapshot["total_cpa_sync_success"] == 0 assert snapshot["total_cpa_sync_failure"] == 0 assert snapshot["total_warmup_pending"] == 1 def test_registration_loop_marks_add_phone_backup_as_warmup_pending(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_mail_provider="cfmail", ) loop = RegistrationLoop(settings) loop._providers = ["cfmail"] pool_file = tmp_path / "fresh@example.com.json" pool_file.write_text( json.dumps({"email": "fresh@example.com", "access_token": "tok", "account_id": "acct"}), encoding="utf-8", ) def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": True, "email": "fresh@example.com", "pool_file": str(pool_file), "written_to_pool": True, "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "post_create_gate": "add_phone", }, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) monkeypatch.setattr("core.registration.get_management_key", lambda: "secret") # type: ignore[no-untyped-def] monkeypatch.setattr("main.classify_token_file", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("readiness probe should not gate direct CPA sync"))) # type: ignore[no-untyped-def] monkeypatch.setattr("platforms.chatgpt.cpa_upload.upload_to_cpa", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("warmup pending account should not upload immediately"))) # type: ignore[no-untyped-def] loop._worker(thread_id=1, initial_provider="cfmail") payload = json.loads(pool_file.read_text(encoding="utf-8")) assert payload["backup_written"] is True assert payload["cpa_sync_status"] == "warmup_pending" assert payload["last_cpa_sync_error"] == "warmup pending" def test_registration_loop_enqueue_pending_token_preserves_proxy_provenance() -> None: loop = RegistrationLoop(_base_settings()) loop._enqueue_pending_token( { "metadata": { "deferred_credentials": { "email": "fresh@example.com", "password": "pw-secret", "registration_proxy_key": "台湾♣备用-1", "registration_proxy_region": "tw", "registration_proxy_url": "socks5://127.0.0.1:17891", "registration_fingerprint_profile": "chrome120_win", "cfmail_profile_name": "cfmail-tw", "add_phone_trace_path": "/tmp/add-phone-trace.json", } } }, thread_id=1, ) assert len(loop._pending_token_queue) == 1 entry = loop._pending_token_queue[0] assert entry["registration_proxy_key"] == "台湾♣备用-1" assert entry["registration_proxy_region"] == "tw" assert entry["registration_proxy_url"] == "socks5://127.0.0.1:17891" assert entry["registration_fingerprint_profile"] == "chrome120_win" assert entry["cfmail_profile_name"] == "cfmail-tw" assert entry["add_phone_trace_path"] == "/tmp/add-phone-trace.json" def test_registration_loop_retry_pending_token_reuses_registration_proxy(monkeypatch, tmp_path: Path) -> None: settings = _base_settings(pool_dir=tmp_path) loop = RegistrationLoop(settings) recorded: dict[str, object] = {} class FakeLease: name = "台湾♣备用-1" proxy_url = "socks5://127.0.0.1:17891" class FakePool: def acquire(self, timeout=5.0, preferred_name=None, preferred_regions=()): # type: ignore[no-untyped-def] recorded["preferred_name"] = preferred_name recorded["preferred_regions"] = tuple(preferred_regions) return FakeLease() def release(self, lease, *, success, stage=None): # type: ignore[no-untyped-def] recorded["released"] = (lease.proxy_url, success, stage) class FakeMailbox: def __init__(self, manager=None): # type: ignore[no-untyped-def] self.manager = manager class FakeAdapter: def __init__(self, mailbox): # type: ignore[no-untyped-def] self.mailbox = mailbox self._account = None class FakeEngine: def __init__(self, email_service, proxy_url=None): # type: ignore[no-untyped-def] recorded["engine_proxy_url"] = proxy_url self.email_service = email_service self.email = "" self.password = "" def _login_for_token(self): # type: ignore[no-untyped-def] return { "access_token": "access-token", "refresh_token": "refresh-token", "id_token": "", "account_id": "acct-123", "last_refresh": "2026-03-31T00:00:00Z", "expired": "2026-04-01T00:00:00Z", } def fake_write_token_record(token_data, pool_dir): # type: ignore[no-untyped-def] recorded["token_data"] = dict(token_data) path = Path(pool_dir) / "fresh@example.com.json" path.write_text(json.dumps(token_data), encoding="utf-8") return path loop._proxy_pool = FakePool() monkeypatch.setattr("core.registration.CfMailMailbox", FakeMailbox, raising=False) monkeypatch.setattr("core.registration.MailboxEmailServiceAdapter", FakeAdapter, raising=False) monkeypatch.setattr("platforms.chatgpt.register.RegistrationEngine", FakeEngine) monkeypatch.setattr("platforms.chatgpt.pool.write_token_record", fake_write_token_record) monkeypatch.setattr(loop, "_sync_cpa_from_success", lambda result, thread_id: (True, "", "fresh@example.com")) entry = { "email": "fresh@example.com", "password": "pw-secret", "mailbox_jwt": "", "mailbox_extra": {}, "registration_proxy_key": "台湾♣备用-1", "registration_proxy_region": "tw", "registration_proxy_url": "socks5://127.0.0.1:17891", "registration_fingerprint_profile": "chrome120_win", "cfmail_profile_name": "cfmail-tw", "retry_count": 0, } loop._retry_pending_token(entry) assert recorded["preferred_name"] == "台湾♣备用-1" assert recorded["preferred_regions"] == ("tw",) assert recorded["engine_proxy_url"] == "socks5://127.0.0.1:17891" assert recorded["released"] == ("socks5://127.0.0.1:17891", True, "deferred_retry") token_data = recorded["token_data"] assert token_data["registration_proxy_key"] == "台湾♣备用-1" assert token_data["registration_proxy_region"] == "tw" assert token_data["registration_proxy_url"] == "socks5://127.0.0.1:17891" assert token_data["registration_fingerprint_profile"] == "chrome120_win" assert token_data["registration_cfmail_profile_name"] == "cfmail-tw" def test_registration_loop_dumps_engine_logs_on_failure(monkeypatch) -> None: settings = _base_settings(register_max_consecutive_failures=1, register_mail_provider="mailtm") loop = RegistrationLoop(settings) loop._providers = ["mailtm"] logged: list[str] = [] original_log = loop._log def capture_log(msg: str) -> None: logged.append(msg) loop._log = capture_log def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] loop._stop_event.set() return { "success": False, "stage": "signup", "error_message": "HTTP 403: access denied", "logs": [ "[09:35:20] check_ip_location: JP", "[09:35:21] created mailbox: test@example.com", "[09:35:22] signup form status: 403", ], } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="mailtm") # Verify stage appears in the failure line fail_lines = [line for line in logged if "failed" in line and "stage=" in line] assert len(fail_lines) == 1 assert "[stage=signup]" in fail_lines[0] assert "HTTP 403: access denied" in fail_lines[0] # Verify engine logs are dumped with ↳ prefix engine_lines = [line for line in logged if "\u21b3" in line] assert len(engine_lines) == 3 assert "signup form status: 403" in engine_lines[2] snapshot = loop.snapshot() assert snapshot["failure_by_stage"]["signup"] == 1 assert snapshot["recent_failure_hotspots"] == [{"key": "signup", "stage": "signup", "count": 1}] def test_registration_loop_rotates_cfmail_domain_after_blacklist_threshold(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail", register_max_consecutive_failures=5) loop = RegistrationLoop(settings) loop._providers = ["cfmail"] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) monkeypatch.setattr(loop, "_ensure_cfmail_domain_pool_target", lambda **_: None) rotation_calls: list[str] = [] class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] rotation_calls.append("rotate") return ProvisionResult( success=True, step="completed", old_domain="nova.example.test", new_domain="auto0322.example.test", ) def provision_additional_domain(self): # type: ignore[no-untyped-def] return ProvisionResult(success=False, step="provision_additional_domain", error="not needed") loop._cfmail_provisioner = FakeProvisioner() attempts = {"count": 0} def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs attempts["count"] += 1 if attempts["count"] >= 2: loop._stop_event.set() return { "success": False, "stage": "create_account", "error_message": "create account failed", "metadata": { "mail_provider": "cfmail", "email_domain": "nova.example.test", "create_account_error_code": "registration_disallowed", "create_account_error_message": "blocked", }, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") assert rotation_calls == ["rotate"] assert loop.snapshot()["cfmail_rotation"]["last_new_domain"] == "auto0322.example.test" def test_registration_loop_does_not_rotate_cfmail_on_mailbox_failure(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail", register_max_consecutive_failures=2) loop = RegistrationLoop(settings) loop._providers = ["cfmail"] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) monkeypatch.setattr(loop, "_ensure_cfmail_domain_pool_target", lambda **_: None) class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] raise AssertionError("rotation should not be called") def provision_additional_domain(self): # type: ignore[no-untyped-def] return ProvisionResult(success=False, step="provision_additional_domain", error="not needed") loop._cfmail_provisioner = FakeProvisioner() def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": False, "stage": "mailbox", "error_message": "create email failed", "metadata": {"mail_provider": "cfmail"}, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") assert loop.snapshot()["cfmail_rotation"]["last_new_domain"] == "" def test_registration_loop_rotates_cfmail_on_invalid_domain_mailbox_failure() -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] return ProvisionResult( success=True, step="rotate", old_domain="bad.example.test", new_domain="auto-new.example.test", ) reload_calls: list[bool] = [] class FakeManager: def reload_if_needed(self, force=False): # type: ignore[no-untyped-def] reload_calls.append(bool(force)) return True loop._cfmail_manager = FakeManager() loop._cfmail_provisioner = FakeProvisioner() loop._cfmail_wait_otp_cooldown_seconds = 60 loop._cfmail_add_phone_cooldown_seconds = 60 result = { "success": False, "stage": "mailbox", "error_message": "create email failed", "logs": ["[10:00:00] create_email failed: 创建邮箱地址失败: 无效的域名"], "metadata": { "mail_provider": "cfmail", "email_domain": "bad.example.test", }, } assert loop._force_rotate_cfmail_for_invalid_mailbox(thread_id=1, result=result) is True snapshot = loop.snapshot() assert snapshot["cfmail_rotation"]["last_new_domain"] == "auto-new.example.test" assert snapshot["cfmail_wait_otp_stoploss"]["active_domain"] == "auto-new.example.test" assert reload_calls == [True] def test_registration_loop_startup_reports_active_domain_pool_ready(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) logs: list[str] = [] class FakeManager: accounts = [ type("Account", (), {"name": "cfmail-a", "email_domain": "a.example.test"})(), type("Account", (), {"name": "cfmail-b", "email_domain": "b.example.test"})(), ] def skip_remaining_seconds(self, account_name): # type: ignore[no-untyped-def] del account_name return 0 loop._cfmail_manager = FakeManager() loop._cfmail_provisioner = object() monkeypatch.setattr(loop, "_log", logs.append) assert loop._ensure_cfmail_active_domain_ready() is True assert any("active-domain pool ready" in line for line in logs) def test_registration_loop_forces_cfmail_rotation_when_all_accounts_in_cooldown(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail", register_max_consecutive_failures=2) loop = RegistrationLoop(settings) loop._providers = ["cfmail"] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) run_calls: list[str] = [] rotation_calls: list[str] = [] class FakeManager: def reload_if_needed(self) -> bool: return False def select_account(self, profile_name=None): # type: ignore[no-untyped-def] del profile_name return None class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] rotation_calls.append("rotate") loop._stop_event.set() return ProvisionResult( success=True, step="completed", old_domain="auto-old.example.test", new_domain="auto-new.example.test", ) def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() run_calls.append("called") raise AssertionError("registration should not run while cfmail is fully unavailable") loop._cfmail_manager = FakeManager() loop._cfmail_provisioner = FakeProvisioner() monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") assert rotation_calls == ["rotate"] assert run_calls == [] assert loop.snapshot()["cfmail_rotation"]["last_new_domain"] == "auto-new.example.test" def test_registration_loop_does_not_penalize_proxy_for_blacklist_failure(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] recorded: dict[str, object] = {} class FakeLease: name = "sg-node" local_port = 17891 proxy_url = "socks5://127.0.0.1:17891" class FakePool: def acquire(self, timeout=5.0, preferred_name=None, preferred_regions=()): # type: ignore[no-untyped-def] return FakeLease() def release(self, lease, *, success, stage=None): # type: ignore[no-untyped-def] recorded["released"] = (success, stage) def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": False, "stage": "create_account", "error_message": "create account failed", "metadata": { "mail_provider": "cfmail", "email_domain": "nova.example.test", "create_account_error_code": "registration_disallowed", }, } loop._proxy_pool = FakePool() monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") assert recorded["released"] == (None, "create_account") def test_registration_loop_records_add_phone_gate_signal(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": False, "stage": "add_phone_gate", "error_message": "post-create flow requires phone gate", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "post_create_gate": "add_phone", }, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") snapshot = loop.snapshot() assert snapshot["failure_by_stage"]["add_phone_gate"] == 1 assert snapshot["failure_signals"]["add_phone_gate"] == 1 assert snapshot["recent_failure_hotspots"] == [{"key": "add_phone_gate", "stage": "add_phone_gate", "count": 1}] assert snapshot["recent_attempts"][0]["post_create_gate"] == "add_phone" def test_registration_loop_records_signup_invalid_auth_step_signal() -> None: loop = RegistrationLoop(_base_settings(register_mail_provider="cfmail")) loop._record_attempt( success=False, stage="signup", error_message='HTTP 400: {"error":{"code":"invalid_auth_step"}}', metadata={ "mail_provider": "cfmail", "email_domain": "demo.example.test", "signup_error_code": "invalid_auth_step", "signup_http_status": 400, }, proxy_key="台湾原生-01", email="demo@example.test", ) snapshot = loop.snapshot() assert snapshot["failure_signals"]["invalid_auth_step"] == 1 attempt = snapshot["recent_attempts"][0] assert attempt["signup_error_code"] == "invalid_auth_step" assert attempt["signup_http_status"] == 400 def test_registration_loop_classifies_mailbox_transport_failures() -> None: loop = RegistrationLoop(_base_settings(register_mail_provider="cfmail")) loop._record_attempt( success=False, stage="mailbox", error_message="create email failed", metadata={ "mail_provider": "cfmail", "email_domain": "demo.example.test", "mailbox_error_kind": "transport_error", "mailbox_error_stage": "create_email", }, proxy_key="新加坡原生-02", ) snapshot = loop.snapshot() assert snapshot["failure_signals"]["mailbox_create_transport_error"] == 1 attempt = snapshot["recent_attempts"][0] assert attempt["mailbox_error_kind"] == "transport_error" assert attempt["mailbox_error_stage"] == "create_email" def test_registration_loop_classifies_user_already_exists_as_mailbox_reused(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs loop._stop_event.set() return { "success": False, "stage": "create_account", "error_message": "create account failed", "email": "dup@example.com", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "create_account_error_code": "user_already_exists", }, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") snapshot = loop.snapshot() assert snapshot["failure_by_stage"]["create_account"] == 1 assert snapshot["failure_signals"]["mailbox_reused"] == 1 assert snapshot["recent_failure_hotspots"] == [{"key": "mailbox_reused", "stage": "create_account", "count": 1}] def test_registration_loop_activates_add_phone_stoploss(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_add_phone_window = 2 loop._cfmail_add_phone_threshold = 2 loop._cfmail_add_phone_max_successes = 0 loop._cfmail_add_phone_cooldown_seconds = 60 attempts = {"count": 0} def fake_run_chatgpt_register_once(**kwargs): # type: ignore[no-untyped-def] del kwargs attempts["count"] += 1 if attempts["count"] >= 2: loop._stop_event.set() return { "success": False, "stage": "add_phone_gate", "error_message": "post-create flow requires phone gate", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "post_create_gate": "add_phone", }, } monkeypatch.setattr("main.run_chatgpt_register_once", fake_run_chatgpt_register_once) loop._worker(thread_id=1, initial_provider="cfmail") snapshot = loop.snapshot() stoploss = snapshot["cfmail_add_phone_stoploss"] assert stoploss["active_domain"] == "demo.example.test" assert stoploss["in_cooldown"] is True assert stoploss["last_add_phone_failures"] == 2 assert stoploss["last_successes"] == 0 def test_registration_loop_snapshot_exposes_active_domain_only_attempts() -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._recent_attempts.extend( [ { "timestamp": "2026-03-24T18:00:00", "success": False, "stage": "create_account", "signal": "registration_disallowed", "error_message": "failed", "email_domain": "old.example.test", "proxy_key": "old-node", "email": "old@old.example.test", }, { "timestamp": "2026-03-24T18:00:10", "success": True, "stage": "completed", "signal": "", "error_message": "", "email_domain": "new.example.test", "proxy_key": "new-node-1", "email": "ok@new.example.test", }, { "timestamp": "2026-03-24T18:00:20", "success": False, "stage": "create_account", "signal": "registration_disallowed", "error_message": "failed", "email_domain": "new.example.test", "proxy_key": "new-node-2", "email": "bad@new.example.test", }, ] ) class FakeTracker: def snapshot(self) -> dict[str, object]: return { "active_domain": "new.example.test", "last_new_domain": "new.example.test", "last_blacklisted_domain": "old.example.test", } loop._cfmail_tracker = FakeTracker() snapshot = loop.snapshot() assert [item["email_domain"] for item in snapshot["active_domain_recent_attempts"]] == [ "new.example.test", "new.example.test", ] assert snapshot["active_domain_failure_by_stage"] == {"create_account": 1} assert snapshot["active_domain_failure_signals"] == {"registration_disallowed": 1} assert snapshot["active_domain_recent_failure_hotspots"] == [ {"key": "registration_disallowed", "stage": "create_account", "count": 1} ] def test_registration_loop_snapshot_infers_active_domain_from_recent_attempts() -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._recent_attempts.extend( [ { "timestamp": "2026-03-24T18:10:00", "success": False, "stage": "create_account", "signal": "registration_disallowed", "error_message": "failed", "email_domain": "old.example.test", "proxy_key": "old-node", "email": "old@old.example.test", }, { "timestamp": "2026-03-24T18:10:10", "success": True, "stage": "completed", "signal": "", "error_message": "", "email_domain": "new.example.test", "proxy_key": "new-node", "email": "ok@new.example.test", }, ] ) snapshot = loop.snapshot() assert [item["email_domain"] for item in snapshot["active_domain_recent_attempts"]] == [ "new.example.test" ] def test_registration_loop_disables_add_phone_cooldown_when_configured_zero(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_CFMAIL_ADD_PHONE_COOLDOWN_SECONDS", "0") settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._cfmail_add_phone_window = 2 loop._cfmail_add_phone_threshold = 2 loop._cfmail_add_phone_max_successes = 0 result = { "success": False, "stage": "add_phone_gate", "error_message": "post-create flow requires phone gate", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "post_create_gate": "add_phone", }, } loop._update_cfmail_add_phone_stoploss(result) loop._update_cfmail_add_phone_stoploss(result) stoploss = loop.snapshot()["cfmail_add_phone_stoploss"] assert stoploss["in_cooldown"] is False assert stoploss["cooldown_remaining_seconds"] == 0 def test_registration_loop_add_phone_stoploss_retires_triggered_domain_without_blocking_other_domains(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._cfmail_add_phone_window = 2 loop._cfmail_add_phone_threshold = 2 loop._cfmail_add_phone_max_successes = 0 loop._cfmail_add_phone_cooldown_seconds = 60 monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [ {"name": "cfmail-tw", "domain": "tw.example.test"}, {"name": "cfmail-sg", "domain": "sg.example.test"}, ], ) retired: list[str] = [] scheduled: list[str] = [] class FakeProvisioner: def retire_domain(self, domain: str) -> ProvisionResult: retired.append(domain) return ProvisionResult(success=True, step="retire_domain", old_domain=domain) loop._cfmail_provisioner = FakeProvisioner() # type: ignore[assignment] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) monkeypatch.setattr(loop, "_reload_cfmail_manager_after_rotation", lambda: None) monkeypatch.setattr( loop, "_schedule_cfmail_domain_pool_replenish", lambda *, trigger_thread_id, reason: scheduled.append(f"{trigger_thread_id}:{reason}"), raising=False, ) result = { "success": False, "stage": "add_phone_gate", "error_message": "post-create flow requires phone gate", "metadata": { "mail_provider": "cfmail", "email_domain": "sg.example.test", "post_create_gate": "add_phone", }, } loop._update_cfmail_add_phone_stoploss(result) loop._update_cfmail_add_phone_stoploss(result) assert loop._wait_if_cfmail_add_phone_stopped(thread_id=3, provider="cfmail") is False assert retired == ["sg.example.test"] assert scheduled == ["3:add_phone stoploss"] def test_registration_loop_activates_wait_otp_stoploss_for_no_message_timeouts(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_wait_otp_window = 2 loop._cfmail_wait_otp_threshold = 2 loop._cfmail_wait_otp_cooldown_seconds = 60 result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_message", "otp_mailbox_message_scan_count": 0, }, } loop._update_cfmail_wait_otp_stoploss(result) loop._update_cfmail_wait_otp_stoploss(result) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["active_domain"] == "demo.example.test" assert stoploss["in_cooldown"] is True assert stoploss["last_no_message_timeouts"] == 2 def test_registration_loop_wait_otp_stoploss_retires_triggered_domain_without_blocking_other_domains(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._cfmail_wait_otp_window = 2 loop._cfmail_wait_otp_threshold = 2 loop._cfmail_wait_otp_cooldown_seconds = 60 monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [ {"name": "cfmail-tw", "domain": "tw.example.test"}, {"name": "cfmail-sg", "domain": "sg.example.test"}, ], ) retired: list[str] = [] scheduled: list[str] = [] class FakeProvisioner: def retire_domain(self, domain: str) -> ProvisionResult: retired.append(domain) return ProvisionResult(success=True, step="retire_domain", old_domain=domain) loop._cfmail_provisioner = FakeProvisioner() # type: ignore[assignment] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) monkeypatch.setattr(loop, "_reload_cfmail_manager_after_rotation", lambda: None) monkeypatch.setattr( loop, "_schedule_cfmail_domain_pool_replenish", lambda *, trigger_thread_id, reason: scheduled.append(f"{trigger_thread_id}:{reason}"), raising=False, ) result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "sg.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_message", "otp_mailbox_message_scan_count": 0, }, } loop._update_cfmail_wait_otp_stoploss(result) loop._update_cfmail_wait_otp_stoploss(result) assert loop._wait_if_cfmail_wait_otp_stopped(thread_id=4, provider="cfmail") is False assert retired == ["sg.example.test"] assert scheduled == ["4:wait_otp stoploss"] def test_registration_loop_replenish_worker_retries_transient_provision_failure(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._cfmail_active_domain_count = 3 monkeypatch.setattr("core.registration.time.sleep", lambda _: None) active_accounts = [ {"name": "cfmail-tw", "domain": "tw.example.test"}, {"name": "cfmail-sg", "domain": "sg.example.test"}, ] monkeypatch.setattr(loop, "_current_cfmail_active_accounts", lambda: list(active_accounts)) monkeypatch.setattr(loop, "_reload_cfmail_manager_after_rotation", lambda: None) attempts = {"count": 0} class FakeProvisioner: def provision_additional_domain(self) -> ProvisionResult: attempts["count"] += 1 if attempts["count"] == 1: return ProvisionResult(success=False, step="provision_additional_domain", error="tls connect error") active_accounts.append({"name": "cfmail-jp", "domain": "jp.example.test"}) return ProvisionResult(success=True, step="provision_additional_domain", new_domain="jp.example.test") loop._cfmail_provisioner = FakeProvisioner() # type: ignore[assignment] loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) loop._cfmail_domain_pool_replenish_worker(trigger_thread_id=2, reason="fresh domain budget reached") assert attempts["count"] == 2 assert len(active_accounts) == 3 def test_registration_loop_schedules_replenish_when_usable_domain_pool_below_target(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._cfmail_active_domain_count = 3 monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [ {"name": "cfmail-tw", "domain": "tw.example.test"}, {"name": "cfmail-sg", "domain": "sg.example.test"}, ], ) loop._cfmail_provisioner = object() # type: ignore[assignment] scheduled: list[str] = [] monkeypatch.setattr( loop, "_schedule_cfmail_domain_pool_replenish", lambda *, trigger_thread_id, reason: scheduled.append(f"{trigger_thread_id}:{reason}"), raising=False, ) loop._ensure_cfmail_domain_pool_target(trigger_thread_id=5, reason="startup") assert scheduled == ["5:startup"] def test_registration_loop_does_not_activate_wait_otp_stoploss_when_window_contains_success(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_wait_otp_window = 2 loop._cfmail_wait_otp_threshold = 2 loop._cfmail_wait_otp_cooldown_seconds = 60 timeout_result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_message", "otp_mailbox_message_scan_count": 0, }, } success_result = { "success": True, "stage": "completed", "email": "ok@demo.example.test", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", }, } loop._update_cfmail_wait_otp_stoploss(timeout_result) loop._update_cfmail_wait_otp_stoploss(success_result) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["active_domain"] == "demo.example.test" assert stoploss["in_cooldown"] is False def test_registration_loop_does_not_activate_wait_otp_stoploss_when_window_contains_message_seen(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_wait_otp_window = 2 loop._cfmail_wait_otp_threshold = 2 loop._cfmail_wait_otp_cooldown_seconds = 60 timeout_result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_message", "otp_mailbox_message_scan_count": 0, }, } message_seen_result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_match", "otp_mailbox_message_scan_count": 1, }, } loop._update_cfmail_wait_otp_stoploss(timeout_result) loop._update_cfmail_wait_otp_stoploss(message_seen_result) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["active_domain"] == "demo.example.test" assert stoploss["in_cooldown"] is False def test_registration_loop_disables_cfmail_canary_gate() -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) assert loop._wait_if_cfmail_canary_pending(thread_id=2, provider="cfmail") is False assert loop._wait_if_cfmail_canary_pending(thread_id=1, provider="cfmail") is False def test_registration_loop_cfmail_canary_snapshot_is_disabled(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._arm_cfmail_canary("demo.example.test") loop._mark_cfmail_canary_ready("demo.example.test", reason="mailbox_message_seen") loop._update_cfmail_canary_after_result(thread_id=3, result={}) snapshot = loop.snapshot()["cfmail_canary"] assert snapshot["pending"] is False assert snapshot["active_domain"] == "" assert snapshot["last_ready_reason"] == "disabled" def test_registration_loop_tracks_fresh_domain_budget_after_mail_seen(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET", "2") settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") first_result = { "success": False, "stage": "add_phone_gate", "error_message": "post-create flow requires phone gate", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_mailbox_message_scan_count": 1, }, } second_result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_mailbox_message_scan_count": 0, }, } loop._update_cfmail_fresh_domain_budget(first_result) snapshot = loop.snapshot()["cfmail_fresh_domain_budget"] assert snapshot["completed_attempts"] == 1 assert snapshot["mail_seen_attempts"] == 1 assert snapshot["last_triggered_at"] == "" loop._update_cfmail_fresh_domain_budget(second_result) snapshot = loop.snapshot()["cfmail_fresh_domain_budget"] assert snapshot["completed_attempts"] == 2 assert snapshot["mail_seen_attempts"] == 1 assert snapshot["last_reason"] == "fresh_domain_attempt_budget_reached" def test_registration_loop_rotates_domain_when_fresh_domain_budget_reached(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET", "2") settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) loop._cfmail_fresh_domain_state.update( { "active_domain": "demo.example.test", "completed_attempts": 2, "mail_seen_attempts": 1, "successes": 0, "last_triggered_at": "2026-03-29T10:00:00", "last_rotation_attempted_at": "", "last_reason": "fresh_domain_attempt_budget_reached", } ) class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] return ProvisionResult( success=True, step="rotate", old_domain="demo.example.test", new_domain="auto-new.example.test", ) reload_calls: list[bool] = [] class FakeManager: def reload_if_needed(self, force=False): # type: ignore[no-untyped-def] reload_calls.append(bool(force)) return True loop._cfmail_manager = FakeManager() loop._cfmail_provisioner = FakeProvisioner() assert loop._rotate_cfmail_for_fresh_domain_budget(thread_id=5) is True snapshot = loop.snapshot() assert snapshot["cfmail_rotation"]["last_new_domain"] == "auto-new.example.test" assert snapshot["cfmail_fresh_domain_budget"]["active_domain"] == "auto-new.example.test" assert snapshot["cfmail_fresh_domain_budget"]["completed_attempts"] == 0 assert snapshot["cfmail_canary"]["pending"] is False assert reload_calls == [True] def test_registration_loop_throttles_cfmail_inflight_attempts(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_CFMAIL_MAX_INFLIGHT", "1") monkeypatch.setenv("ZHUCE6_CFMAIL_START_INTERVAL_SECONDS", "0") settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [{"name": "cfmail-demo", "domain": "demo.example.test"}], ) assert loop._wait_if_cfmail_flow_throttled(thread_id=1, provider="cfmail") is False assert loop._wait_if_cfmail_flow_throttled(thread_id=2, provider="cfmail") is True loop._release_cfmail_flow_slot(thread_id=1) assert loop._wait_if_cfmail_flow_throttled(thread_id=2, provider="cfmail") is False def test_registration_loop_throttles_cfmail_start_interval(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_CFMAIL_MAX_INFLIGHT", "2") monkeypatch.setenv("ZHUCE6_CFMAIL_START_INTERVAL_SECONDS", "15") settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr( loop, "_current_cfmail_active_accounts", lambda: [{"name": "cfmail-demo", "domain": "demo.example.test"}], ) current_time = {"value": 100.0} monkeypatch.setattr("core.registration.time.time", lambda: current_time["value"]) assert loop._wait_if_cfmail_flow_throttled(thread_id=1, provider="cfmail") is False loop._release_cfmail_flow_slot(thread_id=1) current_time["value"] = 105.0 assert loop._wait_if_cfmail_flow_throttled(thread_id=2, provider="cfmail") is True current_time["value"] = 116.0 assert loop._wait_if_cfmail_flow_throttled(thread_id=2, provider="cfmail") is False def test_registration_loop_activates_live_wait_otp_stoploss_on_stalled_waits(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_wait_otp_cooldown_seconds = 60 loop._cfmail_wait_otp_live_threshold = 2 loop._cfmail_wait_otp_live_age_seconds = 30 account_a = MailboxAccount( email="a@demo.example.test", account_id="jwt-a", extra={"email_domain": "demo.example.test"}, ) account_b = MailboxAccount( email="b@demo.example.test", account_id="jwt-b", extra={"email_domain": "demo.example.test"}, ) loop._on_cfmail_wait_progress( account_a, {"message_scan_count": 0, "elapsed_seconds": 35}, ) loop._on_cfmail_wait_progress( account_b, {"message_scan_count": 0, "elapsed_seconds": 36}, ) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["active_domain"] == "demo.example.test" assert stoploss["in_cooldown"] is True assert stoploss["last_reason"] == "live wait_otp no-message threshold reached" assert stoploss["last_no_message_timeouts"] == 2 def test_registration_loop_live_wait_otp_stoploss_can_be_disabled(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_wait_otp_cooldown_seconds = 60 loop._cfmail_wait_otp_live_threshold = 0 loop._cfmail_wait_otp_live_age_seconds = 30 account_a = MailboxAccount( email="a@demo.example.test", account_id="jwt-a", extra={"email_domain": "demo.example.test"}, ) account_b = MailboxAccount( email="b@demo.example.test", account_id="jwt-b", extra={"email_domain": "demo.example.test"}, ) loop._on_cfmail_wait_progress( account_a, {"message_scan_count": 0, "elapsed_seconds": 35}, ) loop._on_cfmail_wait_progress( account_b, {"message_scan_count": 0, "elapsed_seconds": 36}, ) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["in_cooldown"] is False def test_registration_loop_does_not_activate_wait_otp_stoploss_for_non_mailbox_timeout() -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._cfmail_wait_otp_window = 2 loop._cfmail_wait_otp_threshold = 2 loop._cfmail_wait_otp_cooldown_seconds = 60 result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "demo.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_match", "otp_mailbox_message_scan_count": 1, }, } loop._update_cfmail_wait_otp_stoploss(result) loop._update_cfmail_wait_otp_stoploss(result) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["in_cooldown"] is False def test_registration_loop_rotates_cfmail_domain_when_wait_otp_stoploss_active(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) loop._cfmail_wait_otp_cooldown_seconds = 60 loop._cfmail_wait_otp_state.update( { "active_domain": "demo.example.test", "in_cooldown": True, "cooldown_until": 9999999999.0, "last_triggered_at": "2026-03-29T10:00:00", "last_rotation_attempted_at": "", } ) class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] return ProvisionResult( success=True, step="rotate", old_domain="demo.example.test", new_domain="auto-new.example.test", ) reload_calls: list[bool] = [] class FakeManager: def reload_if_needed(self, force=False): # type: ignore[no-untyped-def] reload_calls.append(bool(force)) return True loop._cfmail_manager = FakeManager() loop._cfmail_provisioner = FakeProvisioner() assert loop._wait_if_cfmail_wait_otp_stopped(thread_id=1, provider="cfmail") is True snapshot = loop.snapshot() assert snapshot["cfmail_rotation"]["last_new_domain"] == "auto-new.example.test" assert snapshot["cfmail_wait_otp_stoploss"]["in_cooldown"] is False assert snapshot["cfmail_wait_otp_stoploss"]["active_domain"] == "auto-new.example.test" assert reload_calls == [True] def test_registration_loop_rotates_cfmail_domain_when_add_phone_stoploss_active(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) loop._providers = ["cfmail"] monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_tracker = DomainHealthTracker(window_size=2, blacklist_threshold=2, rotation_cooldown_seconds=1) loop._cfmail_add_phone_cooldown_seconds = 60 loop._cfmail_add_phone_state.update( { "active_domain": "demo.example.test", "in_cooldown": True, "cooldown_until": 9999999999.0, "last_triggered_at": "2026-03-29T10:00:00", "last_rotation_attempted_at": "", } ) class FakeProvisioner: def rotate_active_domain(self): # type: ignore[no-untyped-def] return ProvisionResult( success=True, step="rotate", old_domain="demo.example.test", new_domain="auto-new.example.test", ) loop._cfmail_provisioner = FakeProvisioner() assert loop._wait_if_cfmail_add_phone_stopped(thread_id=1, provider="cfmail") is True snapshot = loop.snapshot() assert snapshot["cfmail_rotation"]["last_new_domain"] == "auto-new.example.test" assert snapshot["cfmail_add_phone_stoploss"]["in_cooldown"] is False assert snapshot["cfmail_add_phone_stoploss"]["active_domain"] == "auto-new.example.test" def test_registration_loop_ignores_stale_wait_otp_result_from_old_domain(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "auto-new.example.test") result = { "success": False, "stage": "wait_otp", "error_message": "otp retrieval failed", "metadata": { "mail_provider": "cfmail", "email_domain": "auto-old.example.test", "otp_wait_failure_reason": "mailbox_timeout_no_message", "otp_mailbox_message_scan_count": 0, }, } loop._update_cfmail_wait_otp_stoploss(result) stoploss = loop.snapshot()["cfmail_wait_otp_stoploss"] assert stoploss["active_domain"] == "" assert stoploss["in_cooldown"] is False def test_registration_loop_does_not_abort_old_domain_wait_after_rotation(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "auto-new.example.test") loop._cfmail_wait_otp_state.update( { "active_domain": "auto-old.example.test", "in_cooldown": True, "cooldown_until": 9999999999.0, "last_triggered_at": "2026-03-29T10:00:00", } ) account = MailboxAccount( email="a@auto-old.example.test", account_id="jwt-old", extra={ "email_domain": "auto-old.example.test", "otp_wait_started_at": 1743242390.0, }, ) assert loop._should_abort_cfmail_wait(account) is False def test_registration_loop_aborts_wait_for_domain_in_wait_otp_cooldown(monkeypatch) -> None: settings = _base_settings(register_mail_provider="cfmail") loop = RegistrationLoop(settings) monkeypatch.setattr(loop, "_current_cfmail_active_domain", lambda: "demo.example.test") loop._cfmail_wait_otp_state.update( { "active_domain": "demo.example.test", "in_cooldown": True, "cooldown_until": 9999999999.0, } ) account = MailboxAccount( email="a@demo.example.test", account_id="jwt-demo", extra={ "email_domain": "demo.example.test", "otp_wait_started_at": 1743242410.0, }, ) assert loop._should_abort_cfmail_wait(account) is True def test_registration_burst_scheduler_runs_batch_with_batch_config(monkeypatch, tmp_path: Path) -> None: settings = _base_settings( register_log_file="", runtime_state_file=tmp_path / "runtime_state.json", register_batch_threads=1, register_batch_target_count=20, register_batch_interval_seconds=60, ) observed: dict[str, int] = {} class FakeLoop: def __init__(self, batch_settings): # type: ignore[no-untyped-def] observed["threads"] = batch_settings.register_threads observed["target"] = batch_settings.register_target_count self._threads = [] def start(self) -> None: return None def stop(self) -> None: return None def snapshot(self) -> dict[str, object]: return { "name": "register", "status": "stopped", "threads_alive": 0, "threads_total": 1, "total_attempts": 22, "total_success": 20, "total_failure": 2, "success_rate": 90.9, "target_count": 20, "target_reached": True, "last_error": "", "proxy": None, "proxy_pool_enabled": False, "mail_provider": "mailtm", "interval_seconds": 5, "run_count": 22, "success_count": 20, "failure_count": 2, "is_running": False, "last_started_at": None, "last_finished_at": None, "last_duration_seconds": None, "next_run_at": None, "failure_by_stage": {"add_phone_gate": 2}, "failure_signals": {"add_phone_gate": 2}, "recent_failure_hotspots": [{"key": "add_phone_gate", "stage": "add_phone_gate", "count": 2}], "recent_attempts": [ { "timestamp": "2026-03-27T10:00:00", "success": False, "stage": "add_phone_gate", "signal": "add_phone_gate", "error_message": "phone gate", "email_domain": "demo.example.test", "post_create_gate": "add_phone", "create_account_error_code": "", "proxy_key": "", "email": "", } ], "active_domain_recent_attempts": [], "active_domain_failure_by_stage": {"add_phone_gate": 2}, "active_domain_failure_signals": {"add_phone_gate": 2}, "active_domain_recent_failure_hotspots": [{"key": "add_phone_gate", "stage": "add_phone_gate", "count": 2}], "cfmail_rotation": None, "cfmail_add_phone_stoploss": {"active_domain": "demo.example.test"} } monkeypatch.setattr("main.RegistrationLoop", FakeLoop) scheduler = RegistrationBurstScheduler(settings) original_absorb = scheduler._absorb_batch_snapshot def absorb_and_stop(snapshot: dict[str, object], *, duration_seconds: float) -> None: original_absorb(snapshot, duration_seconds=duration_seconds) scheduler.stop() monkeypatch.setattr(scheduler, "_absorb_batch_snapshot", absorb_and_stop) scheduler.run() payload = json.loads((tmp_path / "runtime_state.json").read_text(encoding="utf-8")) register_snapshot = payload["register_snapshot"] assert observed == {"threads": 1, "target": 20} assert register_snapshot["status"] == "stopped" assert register_snapshot["scheduler_mode"] == "burst" assert register_snapshot["run_count"] == 1 assert register_snapshot["total_success"] == 20 assert register_snapshot["batch_target_count"] == 20