"""StockPlatform public API runtime readback.

This readback probes only public HTTPS endpoints. It does not SSH, use Docker,
read secrets, write StockPlatform data, or trigger recovery actions.
"""

from __future__ import annotations

import json
import urllib.error
import urllib.request
from collections.abc import Callable
from pathlib import Path
from typing import Any

from src.services.reboot_auto_recovery_slo_scorecard import (
    load_latest_reboot_auto_recovery_slo_scorecard,
)
from src.services.snapshot_paths import default_operations_dir

_API_SCHEMA_VERSION = "stockplatform_public_api_runtime_readback_v1"
_DATA_READINESS_SCHEMA_VERSION = (
    "stockplatform_postgres_readiness_or_data_source_readback_v1"
)
_DATA_READINESS_RECEIPT_KEY = (
    "stockplatform_postgres_readiness_or_data_source_readback"
)
_SOURCE_CONTRACT_SCHEMA_VERSION = "stockplatform_source_contract_control_readback_v1"
_SOURCE_CONTRACT_RECEIPT_KEY = "stockplatform_source_contract_readback"
_AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED = (
    "ai_recommendations_stale_after_core_recovered"
)
_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
_RECOVERY_RECEIPT_FILE = (
    "stockplatform-public-api-runtime-recovery-control-receipt.snapshot.json"
)
_DEFAULT_BASE_URL = "https://stock.wooo.work"
_DEFAULT_TIMEOUT_SECONDS = 4.0
_DEFAULT_DATA_ENDPOINT_TIMEOUT_SECONDS = 10.0
_DEFAULT_PROBE_ATTEMPTS = 2
_MAX_READBACK_BODY_BYTES = 500_000

Probe = Callable[[str, float], dict[str, Any]]


def load_latest_stockplatform_public_api_runtime_readback(
    *,
    base_url: str = _DEFAULT_BASE_URL,
    timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
    data_endpoint_timeout_seconds: float = _DEFAULT_DATA_ENDPOINT_TIMEOUT_SECONDS,
    probe_attempts: int = _DEFAULT_PROBE_ATTEMPTS,
    operations_dir: Path | None = None,
    probe: Probe | None = None,
) -> dict[str, Any]:
    """Build a live public readback for StockPlatform web/API health."""
    http_probe = probe or _http_probe
    normalized_base_url = base_url.rstrip("/")
    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
    committed_scorecard = load_latest_reboot_auto_recovery_slo_scorecard(
        directory
    )
    recovery_control_receipt = _load_recovery_control_receipt(directory)
    committed_stock = _dict(committed_scorecard.get("stockplatform_data_freshness"))
    endpoints = {
        "public_web_healthz": {
            "path": "/healthz",
            "parse_json": False,
            "timeout_seconds": timeout_seconds,
        },
        "public_api_healthz": {
            "path": "/api/healthz",
            "parse_json": False,
            "timeout_seconds": timeout_seconds,
        },
        "freshness": {
            "path": "/api/v1/system/freshness",
            "parse_json": True,
            "timeout_seconds": data_endpoint_timeout_seconds,
        },
        "ingestion": {
            "path": "/api/v1/system/ingestion",
            "parse_json": True,
            "timeout_seconds": data_endpoint_timeout_seconds,
        },
    }
    probes = {
        name: _probe_endpoint(
            http_probe,
            f"{normalized_base_url}{endpoint['path']}",
            float(endpoint["timeout_seconds"]),
            parse_json=bool(endpoint["parse_json"]),
            attempts=probe_attempts,
        )
        for name, endpoint in endpoints.items()
    }
    return _build_payload(
        base_url=normalized_base_url,
        timeout_seconds=timeout_seconds,
        data_endpoint_timeout_seconds=data_endpoint_timeout_seconds,
        probe_attempts=probe_attempts,
        probes=probes,
        committed_stockplatform=committed_stock,
        recovery_control_receipt=recovery_control_receipt,
    )


def _build_payload(
    *,
    base_url: str,
    timeout_seconds: float,
    data_endpoint_timeout_seconds: float,
    probe_attempts: int,
    probes: dict[str, dict[str, Any]],
    committed_stockplatform: dict[str, Any],
    recovery_control_receipt: dict[str, Any],
) -> dict[str, Any]:
    web = _dict(probes.get("public_web_healthz"))
    api = _dict(probes.get("public_api_healthz"))
    freshness = _dict(probes.get("freshness"))
    ingestion = _dict(probes.get("ingestion"))
    freshness_json = _dict(freshness.get("json"))
    ingestion_json = _dict(ingestion.get("json"))
    freshness_sources = _sources(freshness_json.get("sources"))

    checks = {
        "public_web_health_ok": web.get("http_status") == 200,
        "public_api_health_ok": api.get("http_status") == 200,
        "freshness_readback_ok": freshness.get("http_status") == 200
        and freshness_json.get("status") == "ok",
        "ingestion_readback_ok": ingestion.get("http_status") == 200
        and ingestion_json.get("status") == "ok",
    }
    active_blockers = _active_blockers(
        api=api,
        freshness=freshness,
        ingestion=ingestion,
        freshness_json=freshness_json,
        ingestion_json=ingestion_json,
    )
    ready = all(checks.values())
    data_dependency_classification = _data_dependency_classification(
        active_blockers,
        api_health_ok=checks["public_api_health_ok"],
    )
    postgres_not_ready = data_dependency_classification == "postgres_not_ready"
    data_readiness_control_readback = _data_readiness_control_readback(
        checks=checks,
        freshness=freshness,
        ingestion=ingestion,
        freshness_json=freshness_json,
        ingestion_json=ingestion_json,
        data_dependency_classification=data_dependency_classification,
        postgres_not_ready=postgres_not_ready,
        data_dependency_blockers=active_blockers,
    )
    source_contract_control_readback = _source_contract_control_readback(
        freshness=freshness,
        ingestion=ingestion,
        freshness_json=freshness_json,
        ingestion_json=ingestion_json,
        data_dependency_classification=data_dependency_classification,
        postgres_not_ready=postgres_not_ready,
        data_dependency_blockers=active_blockers,
    )
    recovery_control_path = _recovery_control_path_readback(
        runtime_ready=ready,
        receipt=recovery_control_receipt,
        data_readiness_control_readback=data_readiness_control_readback,
        source_contract_control_readback=source_contract_control_readback,
    )
    recovery_control_path_blockers = _strings(
        recovery_control_path.get("active_blockers")
    )
    active_blockers = _unique_strings(active_blockers + recovery_control_path_blockers)
    committed_freshness_ok = committed_stockplatform.get("freshness_status") == "ok"
    committed_ingestion_ok = committed_stockplatform.get("ingestion_status") == "ok"
    committed_ok = committed_freshness_ok and committed_ingestion_ok
    live_drift_from_committed_scorecard = committed_ok and not ready
    status = (
        "stockplatform_public_api_runtime_ready"
        if ready
        else "blocked_stockplatform_public_api_runtime_drift"
    )
    safe_next_step = (
        "stockplatform_public_api_ready_keep_p0_006_reboot_window_gate"
        if ready
        else _blocked_safe_next_step(
            data_dependency_classification=data_dependency_classification,
        )
    )
    http_statuses = {
        name: _dict(probe).get("http_status") for name, probe in probes.items()
    }
    return {
        "schema_version": _API_SCHEMA_VERSION,
        "scope": "stockplatform_public_api_runtime_readback",
        "priority": "P0-006",
        "status": status,
        "safe_next_step": safe_next_step,
        "active_blockers": active_blockers,
        "active_blocker_count": len(active_blockers),
        "runtime_ready": ready,
        "live_drift_from_committed_scorecard": live_drift_from_committed_scorecard,
        "base_url": base_url,
        "timeout_seconds": timeout_seconds,
        "data_endpoint_timeout_seconds": data_endpoint_timeout_seconds,
        "probe_attempts": probe_attempts,
        "checks": checks,
        "readback": {
            "web_health_http_status": web.get("http_status"),
            "api_health_http_status": api.get("http_status"),
            "freshness_http_status": freshness.get("http_status"),
            "ingestion_http_status": ingestion.get("http_status"),
            "freshness_status": str(freshness_json.get("status") or "unknown"),
            "ingestion_status": str(ingestion_json.get("status") or "unknown"),
            "freshness_latest_trading_date": str(
                freshness_json.get("latest_trading_date") or ""
            ),
            "ingestion_latest_trading_date": str(
                ingestion_json.get("latest_trading_date") or ""
            ),
            "freshness_blockers": _strings(freshness_json.get("blockers")),
            "ingestion_blockers": _strings(ingestion_json.get("blockers")),
            "freshness_source_count": len(freshness_sources),
            "freshness_sla_source_count": _sla_source_count(freshness_sources),
            "freshness_non_ok_source_count": _non_ok_source_count(
                freshness_sources
            ),
            "data_dependency_classification": data_dependency_classification,
            "ai_recommendations_stale_after_core_recovered": (
                data_dependency_classification
                == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
            ),
            "postgres_not_ready": postgres_not_ready,
            "committed_freshness_status": str(
                committed_stockplatform.get("freshness_status") or "unknown"
            ),
            "committed_ingestion_status": str(
                committed_stockplatform.get("ingestion_status") or "unknown"
            ),
            "live_drift_from_committed_scorecard": (
                live_drift_from_committed_scorecard
            ),
            "recovery_control_path_status": recovery_control_path.get("status"),
            "recovery_control_path_active_blockers": recovery_control_path_blockers,
            "data_readiness_control_readback_status": (
                data_readiness_control_readback.get("status")
            ),
            "data_readiness_control_readback_present": (
                data_readiness_control_readback.get("readback_present") is True
            ),
            "source_contract_control_readback_status": (
                source_contract_control_readback.get("status")
            ),
            "source_contract_control_readback_present": (
                source_contract_control_readback.get("readback_present") is True
            ),
            "source_contract_receipt_provided": (
                source_contract_control_readback.get("receipt_provided") is True
            ),
            "source_contract_non_ok_source_count": _int_or_none(
                source_contract_control_readback.get("non_ok_source_count")
            )
            or 0,
            "source_contract_blocker_count": len(
                _strings(source_contract_control_readback.get("active_blockers"))
            ),
        },
        "rollups": {
            "runtime_ready": ready,
            "public_web_health_ok": checks["public_web_health_ok"],
            "public_api_health_ok": checks["public_api_health_ok"],
            "freshness_readback_ok": checks["freshness_readback_ok"],
            "ingestion_readback_ok": checks["ingestion_readback_ok"],
            "http_502_count": sum(
                1 for status_code in http_statuses.values() if status_code == 502
            ),
            "http_non_200_count": sum(
                1
                for status_code in http_statuses.values()
                if isinstance(status_code, int) and status_code != 200
            ),
            "committed_stockplatform_freshness_ok": committed_freshness_ok,
            "committed_stockplatform_ingestion_ok": committed_ingestion_ok,
            "live_drift_from_committed_scorecard": (
                live_drift_from_committed_scorecard
            ),
            "recovery_control_path_ready": (
                recovery_control_path.get("status")
                == "not_required_stockplatform_runtime_ready"
            ),
            "recovery_control_path_blocked": bool(recovery_control_path_blockers),
            "recovery_control_path_blocker_count": len(
                recovery_control_path_blockers
            ),
            "data_dependency_classification": data_dependency_classification,
            "ai_recommendations_stale_after_core_recovered": (
                data_dependency_classification
                == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
            ),
            "postgres_not_ready": postgres_not_ready,
            "postgres_readiness_or_data_source_readback_present": (
                data_readiness_control_readback.get("readback_present") is True
            ),
            "postgres_readiness_or_data_source_receipt_provided": (
                data_readiness_control_readback.get("receipt_provided") is True
            ),
            "source_contract_readback_present": (
                source_contract_control_readback.get("readback_present") is True
            ),
            "source_contract_receipt_provided": (
                source_contract_control_readback.get("receipt_provided") is True
            ),
            "source_contract_ready": (
                source_contract_control_readback.get("source_contract_ready")
                is True
            ),
            "source_contract_non_ok_source_count": _int_or_none(
                source_contract_control_readback.get("non_ok_source_count")
            )
            or 0,
            "source_contract_blocker_count": len(
                _strings(source_contract_control_readback.get("active_blockers"))
            ),
        },
        "data_readiness_control_readback": data_readiness_control_readback,
        "source_contract_control_readback": source_contract_control_readback,
        "recovery_control_path": recovery_control_path,
        "probes": probes,
        "operation_boundaries": {
            "read_only_public_https_probe": True,
            "ssh_used": False,
            "docker_command_performed": False,
            "docker_restart_performed": False,
            "docker_daemon_restart_performed": False,
            "host_reboot_performed": False,
            "service_restart_performed": False,
            "database_write_or_restore_performed": False,
            "stockplatform_manual_data_write_performed": False,
            "secret_value_collection_allowed": False,
            "github_api_used": False,
            "workflow_trigger_performed": False,
            "runtime_write_allowed": False,
        },
    }


def _recovery_control_path_readback(
    *,
    runtime_ready: bool,
    receipt: dict[str, Any],
    data_readiness_control_readback: dict[str, Any],
    source_contract_control_readback: dict[str, Any],
) -> dict[str, Any]:
    if runtime_ready:
        return {
            "status": "not_required_stockplatform_runtime_ready",
            "active_blockers": [],
            "receipt": receipt,
            "provided_receipts": _provided_receipts_with_live_readback(
                _strings(receipt.get("provided_receipts")),
                data_readiness_control_readback,
                source_contract_control_readback,
            ),
            "missing_receipts": _missing_receipts_after_live_readback(
                _strings(receipt.get("missing_receipts")),
                data_readiness_control_readback,
                source_contract_control_readback,
            ),
            "safe_recovery_channels": [
                "keep_monitoring_public_api_runtime_readback",
            ],
        }
    if receipt:
        provided_receipts = _provided_receipts_with_live_readback(
            _strings(receipt.get("provided_receipts")),
            data_readiness_control_readback,
            source_contract_control_readback,
        )
        missing_receipts = _missing_receipts_after_live_readback(
            _strings(receipt.get("missing_receipts")),
            data_readiness_control_readback,
            source_contract_control_readback,
        )
        active_blockers = _recovery_control_active_blockers(
            receipt_blockers=_strings(receipt.get("active_blockers")),
            missing_receipts=missing_receipts,
            data_readiness_control_readback=data_readiness_control_readback,
        )
        return {
            "status": str(
                receipt.get("status")
                or "blocked_recovery_control_path_receipt_incomplete"
            ),
            "active_blockers": active_blockers,
            "receipt": receipt,
            "provided_receipts": provided_receipts,
            "missing_receipts": missing_receipts,
            "live_data_readiness_receipt_key": (
                _DATA_READINESS_RECEIPT_KEY
                if data_readiness_control_readback.get("receipt_provided") is True
                else ""
            ),
            "live_source_contract_receipt_key": (
                _SOURCE_CONTRACT_RECEIPT_KEY
                if source_contract_control_readback.get("receipt_provided") is True
                else ""
            ),
            "safe_recovery_channels": _strings(
                receipt.get("safe_recovery_channels")
            ),
            "forbidden_recovery_channels": _strings(
                receipt.get("forbidden_recovery_channels")
            ),
            "required_receipts": _strings(receipt.get("required_receipts")),
        }
    return {
        "status": "blocked_recovery_control_path_receipt_missing",
        "active_blockers": [
            "stockplatform_recovery_control_path_receipt_missing",
            "stockplatform_controlled_deploy_or_ssh_readback_required",
        ],
        "safe_recovery_channels": [
            "existing_stockplatform_manual_deploy_workflow_dispatch_with_runner_pressure_guard",
            "bounded_ssh_or_runner_recovery_readback_without_docker_daemon_restart",
        ],
        "forbidden_recovery_channels": [
            "docker_daemon_restart",
            "host_reboot",
            "docker_prune_or_volume_restore",
            "manual_database_rows_or_fake_freshness",
            "restore_push_deploy_trigger_during_110_capacity_freeze",
            "read_secret_or_raw_session_values",
        ],
        "required_receipts": [
            "target_selector",
            "source_of_truth_diff_or_runtime_state_diff",
            "check_mode_or_dry_run",
            "rollback_command",
            "post_apply_public_api_verifier",
            "km_playbook_log_writeback_receipt",
        ],
    }


def _data_readiness_control_readback(
    *,
    checks: dict[str, bool],
    freshness: dict[str, Any],
    ingestion: dict[str, Any],
    freshness_json: dict[str, Any],
    ingestion_json: dict[str, Any],
    data_dependency_classification: str,
    postgres_not_ready: bool,
    data_dependency_blockers: list[str],
) -> dict[str, Any]:
    freshness_http_status = freshness.get("http_status")
    ingestion_http_status = ingestion.get("http_status")
    readback_present = freshness_http_status == 200 and ingestion_http_status == 200
    data_ready = (
        readback_present
        and freshness_json.get("status") == "ok"
        and ingestion_json.get("status") == "ok"
    )
    if data_ready:
        status = "stockplatform_data_readiness_ready"
        safe_next_step = "run_post_apply_public_api_verifier_and_clear_p0_006_blocker"
    elif (
        data_dependency_classification
        == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
    ):
        status = "blocked_stockplatform_ai_recommendations_stale_after_core_recovered"
        safe_next_step = (
            "run_stockplatform_ai_recommendation_pipeline_readback_then_rerun_"
            "public_api_verifier"
        )
    elif readback_present and postgres_not_ready:
        status = "blocked_stockplatform_postgres_not_ready"
        safe_next_step = (
            "run_stockplatform_production_migration_path_after_control_channel_"
            "readback_no_manual_db_write"
        )
    elif readback_present:
        status = "blocked_stockplatform_data_dependency_not_ready"
        safe_next_step = (
            "diff_stockplatform_freshness_ingestion_source_contract_then_rerun_"
            "public_api_verifier"
        )
    else:
        status = "blocked_stockplatform_data_readiness_readback_unavailable"
        safe_next_step = (
            "restore_public_api_data_endpoints_or_control_channel_readback_then_"
            "rerun_data_readiness_readback"
        )
    missing_evidence = []
    if freshness_http_status != 200:
        missing_evidence.append("freshness_endpoint_http_200")
    if ingestion_http_status != 200:
        missing_evidence.append("ingestion_endpoint_http_200")
    freshness_sources = _sources(freshness_json.get("sources"))
    return {
        "schema_version": _DATA_READINESS_SCHEMA_VERSION,
        "receipt_key": _DATA_READINESS_RECEIPT_KEY,
        "status": status,
        "source": "stockplatform_public_api_freshness_ingestion_probe",
        "readback_present": readback_present,
        "receipt_provided": readback_present,
        "safe_next_step": safe_next_step,
        "data_ready": data_ready,
        "postgres_ready": data_ready and not postgres_not_ready,
        "postgres_not_ready": postgres_not_ready,
        "data_dependency_classification": data_dependency_classification,
        "ai_recommendations_stale_after_core_recovered": (
            data_dependency_classification
            == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
        ),
        "freshness_http_status": freshness_http_status,
        "ingestion_http_status": ingestion_http_status,
        "freshness_status": str(freshness_json.get("status") or "unknown"),
        "ingestion_status": str(ingestion_json.get("status") or "unknown"),
        "freshness_latest_trading_date": str(
            freshness_json.get("latest_trading_date") or ""
        ),
        "ingestion_latest_trading_date": str(
            ingestion_json.get("latest_trading_date") or ""
        ),
        "freshness_blockers": _strings(freshness_json.get("blockers")),
        "ingestion_blockers": _strings(ingestion_json.get("blockers")),
        "freshness_source_count": len(freshness_sources),
        "freshness_sla_source_count": _sla_source_count(freshness_sources),
        "freshness_non_ok_source_count": _non_ok_source_count(freshness_sources),
        "active_blockers": [
            blocker
            for blocker in data_dependency_blockers
            if blocker.startswith("stockplatform_freshness_")
            or blocker.startswith("stockplatform_ingestion_")
        ],
        "missing_evidence": missing_evidence,
        "operation_boundaries": {
            "public_https_probe_only": True,
            "ssh_used": False,
            "docker_command_performed": False,
            "database_write_or_restore_performed": False,
            "stockplatform_manual_data_write_performed": False,
            "secret_value_collection_allowed": False,
            "runtime_action_performed": False,
        },
    }


def _source_contract_control_readback(
    *,
    freshness: dict[str, Any],
    ingestion: dict[str, Any],
    freshness_json: dict[str, Any],
    ingestion_json: dict[str, Any],
    data_dependency_classification: str,
    postgres_not_ready: bool,
    data_dependency_blockers: list[str],
) -> dict[str, Any]:
    freshness_http_status = freshness.get("http_status")
    ingestion_http_status = ingestion.get("http_status")
    readback_present = freshness_http_status == 200 and ingestion_http_status == 200
    freshness_sources = _sources(freshness_json.get("sources"))
    ingestion_sources = _sources(ingestion_json.get("sources"))
    all_sources = [
        *_source_entries(freshness_sources, "freshness"),
        *_source_entries(ingestion_sources, "ingestion"),
    ]
    blocking_sources = _blocking_source_entries(
        source_entries=all_sources,
        freshness_blockers=_strings(freshness_json.get("blockers")),
        ingestion_blockers=_strings(ingestion_json.get("blockers")),
    )
    data_ready = (
        readback_present
        and freshness_json.get("status") == "ok"
        and ingestion_json.get("status") == "ok"
    )
    if data_ready:
        status = "stockplatform_source_contract_ready"
        safe_next_step = "run_post_apply_public_api_verifier_and_clear_p0_006_blocker"
    elif (
        readback_present
        and data_dependency_classification
        == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
    ):
        status = "stockplatform_source_contract_core_ready_ai_recommendations_stale"
        safe_next_step = (
            "run_stockplatform_ai_recommendation_pipeline_readback_then_rerun_"
            "public_api_verifier"
        )
    elif readback_present and data_dependency_classification == (
        "freshness_or_ingestion_not_ready"
    ):
        status = "blocked_stockplatform_source_contract_not_ready"
        safe_next_step = (
            "diff_stockplatform_source_contract_blocking_sources_then_rerun_"
            "public_api_verifier"
        )
    elif readback_present and postgres_not_ready:
        status = "stockplatform_source_contract_not_required_postgres_path"
        safe_next_step = (
            "run_stockplatform_production_migration_path_after_control_channel_"
            "readback_no_manual_db_write"
        )
    elif readback_present:
        status = "stockplatform_source_contract_review_required"
        safe_next_step = "review_stockplatform_data_dependency_classification"
    else:
        status = "blocked_stockplatform_source_contract_readback_unavailable"
        safe_next_step = (
            "restore_public_api_freshness_ingestion_readback_then_rerun_"
            "source_contract_readback"
        )

    active_blockers = [
        blocker
        for blocker in data_dependency_blockers
        if blocker.startswith("stockplatform_freshness_")
        or blocker.startswith("stockplatform_ingestion_")
    ]
    missing_evidence = []
    if freshness_http_status != 200:
        missing_evidence.append("freshness_endpoint_http_200")
    if ingestion_http_status != 200:
        missing_evidence.append("ingestion_endpoint_http_200")
    if readback_present and not all_sources and not active_blockers and not data_ready:
        missing_evidence.append("source_contract_sources_or_blockers")

    return {
        "schema_version": _SOURCE_CONTRACT_SCHEMA_VERSION,
        "receipt_key": _SOURCE_CONTRACT_RECEIPT_KEY,
        "status": status,
        "source": "stockplatform_public_api_freshness_ingestion_source_contract",
        "readback_present": readback_present,
        "receipt_provided": readback_present,
        "safe_next_step": safe_next_step,
        "source_contract_ready": data_ready,
        "data_dependency_classification": data_dependency_classification,
        "ai_recommendations_stale_after_core_recovered": (
            data_dependency_classification
            == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
        ),
        "postgres_not_ready": postgres_not_ready,
        "freshness_status": str(freshness_json.get("status") or "unknown"),
        "ingestion_status": str(ingestion_json.get("status") or "unknown"),
        "latest_trading_date": str(
            freshness_json.get("latest_trading_date")
            or ingestion_json.get("latest_trading_date")
            or ""
        ),
        "freshness_source_count": len(freshness_sources),
        "ingestion_source_count": len(ingestion_sources),
        "source_count": len(all_sources),
        "sla_source_count": _sla_source_count(freshness_sources)
        + _sla_source_count(ingestion_sources),
        "non_ok_source_count": len(blocking_sources),
        "source_contract_blockers": _unique_strings(
            [
                *(f"freshness:{blocker}" for blocker in _strings(freshness_json.get("blockers"))),
                *(f"ingestion:{blocker}" for blocker in _strings(ingestion_json.get("blockers"))),
            ]
        ),
        "blocking_sources": blocking_sources,
        "active_blockers": active_blockers,
        "missing_evidence": missing_evidence,
        "post_apply_verifier_required": not data_ready,
        "operation_boundaries": {
            "public_https_probe_only": True,
            "ssh_used": False,
            "docker_command_performed": False,
            "database_write_or_restore_performed": False,
            "stockplatform_manual_data_write_performed": False,
            "secret_value_collection_allowed": False,
            "runtime_action_performed": False,
        },
    }


def _provided_receipts_with_live_readback(
    provided_receipts: list[str],
    data_readiness_control_readback: dict[str, Any],
    source_contract_control_readback: dict[str, Any],
) -> list[str]:
    receipts = list(provided_receipts)
    if data_readiness_control_readback.get("receipt_provided") is not True:
        receipts = receipts
    else:
        receipts.append(_DATA_READINESS_RECEIPT_KEY)
    if source_contract_control_readback.get("receipt_provided") is True:
        receipts.append(_SOURCE_CONTRACT_RECEIPT_KEY)
    return _unique_strings(receipts)


def _missing_receipts_after_live_readback(
    missing_receipts: list[str],
    data_readiness_control_readback: dict[str, Any],
    source_contract_control_readback: dict[str, Any],
) -> list[str]:
    provided = set()
    if data_readiness_control_readback.get("receipt_provided") is True:
        provided.add(_DATA_READINESS_RECEIPT_KEY)
    if source_contract_control_readback.get("receipt_provided") is True:
        provided.add(_SOURCE_CONTRACT_RECEIPT_KEY)
    return [
        receipt
        for receipt in missing_receipts
        if receipt not in provided
    ]


def _recovery_control_active_blockers(
    *,
    receipt_blockers: list[str],
    missing_receipts: list[str],
    data_readiness_control_readback: dict[str, Any],
) -> list[str]:
    blockers = receipt_blockers or missing_receipts
    if data_readiness_control_readback.get("receipt_provided") is not True:
        return _unique_strings(blockers)
    if data_readiness_control_readback.get("postgres_not_ready") is True:
        return _unique_strings(blockers)
    return _unique_strings(
        [
            blocker
            for blocker in blockers
            if blocker != "stockplatform_postgres_not_ready"
        ]
    )


def _load_recovery_control_receipt(operations_dir: Path) -> dict[str, Any]:
    path = operations_dir / _RECOVERY_RECEIPT_FILE
    if not path.exists():
        return {}
    with path.open(encoding="utf-8") as handle:
        payload = json.load(handle)
    if not isinstance(payload, dict):
        raise ValueError(f"{path}: expected JSON object")
    if payload.get("schema_version") != (
        "stockplatform_public_api_runtime_recovery_control_receipt_v1"
    ):
        raise ValueError(
            f"{path}: unsupported schema_version={payload.get('schema_version')!r}"
        )
    return payload


def _blocked_safe_next_step(*, data_dependency_classification: str) -> str:
    if data_dependency_classification == "postgres_not_ready":
        return (
            "run_stockplatform_production_migration_path_after_control_channel_"
            "readback_then_rerun_public_api_readback"
        )
    if (
        data_dependency_classification
        == _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
    ):
        return (
            "run_stockplatform_ai_recommendation_pipeline_readback_then_rerun_"
            "public_api_verifier"
        )
    if data_dependency_classification == "freshness_or_ingestion_not_ready":
        return (
            "run_stockplatform_data_dependency_source_contract_readback_then_"
            "rerun_public_api_verifier"
        )
    return (
        "recover_stockplatform_api_container_runtime_or_docker_control_path_"
        "without_daemon_restart_then_rerun_public_api_readback"
    )


def _data_dependency_classification(
    active_blockers: list[str],
    *,
    api_health_ok: bool,
) -> str:
    if not api_health_ok:
        return "none"
    if any("postgres_not_ready" in blocker for blocker in active_blockers):
        return "postgres_not_ready"
    data_blockers = [
        blocker
        for blocker in active_blockers
        if blocker.startswith("stockplatform_freshness_")
        or blocker.startswith("stockplatform_ingestion_")
    ]
    if data_blockers == ["stockplatform_freshness_ai_recommendations_stale"]:
        return _AI_RECOMMENDATIONS_STALE_AFTER_CORE_RECOVERED
    if any(
        blocker.startswith("stockplatform_freshness_")
        or blocker.startswith("stockplatform_ingestion_")
        for blocker in active_blockers
    ):
        return "freshness_or_ingestion_not_ready"
    return "none"


def _active_blockers(
    *,
    api: dict[str, Any],
    freshness: dict[str, Any],
    ingestion: dict[str, Any],
    freshness_json: dict[str, Any],
    ingestion_json: dict[str, Any],
) -> list[str]:
    blockers: list[str] = []
    _append_http_blocker(blockers, "stockplatform_public_api_healthz", api)
    _append_http_blocker(blockers, "stockplatform_freshness", freshness)
    _append_http_blocker(blockers, "stockplatform_ingestion", ingestion)
    if freshness.get("http_status") == 200 and freshness_json.get("status") != "ok":
        blockers.extend(
            f"stockplatform_freshness_{item}"
            for item in _strings(freshness_json.get("blockers"))
        )
        if not _strings(freshness_json.get("blockers")):
            blockers.append("stockplatform_freshness_status_not_ok")
    if ingestion.get("http_status") == 200 and ingestion_json.get("status") != "ok":
        blockers.extend(
            f"stockplatform_ingestion_{item}"
            for item in _strings(ingestion_json.get("blockers"))
        )
        if not _strings(ingestion_json.get("blockers")):
            blockers.append("stockplatform_ingestion_status_not_ok")
    return _unique_strings(blockers)


def _append_http_blocker(
    blockers: list[str],
    prefix: str,
    probe: dict[str, Any],
) -> None:
    http_status = probe.get("http_status")
    if http_status == 200:
        return
    if isinstance(http_status, int):
        blockers.append(f"{prefix}_http_{http_status}")
        return
    blockers.append(f"{prefix}_unreachable")


def _probe_endpoint(
    probe: Probe,
    url: str,
    timeout_seconds: float,
    *,
    parse_json: bool,
    attempts: int,
) -> dict[str, Any]:
    bounded_attempts = max(1, int(attempts))
    last_payload: dict[str, Any] = {
        "url": url,
        "http_status": None,
        "ok": False,
        "error": "not_attempted",
        "attempt_count": 0,
    }
    for attempt in range(1, bounded_attempts + 1):
        result = probe(url, timeout_seconds)
        http_status = _int_or_none(result.get("http_status"))
        body = str(result.get("body") or "")
        payload: dict[str, Any] = {
            "url": url,
            "http_status": http_status,
            "ok": http_status == 200,
            "error": str(result.get("error") or ""),
            "attempt_count": attempt,
        }
        if parse_json and http_status == 200:
            try:
                payload["json"] = json.loads(body)
            except json.JSONDecodeError:
                payload["json"] = {}
                payload["error"] = "invalid_json"
        if payload.get("ok") and (not parse_json or payload.get("json")):
            return payload
        last_payload = payload
    return last_payload


def _http_probe(url: str, timeout_seconds: float) -> dict[str, Any]:
    request = urllib.request.Request(
        url,
        headers={"User-Agent": "awoooi-stockplatform-public-api-readback"},
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
            body = response.read(_MAX_READBACK_BODY_BYTES).decode(
                "utf-8", "replace"
            )
            return {"http_status": response.status, "body": body, "error": ""}
    except urllib.error.HTTPError as exc:
        body = exc.read(_MAX_READBACK_BODY_BYTES).decode("utf-8", "replace")
        return {"http_status": exc.code, "body": body, "error": ""}
    except Exception as exc:  # noqa: BLE001 - readback must fail closed.
        return {"http_status": None, "body": "", "error": _error_text(exc)}


def _error_text(exc: Exception) -> str:
    reason = getattr(exc, "reason", "")
    if reason:
        return f"{type(exc).__name__}: {reason}"
    return type(exc).__name__


def _dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def _sources(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    return [item for item in value if isinstance(item, dict)]


def _source_entries(
    sources: list[dict[str, Any]],
    category: str,
) -> list[dict[str, Any]]:
    entries: list[dict[str, Any]] = []
    for source in sources:
        source_id = str(
            source.get("source")
            or source.get("name")
            or source.get("id")
            or source.get("table")
            or "unknown"
        )
        status = str(source.get("status") or "unknown")
        reason = str(
            source.get("reason")
            or source.get("message")
            or source.get("blocker")
            or ""
        )
        entries.append(
            {
                "source": source_id,
                "category": category,
                "status": status,
                "reason": reason,
                "has_sla": isinstance(source.get("sla"), dict),
            }
        )
    return entries


def _blocking_source_entries(
    *,
    source_entries: list[dict[str, Any]],
    freshness_blockers: list[str],
    ingestion_blockers: list[str],
) -> list[dict[str, Any]]:
    blocking = [
        {
            **entry,
            "classification": f"{entry['category']}_source_non_ok",
        }
        for entry in source_entries
        if entry.get("status") != "ok"
    ]
    if blocking:
        return blocking
    return [
        {
            "source": blocker,
            "category": "freshness",
            "status": "blocked",
            "reason": blocker,
            "has_sla": False,
            "classification": "freshness_blocker_without_source_entry",
        }
        for blocker in freshness_blockers
    ] + [
        {
            "source": blocker,
            "category": "ingestion",
            "status": "blocked",
            "reason": blocker,
            "has_sla": False,
            "classification": "ingestion_blocker_without_source_entry",
        }
        for blocker in ingestion_blockers
    ]


def _sla_source_count(sources: list[dict[str, Any]]) -> int:
    return sum(1 for source in sources if isinstance(source.get("sla"), dict))


def _non_ok_source_count(sources: list[dict[str, Any]]) -> int:
    return sum(1 for source in sources if source.get("status") != "ok")


def _int_or_none(value: Any) -> int | None:
    if isinstance(value, bool):
        return int(value)
    if isinstance(value, int | float):
        return int(value)
    return None


def _strings(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    return [str(item) for item in value if item is not None]


def _unique_strings(values: list[str]) -> list[str]:
    seen: set[str] = set()
    unique: list[str] = []
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        unique.append(value)
    return unique
