chatgpt-register/ops/responses_survival.py

"""Fixed cohort survival tracking using the codex responses API."""

from __future__ import annotations

import argparse
from datetime import datetime
import json
import time
from pathlib import Path
from typing import Any

from core.proxy_pool import ProxyLease, ProxyPool
from core.settings import AppSettings
from ops.common import get_management_key
from platforms.chatgpt.fingerprint import OPENAI_FINGERPRINT_PROFILE
from platforms.chatgpt.constants import OPENAI_USER_AGENT
from platforms.chatgpt.pool import load_token_record, update_token_record

from .scan import (
    ScanResult,
    _extract_credentials,
    _load_token_payload,
    _probe_responses_path,
)


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def _parse_iso(value: str) -> datetime | None:
    raw = str(value or "").strip()
    if not raw:
        return None
    try:
        return datetime.fromisoformat(raw)
    except Exception:
        return None


def _duration_seconds(started_at: str, ended_at: str) -> int | None:
    start_dt = _parse_iso(started_at)
    end_dt = _parse_iso(ended_at)
    if start_dt is None or end_dt is None:
        return None
    return max(0, int((end_dt - start_dt).total_seconds()))


def _member_age_seconds(member: dict[str, Any], probed_at: str) -> int | None:
    return _duration_seconds(str(member.get("created_at") or ""), probed_at)


def _compact_text(value: str, limit: int = 320) -> str:
    return " ".join(str(value or "").split())[:limit]


def _extract_error_facts(detail: str) -> tuple[str, str]:
    raw = str(detail or "").strip()
    if not raw:
        return "", ""
    try:
        payload = json.loads(raw)
    except Exception:
        return "", raw[:160]
    error = payload.get("error")
    if not isinstance(error, dict):
        return "", raw[:160]
    return str(error.get("code") or "").strip(), str(error.get("message") or "").strip()[:160]


def _state_template(
    *,
    pool_dir: Path,
    cohort_size: int,
    proxy: str | None,
    timeout_seconds: int,
    seed_source: str = "latest_generated_pool_files",
) -> dict[str, Any]:
    return {
        "probe_mode": "responses",
        "probe_target": "codex_responses",
        "updated_at": "",
        "seeded_at": "",
        "seed_source": seed_source,
        "pool_dir": str(pool_dir),
        "cohort_size": max(1, int(cohort_size)),
        "proxy": str(proxy or "").strip() or None,
        "probe_fingerprint_profile": OPENAI_FINGERPRINT_PROFILE,
        "probe_user_agent": OPENAI_USER_AGENT,
        "timeout_seconds": max(5, int(timeout_seconds)),
        "members": [],
        "summary": {
            "tracked": 0,
            "alive": 0,
            "invalid": 0,
            "missing": 0,
            "removed_after_invalid": 0,
            "transport_error": 0,
            "suspicious": 0,
            "never_probed": 0,
            "first_invalid_count": 0,
        },
        "promotion_stats": {
            "promoted_success_total": 0,
            "promoted_failure_total": 0,
            "last_promoted_at": "",
        },
        "changes": [],
        "round_count": 0,
    }


def load_responses_survival_state(path: Path) -> dict[str, Any]:
    if not path.is_file():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return payload if isinstance(payload, dict) else {}


def _persist_state(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = path.with_name(f"{path.name}.tmp")
    tmp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp_path.replace(path)


def _seed_member(path: Path) -> dict[str, Any] | None:
    try:
        payload = load_token_record(path)
    except Exception:
        return None
    email = str(payload.get("email") or "").strip()
    access_token = str(payload.get("access_token") or "").strip()
    account_id = str(payload.get("account_id") or "").strip()
    if not email or not access_token or not account_id:
        return None
    created_at = str(payload.get("created_at") or "").strip()
    if not created_at:
        created_at = datetime.fromtimestamp(path.stat().st_mtime).astimezone().isoformat(timespec="seconds")
    return {
        "email": email,
        "file_name": path.name,
        "path": str(path),
        "created_at": created_at,
        "selected_at": now_iso(),
        "first_probe_at": "",
        "last_probe_at": "",
        "probe_count": 0,
        "last_probe_status_code": None,
        "last_probe_category": "",
        "last_probe_detail": "",
        "transport_error_count": 0,
        "suspicious_count": 0,
        "missing_at": "",
        "removed_after_invalid_at": "",
        "last_missing_detail": "",
        "first_invalid_at": "",
        "first_invalid_error_code": "",
        "first_invalid_error_message": "",
        "first_use_at": "",
        "first_use_age_seconds": None,
        "first_use_fingerprint_profile": "",
        "fingerprint_consistent": None,
        "registration_fingerprint_profile": str(payload.get("registration_fingerprint_profile") or "").strip(),
        "registration_proxy_key": str(payload.get("registration_proxy_key") or "").strip(),
        "registration_proxy_region": str(payload.get("registration_proxy_region") or "").strip(),
        "registration_post_create_gate": str(payload.get("registration_post_create_gate") or "").strip(),
        "warmup_required": bool(payload.get("warmup_required")) or str(payload.get("registration_post_create_gate") or "").strip().lower() == "add_phone",
        "warmup_state": str(payload.get("warmup_state") or "").strip()
        or ("pending" if str(payload.get("registration_post_create_gate") or "").strip().lower() == "add_phone" else "not_required"),
        "warmup_passed": bool(payload.get("warmup_passed")) or str(payload.get("warmup_state") or "").strip() == "passed",
        "warmup_completed_at": str(payload.get("warmup_completed_at") or "").strip(),
        "successful_probe_count": int(payload.get("successful_probe_count") or 0),
        "warmup_promotion_recorded": bool(payload.get("warmup_promotion_recorded")),
        "warmup_promotion_result": str(payload.get("warmup_promotion_result") or "").strip(),
        "first_use_proxy_key": str(payload.get("first_use_proxy_key") or "").strip(),
        "first_use_proxy_region": str(payload.get("first_use_proxy_region") or "").strip(),
        "first_invalid_proxy_key": str(payload.get("first_invalid_proxy_key") or "").strip(),
        "first_invalid_proxy_region": str(payload.get("first_invalid_proxy_region") or "").strip(),
        "survival_seconds": None,
        "state": "tracking",
    }


def _collect_seed_members(
    pool_dir: Path,
    *,
    require_provenance: bool = False,
    recent_window_seconds: int = 0,
) -> list[dict[str, Any]]:
    recent_with_provenance: list[tuple[float, dict[str, Any]]] = []
    recent_without_provenance: list[tuple[float, dict[str, Any]]] = []
    cutoff_seconds = max(0, int(recent_window_seconds or 0))
    cutoff_ts = time.time() - cutoff_seconds if cutoff_seconds > 0 else 0.0
    for path in pool_dir.glob("*.json"):
        if not path.is_file():
            continue
        member = _seed_member(path)
        if member is None:
            continue
        created_at = _parse_iso(str(member.get("created_at") or ""))
        sort_ts = created_at.timestamp() if created_at is not None else path.stat().st_mtime
        if cutoff_ts and sort_ts < cutoff_ts:
            continue
        has_provenance = bool(str(member.get("registration_fingerprint_profile") or "").strip())
        if has_provenance:
            recent_with_provenance.append((sort_ts, member))
        elif not require_provenance:
            recent_without_provenance.append((sort_ts, member))
    recent_with_provenance.sort(key=lambda item: item[0], reverse=True)
    recent_without_provenance.sort(key=lambda item: item[0], reverse=True)
    ordered = [member for _ts, member in recent_with_provenance]
    ordered.extend(member for _ts, member in recent_without_provenance)
    return ordered


def _seed_members(
    pool_dir: Path,
    cohort_size: int,
    *,
    require_provenance: bool = False,
    recent_window_seconds: int = 0,
) -> list[dict[str, Any]]:
    ordered = _collect_seed_members(
        pool_dir,
        require_provenance=require_provenance,
        recent_window_seconds=recent_window_seconds,
    )
    return ordered[: max(1, int(cohort_size))]


def _member_identity(member: dict[str, Any]) -> str:
    return str(member.get("path") or member.get("file_name") or member.get("email") or "").strip()


def _member_created_ts(member: dict[str, Any]) -> float:
    created = _parse_iso(str(member.get("created_at") or "").strip())
    if created is not None:
        return created.timestamp()
    path = Path(str(member.get("path") or "")).expanduser()
    if path.is_file():
        return path.stat().st_mtime
    return 0.0


def _is_pending_warmup_member(member: dict[str, Any]) -> bool:
    return bool(member.get("warmup_required")) and not bool(member.get("warmup_passed")) and not _member_has_invalid_history(member)


def _merge_member_state(existing: dict[str, Any], fresh: dict[str, Any]) -> dict[str, Any]:
    merged = dict(fresh)
    merged.update(existing)
    merged["selected_at"] = str(existing.get("selected_at") or fresh.get("selected_at") or now_iso())
    return merged


def _member_priority(member: dict[str, Any]) -> tuple[float, float]:
    created_ts = _member_created_ts(member)
    if _is_pending_warmup_member(member):
        return (0.0, created_ts)
    last_probe = _parse_iso(str(member.get("last_probe_at") or "").strip())
    last_probe_ts = last_probe.timestamp() if last_probe is not None else 0.0
    if _member_has_invalid_history(member):
        return (1.0, -last_probe_ts)
    if str(member.get("last_probe_at") or "").strip():
        return (2.0, -last_probe_ts)
    return (3.0, -created_ts)


def _refresh_active_members(
    pool_dir: Path,
    members: list[dict[str, Any]],
    *,
    cohort_size: int,
    require_provenance: bool = False,
    recent_window_seconds: int = 0,
) -> list[dict[str, Any]]:
    candidates = _collect_seed_members(
        pool_dir,
        require_provenance=require_provenance,
        recent_window_seconds=recent_window_seconds,
    )
    candidate_by_key = {_member_identity(member): member for member in candidates if _member_identity(member)}
    existing_keys: set[str] = set()
    combined: list[dict[str, Any]] = []

    for raw_member in members:
        if not isinstance(raw_member, dict):
            continue
        key = _member_identity(raw_member)
        if key:
            existing_keys.add(key)
        fresh = candidate_by_key.get(key)
        combined.append(_merge_member_state(raw_member, fresh) if isinstance(fresh, dict) else raw_member)

    for candidate in candidates:
        key = _member_identity(candidate)
        if not key or key in existing_keys:
            continue
        combined.append(candidate)

    deduped: list[dict[str, Any]] = []
    seen: set[str] = set()
    for member in sorted(combined, key=_member_priority):
        key = _member_identity(member)
        if key and key in seen:
            continue
        if key:
            seen.add(key)
        deduped.append(member)
        if len(deduped) >= max(1, int(cohort_size)):
            break
    return deduped


def probe_responses_token_file(path: Path, proxy: str | None, timeout_seconds: int) -> ScanResult:
    payload, load_error = _load_token_payload(path)
    if load_error is not None:
        return load_error
    assert payload is not None
    credentials = _extract_credentials(path, payload)
    if isinstance(credentials, ScanResult):
        return credentials
    access_token, account_id = credentials
    return _probe_responses_path(path, access_token, account_id, proxy, timeout_seconds)


def _member_outcome(member: dict[str, Any]) -> str:
    state = str(member.get("state") or "").strip()
    if state == "invalid_removed":
        return "invalid_removed"
    category = str(member.get("last_probe_category") or "").strip()
    return category or "never_probed"


def _member_has_invalid_history(member: dict[str, Any]) -> bool:
    state = str(member.get("state") or "").strip()
    category = str(member.get("last_probe_category") or "").strip()
    return bool(str(member.get("first_invalid_at") or "").strip()) or state in {"invalid", "invalid_removed"} or category == "invalid"


def _preserve_terminal_invalid(member: dict[str, Any]) -> None:
    if str(member.get("last_probe_category") or "").strip() != "invalid":
        member["last_probe_category"] = "invalid"
    if member.get("last_probe_status_code") in {None, ""}:
        member["last_probe_status_code"] = 401
    detail = str(member.get("last_probe_detail") or "").strip()
    if not detail or detail.startswith("missing_file:"):
        member["last_probe_detail"] = "invalid_before_pool_removal"


def _persist_member_fields(member: dict[str, Any]) -> None:
    path = Path(str(member.get("path") or "")).expanduser()
    if not path.is_file():
        return
    try:
        update_token_record(
            path,
            warmup_required=bool(member.get("warmup_required")),
            warmup_state=str(member.get("warmup_state") or "").strip(),
            warmup_passed=bool(member.get("warmup_passed")),
            warmup_completed_at=str(member.get("warmup_completed_at") or "").strip(),
            successful_probe_count=int(member.get("successful_probe_count") or 0),
            warmup_promotion_recorded=bool(member.get("warmup_promotion_recorded")),
            warmup_promotion_result=str(member.get("warmup_promotion_result") or "").strip(),
            first_use_at=str(member.get("first_use_at") or "").strip(),
            first_use_age_seconds=member.get("first_use_age_seconds"),
            first_use_proxy_key=str(member.get("first_use_proxy_key") or "").strip(),
            first_use_proxy_region=str(member.get("first_use_proxy_region") or "").strip(),
            first_invalid_at=str(member.get("first_invalid_at") or "").strip(),
            first_invalid_error_code=str(member.get("first_invalid_error_code") or "").strip(),
            first_invalid_error_message=str(member.get("first_invalid_error_message") or "").strip(),
            first_invalid_proxy_key=str(member.get("first_invalid_proxy_key") or "").strip(),
            first_invalid_proxy_region=str(member.get("first_invalid_proxy_region") or "").strip(),
            survival_seconds=member.get("survival_seconds"),
        )
    except Exception:
        return


def _maybe_promote_warmup_member(
    member: dict[str, Any],
    *,
    settings: AppSettings | None,
    probed_at: str,
) -> str | None:
    if settings is None or str(settings.backend or "").strip().lower() != "cpa":
        return None
    if not bool(member.get("warmup_required")) or not bool(member.get("warmup_passed")):
        return None
    path = Path(str(member.get("path") or "")).expanduser()
    if not path.is_file():
        return None
    try:
        payload = load_token_record(path)
    except Exception:
        return None
    if not isinstance(payload, dict):
        return None
    if str(payload.get("cpa_sync_status") or "").strip().lower() == "synced":
        return None
    key = str(get_management_key() or "").strip()
    if not key:
        update_token_record(
            path,
            cpa_sync_status="failed",
            last_cpa_sync_at=probed_at,
            last_cpa_sync_error="CPA management key unavailable",
        )
        member["cpa_sync_status"] = "failed"
        return "failed"
    api_url = str(settings.cpa_management_base_url or "").strip()
    if api_url.endswith("/v0/management"):
        api_url = api_url[: -len("/v0/management")]
    from platforms.chatgpt.cpa_upload import upload_to_cpa

    ok, message = upload_to_cpa(payload, api_url=api_url.rstrip("/"), api_key=key, proxy=None)
    if ok:
        update_token_record(
            path,
            health_status="good",
            cpa_sync_status="synced",
            last_cpa_sync_at=probed_at,
            last_cpa_sync_error="",
        )
        member["cpa_sync_status"] = "synced"
        return "success"
    else:
        update_token_record(
            path,
            cpa_sync_status="failed",
            last_cpa_sync_at=probed_at,
            last_cpa_sync_error=message,
        )
        member["cpa_sync_status"] = "failed"
        return "failed"


def _update_member(
    member: dict[str, Any],
    result: ScanResult,
    probed_at: str,
    *,
    probe_proxy_key: str = "",
    probe_proxy_region: str = "",
    warmup_min_age_seconds: int = 600,
    warmup_min_successful_probes: int = 2,
) -> dict[str, Any]:
    previous_outcome = _member_outcome(member)
    member["last_probe_at"] = probed_at
    if not str(member.get("first_probe_at") or "").strip():
        member["first_probe_at"] = probed_at
    if not str(member.get("first_use_at") or "").strip():
        member["first_use_at"] = probed_at
        member["first_use_age_seconds"] = _duration_seconds(
            str(member.get("created_at") or "").strip(),
            probed_at,
        )
        member["first_use_fingerprint_profile"] = OPENAI_FINGERPRINT_PROFILE
        member["first_use_proxy_key"] = str(probe_proxy_key or "").strip()
        member["first_use_proxy_region"] = str(probe_proxy_region or "").strip()
        registration_profile = str(member.get("registration_fingerprint_profile") or "").strip()
        if registration_profile:
            member["fingerprint_consistent"] = registration_profile == OPENAI_FINGERPRINT_PROFILE
    member["probe_count"] = int(member.get("probe_count") or 0) + 1

    if result.category == "missing" and _member_has_invalid_history(member):
        if not str(member.get("missing_at") or "").strip():
            member["missing_at"] = probed_at
        if not str(member.get("removed_after_invalid_at") or "").strip():
            member["removed_after_invalid_at"] = probed_at
        member["last_missing_detail"] = _compact_text(result.detail or "")
        _preserve_terminal_invalid(member)
        member["state"] = "invalid_removed"
        next_outcome = _member_outcome(member)
        detail = _compact_text(
            f"removed_after_invalid | {member.get('last_missing_detail') or ''}"
        )
        return {
            "email": str(member.get("email") or "").strip(),
            "from": previous_outcome,
            "to": next_outcome,
            "probed_at": probed_at,
            "survival_seconds": member.get("survival_seconds"),
            "detail": detail,
        }

    if result.category != "invalid" and _member_has_invalid_history(member):
        member["post_invalid_probe_at"] = probed_at
        member["post_invalid_probe_category"] = result.category
        member["post_invalid_probe_detail"] = _compact_text(result.detail or "")
        _preserve_terminal_invalid(member)
        member["state"] = "invalid"
        next_outcome = _member_outcome(member)
        return {
            "email": str(member.get("email") or "").strip(),
            "from": previous_outcome,
            "to": next_outcome,
            "probed_at": probed_at,
            "survival_seconds": member.get("survival_seconds"),
            "detail": member["post_invalid_probe_detail"],
        }

    member["last_probe_status_code"] = result.status_code
    member["last_probe_category"] = result.category
    member["last_probe_detail"] = _compact_text(result.detail or "")

    if result.category == "transport_error":
        member["transport_error_count"] = int(member.get("transport_error_count") or 0) + 1
    elif result.category not in {"normal", "invalid", "missing"}:
        member["suspicious_count"] = int(member.get("suspicious_count") or 0) + 1
    elif result.category == "missing" and not str(member.get("missing_at") or "").strip():
        member["missing_at"] = probed_at

    if result.category == "invalid":
        if not str(member.get("first_invalid_at") or "").strip():
            member["first_invalid_at"] = probed_at
            member["survival_seconds"] = _duration_seconds(
                str(member.get("created_at") or "").strip() or str(member.get("first_probe_at") or "").strip(),
                probed_at,
            )
            error_code, error_message = _extract_error_facts(result.detail or "")
            member["first_invalid_error_code"] = error_code
            member["first_invalid_error_message"] = error_message
            member["first_invalid_proxy_key"] = str(probe_proxy_key or "").strip()
            member["first_invalid_proxy_region"] = str(probe_proxy_region or "").strip()
        if bool(member.get("warmup_required")):
            member["warmup_state"] = "failed"
            member["warmup_passed"] = False
        member["state"] = "invalid"
    elif result.category == "missing":
        member["state"] = "missing"
    else:
        if result.category == "normal":
            member["successful_probe_count"] = int(member.get("successful_probe_count") or 0) + 1
            if bool(member.get("warmup_required")) and not bool(member.get("warmup_passed")):
                current_age = _member_age_seconds(member, probed_at)
                if (
                    int(member.get("successful_probe_count") or 0) >= max(1, int(warmup_min_successful_probes))
                    and (current_age is not None and int(current_age) >= max(0, int(warmup_min_age_seconds)))
                ):
                    member["warmup_passed"] = True
                    member["warmup_state"] = "passed"
                    member["warmup_completed_at"] = probed_at
        member["state"] = "tracking"
    _persist_member_fields(member)

    next_outcome = _member_outcome(member)
    return {
        "email": str(member.get("email") or "").strip(),
        "from": previous_outcome,
        "to": next_outcome,
        "probed_at": probed_at,
        "survival_seconds": member.get("survival_seconds"),
        "detail": member["last_probe_detail"],
    }


def _probe_member_proxy(
    member: dict[str, Any],
    *,
    default_proxy: str | None,
    proxy_pool: ProxyPool | None,
) -> tuple[str | None, str, str, ProxyLease | None]:
    preferred_name = str(member.get("registration_proxy_key") or "").strip()
    preferred_region = str(member.get("registration_proxy_region") or "").strip().lower()
    if proxy_pool is None:
        return default_proxy, preferred_name, preferred_region, None
    try:
        lease = proxy_pool.acquire(
            timeout=5.0,
            preferred_name=preferred_name or None,
            preferred_regions=(preferred_region,) if preferred_region else (),
        )
    except Exception:
        return default_proxy, preferred_name, preferred_region, None
    return lease.proxy_url, lease.name, preferred_region or "", lease


def _build_summary(members: list[dict[str, Any]]) -> dict[str, int]:
    summary = {
        "tracked": len(members),
        "alive": 0,
        "invalid": 0,
        "missing": 0,
        "removed_after_invalid": 0,
        "transport_error": 0,
        "suspicious": 0,
        "never_probed": 0,
        "first_invalid_count": 0,
    }
    for member in members:
        outcome = _member_outcome(member)
        if outcome == "never_probed":
            summary["never_probed"] += 1
        elif outcome == "normal":
            summary["alive"] += 1
        elif outcome == "invalid":
            summary["invalid"] += 1
        elif outcome == "invalid_removed":
            summary["invalid"] += 1
            summary["removed_after_invalid"] += 1
        elif outcome == "missing":
            summary["missing"] += 1
        elif outcome == "transport_error":
            summary["transport_error"] += 1
        else:
            summary["suspicious"] += 1
        if str(member.get("first_invalid_at") or "").strip():
            summary["first_invalid_count"] += 1
    return summary


def responses_survival_once(
    *,
    pool_dir: Path,
    state_file: Path,
    cohort_size: int,
    proxy: str | None,
    timeout_seconds: int,
    reseed: bool = False,
    proxy_pool: ProxyPool | None = None,
    require_provenance: bool = False,
    recent_window_seconds: int = 0,
    warmup_min_age_seconds: int = 600,
    warmup_min_successful_probes: int = 2,
    settings: AppSettings | None = None,
) -> dict[str, Any]:
    state = load_responses_survival_state(state_file)
    seeded = False
    reseeded = False

    if not state or reseed:
        state = _state_template(
            pool_dir=pool_dir,
            cohort_size=cohort_size,
            proxy=proxy,
            timeout_seconds=timeout_seconds,
        )
        state["members"] = _seed_members(
            pool_dir,
            int(state.get("cohort_size") or cohort_size),
            require_provenance=require_provenance,
            recent_window_seconds=recent_window_seconds,
        )
        state["seeded_at"] = now_iso()
        seeded = True
        reseeded = reseed
    else:
        state.setdefault("probe_mode", "responses")
        state.setdefault("probe_target", "codex_responses")
        state.setdefault("pool_dir", str(pool_dir))
        state.setdefault("cohort_size", max(1, int(cohort_size)))
        state.setdefault("proxy", str(proxy or "").strip() or None)
        state.setdefault("probe_fingerprint_profile", OPENAI_FINGERPRINT_PROFILE)
        state.setdefault("probe_user_agent", OPENAI_USER_AGENT)
        state.setdefault("timeout_seconds", max(5, int(timeout_seconds)))
        state.setdefault("members", [])
        state.setdefault("summary", {})
        state.setdefault("changes", [])
        state.setdefault(
            "promotion_stats",
            {
                "promoted_success_total": 0,
                "promoted_failure_total": 0,
                "last_promoted_at": "",
            },
        )
        state.setdefault("seed_source", "latest_generated_pool_files")
        state.setdefault("round_count", 0)

    if not isinstance(state.get("members"), list):
        state["members"] = []
    if not isinstance(state.get("promotion_stats"), dict):
        state["promotion_stats"] = {
            "promoted_success_total": 0,
            "promoted_failure_total": 0,
            "last_promoted_at": "",
        }
    if not state["members"]:
        state["members"] = _seed_members(
            pool_dir,
            int(state.get("cohort_size") or cohort_size),
            require_provenance=require_provenance,
            recent_window_seconds=recent_window_seconds,
        )
        state["seeded_at"] = now_iso()
        seeded = True
    else:
        state["members"] = _refresh_active_members(
            pool_dir,
            [member for member in state["members"] if isinstance(member, dict)],
            cohort_size=int(state.get("cohort_size") or cohort_size),
            require_provenance=require_provenance,
            recent_window_seconds=recent_window_seconds,
        )

    changes: list[dict[str, Any]] = []
    for raw_member in state["members"]:
        if not isinstance(raw_member, dict):
            continue
        member = raw_member
        probed_at = now_iso()
        selected_proxy, probe_proxy_key, probe_proxy_region, lease = _probe_member_proxy(
            member,
            default_proxy=str(state.get("proxy") or "").strip() or None,
            proxy_pool=proxy_pool,
        )
        try:
            result = probe_responses_token_file(
                Path(str(member.get("path") or "")),
                selected_proxy,
                max(5, int(state.get("timeout_seconds") or timeout_seconds)),
            )
        finally:
            if lease is not None and proxy_pool is not None:
                try:
                    proxy_pool.release(lease, success=result.category == "normal" if 'result' in locals() else None, stage=result.category if 'result' in locals() else None)
                except Exception:
                    pass
        change = _update_member(
            member,
            result,
            probed_at,
            probe_proxy_key=probe_proxy_key,
            probe_proxy_region=probe_proxy_region,
            warmup_min_age_seconds=warmup_min_age_seconds,
            warmup_min_successful_probes=warmup_min_successful_probes,
        )
        promotion_outcome = _maybe_promote_warmup_member(
            member,
            settings=settings,
            probed_at=probed_at,
        )
        if promotion_outcome in {"success", "failed"} and not bool(member.get("warmup_promotion_recorded")):
            stats = state["promotion_stats"]
            member["warmup_promotion_recorded"] = True
            member["warmup_promotion_result"] = promotion_outcome
            if promotion_outcome == "success":
                stats["promoted_success_total"] = int(stats.get("promoted_success_total") or 0) + 1
            else:
                stats["promoted_failure_total"] = int(stats.get("promoted_failure_total") or 0) + 1
            stats["last_promoted_at"] = probed_at
            _persist_member_fields(member)
        if change["from"] != change["to"]:
            changes.append(change)

    state["updated_at"] = now_iso()
    state["summary"] = _build_summary([member for member in state["members"] if isinstance(member, dict)])
    state["changes"] = changes
    state["seeded"] = seeded
    state["reseeded"] = reseeded
    state["state_file"] = str(state_file)
    state["round_count"] = int(state.get("round_count") or 0) + 1
    _persist_state(state_file, state)
    return state


def print_responses_survival_summary(result: dict[str, Any]) -> None:
    summary = result.get("summary") if isinstance(result.get("summary"), dict) else {}
    print(
        "[responses-survival] summary | "
        f"tracked={int(summary.get('tracked') or 0)} | alive={int(summary.get('alive') or 0)} "
        f"| invalid={int(summary.get('invalid') or 0)} | missing={int(summary.get('missing') or 0)} "
        f"| removed_after_invalid={int(summary.get('removed_after_invalid') or 0)} "
        f"| transport_error={int(summary.get('transport_error') or 0)} "
        f"| suspicious={int(summary.get('suspicious') or 0)}"
    )
    for change in result.get("changes") or []:
        if not isinstance(change, dict):
            continue
        survival_seconds = change.get("survival_seconds")
        survival_text = f" | survival={survival_seconds}s" if survival_seconds is not None else ""
        print(
            f"[responses-survival] state change | {change.get('email') or '?'} | "
            f"{change.get('from') or 'never_probed'} -> {change.get('to') or '?'}{survival_text}"
        )
    print(f"[responses-survival] state={result.get('state_file') or '-'}")


def run_responses_survival_loop(
    *,
    pool_dir: Path,
    state_file: Path,
    cohort_size: int,
    proxy: str | None,
    timeout_seconds: int,
    interval_seconds: int,
    reseed: bool = False,
    max_rounds: int = 0,
    settings: AppSettings | None = None,
) -> dict[str, Any]:
    round_index = 0
    last_result: dict[str, Any] = {}
    proxy_pool = ProxyPool.from_settings(settings) if settings is not None else None
    if proxy_pool is not None:
        proxy_pool.start()
    while True:
        round_index += 1
        last_result = responses_survival_once(
            pool_dir=pool_dir,
            state_file=state_file,
            cohort_size=cohort_size,
            proxy=proxy,
            timeout_seconds=timeout_seconds,
            reseed=reseed and round_index == 1,
            proxy_pool=proxy_pool,
            require_provenance=bool(settings.responses_survival_require_provenance) if settings is not None else False,
            recent_window_seconds=int(settings.responses_survival_recent_window_seconds) if settings is not None else 0,
            warmup_min_age_seconds=int(settings.warmup_min_age_seconds) if settings is not None else 600,
            warmup_min_successful_probes=int(settings.warmup_min_successful_probes) if settings is not None else 2,
            settings=settings,
        )
        print_responses_survival_summary(last_result)
        if max_rounds > 0 and round_index >= max_rounds:
            if proxy_pool is not None:
                proxy_pool.close()
            return last_result
        time.sleep(max(5, int(interval_seconds)))


def build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run responses survival tracking for recently created accounts.")
    parser.add_argument("--pool-dir", required=True)
    parser.add_argument("--state-file", required=True)
    parser.add_argument("--cohort-size", type=int, default=8)
    parser.add_argument("--proxy", default="")
    parser.add_argument("--timeout-seconds", type=int, default=30)
    parser.add_argument("--interval-seconds", type=int, default=60)
    parser.add_argument("--max-rounds", type=int, default=0, help="0 means run forever")
    parser.add_argument("--reseed", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_arg_parser()
    args = parser.parse_args(argv)
    run_responses_survival_loop(
        pool_dir=Path(args.pool_dir).expanduser().resolve(),
        state_file=Path(args.state_file).expanduser().resolve(),
        cohort_size=max(1, int(args.cohort_size)),
        proxy=str(args.proxy or "").strip() or None,
        timeout_seconds=max(5, int(args.timeout_seconds)),
        interval_seconds=max(5, int(args.interval_seconds)),
        reseed=bool(args.reseed),
        max_rounds=max(0, int(args.max_rounds)),
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())