"""Harbor registry controlled recovery receipt validation.

This validator turns non-secret 110 local repair/watchdog output plus public
registry verifiers into a machine-readable receipt. It never executes SSH,
Docker, systemctl, workflow dispatch, or runtime writes, and it does not echo
raw log text back to callers.
"""

from __future__ import annotations

from typing import Any

_SCHEMA_VERSION = "harbor_registry_controlled_recovery_receipt_v1"
_READY_HTTP_STATUSES = {200, 401}


def validate_harbor_registry_controlled_recovery_receipt(
    receipt_payload: dict[str, Any],
) -> dict[str, Any]:
    """Validate a Harbor registry controlled recovery receipt payload."""
    if not isinstance(receipt_payload, dict):
        raise ValueError("receipt payload must be a JSON object")

    ssh_diagnosis_output = _text(
        receipt_payload.get("ssh_publickey_diagnosis_output")
    )
    local_console_output = _text(receipt_payload.get("local_console_recovery_output"))
    ssh_local_output = _text(receipt_payload.get("ssh_local_repair_output"))
    watchdog_check_output = _text(receipt_payload.get("watchdog_check_output"))
    watchdog_repair_output = _text(receipt_payload.get("watchdog_repair_output"))
    controlled_cd_lane_output = _text(
        receipt_payload.get("controlled_cd_lane_readiness_output")
    )
    non110_runner_output = _text(
        receipt_payload.get("non110_runner_readiness_output")
    )

    ssh_diagnosis = _parse_ssh_publickey_diagnosis_output(
        _receipt_output(ssh_diagnosis_output, local_console_output)
    )
    ssh_local = _parse_ssh_local_repair_output(
        _receipt_output(ssh_local_output, local_console_output)
    )
    watchdog_check = _parse_watchdog_output(
        _receipt_output(watchdog_check_output, local_console_output)
    )
    watchdog_repair = _parse_watchdog_output(
        _receipt_output(watchdog_repair_output, local_console_output)
    )
    controlled_cd_lane = _parse_controlled_cd_lane_readiness_output(
        _receipt_output(controlled_cd_lane_output, local_console_output)
    )
    non110_runner = _parse_non110_runner_readiness_output(
        _receipt_output(non110_runner_output, local_console_output)
    )
    verifier = _post_apply_verifier(receipt_payload)
    gitea_queue = _gitea_queue_readback(
        receipt_payload.get("gitea_actions_queue_readback")
    )
    deploy_marker = _deploy_marker_readback(
        receipt_payload.get("deploy_marker_readback")
    )

    active_blockers = _active_blockers(
        ssh_diagnosis=ssh_diagnosis,
        ssh_local=ssh_local,
        watchdog_check=watchdog_check,
        watchdog_repair=watchdog_repair,
        controlled_cd_lane=controlled_cd_lane,
        non110_runner=non110_runner,
        verifier=verifier,
        gitea_queue=gitea_queue,
        deploy_marker=deploy_marker,
    )
    status = _status(
        ssh_diagnosis=ssh_diagnosis,
        ssh_local=ssh_local,
        watchdog_check=watchdog_check,
        watchdog_repair=watchdog_repair,
        controlled_cd_lane=controlled_cd_lane,
        non110_runner=non110_runner,
        verifier=verifier,
        gitea_queue=gitea_queue,
        deploy_marker=deploy_marker,
        active_blockers=active_blockers,
    )
    local_console_phase_readback = _local_console_phase_readback(
        ssh_diagnosis=ssh_diagnosis,
        ssh_local=ssh_local,
        watchdog_check=watchdog_check,
        watchdog_repair=watchdog_repair,
        controlled_cd_lane=controlled_cd_lane,
        non110_runner=non110_runner,
        verifier=verifier,
        gitea_queue=gitea_queue,
    )
    control_path_readiness = _control_path_readiness(
        ssh_diagnosis=ssh_diagnosis,
        non110_runner=non110_runner,
        verifier=verifier,
        gitea_queue=gitea_queue,
    )

    return {
        "schema_version": _SCHEMA_VERSION,
        "priority": "P0-006",
        "scope": "harbor_registry_controlled_recovery_receipt",
        "status": status,
        "accepted": status == "harbor_registry_recovery_receipt_verified",
        "safe_next_step": _safe_next_step(status=status),
        "active_blockers": active_blockers,
        "active_blocker_count": len(active_blockers),
        "input_redaction": {
            "raw_output_returned": False,
            "ssh_publickey_diagnosis_output": _text_stats(
                ssh_diagnosis_output
            ),
            "ssh_local_repair_output": _text_stats(ssh_local_output),
            "local_console_recovery_output": _text_stats(local_console_output),
            "watchdog_check_output": _text_stats(watchdog_check_output),
            "watchdog_repair_output": _text_stats(watchdog_repair_output),
            "controlled_cd_lane_readiness_output": _text_stats(
                controlled_cd_lane_output
            ),
            "non110_runner_readiness_output": _text_stats(non110_runner_output),
            "gitea_actions_queue_readback": {
                "provided": gitea_queue["receipt_seen"],
                "metadata_only": True,
                "raw_output_returned": False,
                "top_visible_runs_returned": False,
            },
            "deploy_marker_readback": {
                "provided": deploy_marker["receipt_seen"],
                "metadata_only": True,
                "raw_output_returned": False,
            },
        },
        "readback": {
            "ssh_publickey_diagnosis": ssh_diagnosis,
            "ssh_local_repair": ssh_local,
            "watchdog_check": watchdog_check,
            "watchdog_repair": watchdog_repair,
            "controlled_cd_lane_readiness": controlled_cd_lane,
            "non110_runner_readiness": non110_runner,
            "post_apply_verifier": verifier,
            "gitea_actions_queue": gitea_queue,
            "deploy_marker": deploy_marker,
            "control_path_readiness": control_path_readiness,
        },
        "local_console_phase_readback": local_console_phase_readback,
        "controlled_apply_policy": {
            "risk_level": "high",
            "break_glass_required": False,
            "manual_end_state": False,
            "current_apply_allowed": False,
            "current_apply_blocker": _current_apply_blocker(status=status),
            "allowed_scope": [
                "validate_non_secret_ssh_publickey_diagnosis_receipt",
                "validate_non_secret_ssh_local_repair_receipt",
                "validate_harbor_watchdog_check_receipt",
                "validate_harbor_watchdog_repair_once_receipt",
                "validate_110_controlled_cd_lane_readiness_receipt",
                "validate_non110_runner_readiness_receipt",
                "validate_public_and_internal_registry_v2_verifier",
                "km_rag_mcp_playbook_metadata_writeback",
                "retry_gitea_cd_after_registry_v2_green",
            ],
            "forbidden_scope": [
                "secret_read",
                "docker_daemon_restart",
                "host_reboot",
                "node_drain",
                "nginx_reload_or_restart",
                "database_restore_or_prune",
                "workflow_dispatch_from_receipt_validator",
                "force_push_or_repo_ref_mutation",
                "registry_provider_switch",
            ],
        },
        "learning_writeback_contracts": _learning_writeback_contracts(status=status),
        "rollups": {
            "ssh_publickey_diagnosis_receipt_seen": ssh_diagnosis[
                "receipt_seen"
            ],
            "ssh_publickey_diagnosis_ready": ssh_diagnosis["diagnosis_ready"],
            "ssh_publickey_wooo_publickey_classification": ssh_diagnosis[
                "wooo_publickey_classification"
            ],
            "ssh_publickey_wooo_command_path_classification": ssh_diagnosis[
                "wooo_command_path_classification"
            ],
            "ssh_publickey_wooo_command_path_ready": ssh_diagnosis[
                "wooo_command_path_ready"
            ],
            "ssh_publickey_offer_timeout_seen": ssh_diagnosis[
                "publickey_offer_timeout_seen"
            ],
            "ssh_publickey_offer_timeout_effective_seen": ssh_diagnosis[
                "publickey_offer_timeout_effective_seen"
            ],
            "ssh_publickey_server_accepts_key_then_timeout_seen": (
                ssh_diagnosis["server_accepts_key_then_timeout_seen"]
            ),
            "ssh_publickey_server_accepts_key_then_timeout_effective_seen": (
                ssh_diagnosis[
                    "server_accepts_key_then_timeout_effective_seen"
                ]
            ),
            "ssh_publickey_runner_systemctl_show_timeout_seen": (
                ssh_diagnosis["runner_systemctl_show_timeout_seen"]
            ),
            "ssh_publickey_runner_systemctl_show_timeout_effective_seen": (
                ssh_diagnosis["runner_systemctl_show_timeout_effective_seen"]
            ),
            "ssh_publickey_node_high_load_seen": ssh_diagnosis[
                "node_high_load_seen"
            ],
            "ssh_publickey_node_load_classifier": ssh_diagnosis[
                "node_load_classifier"
            ],
            "ssh_publickey_node_load1_per_cpu": ssh_diagnosis[
                "node_load1_per_cpu"
            ],
            "ssh_publickey_node_exporter_ok": ssh_diagnosis["node_exporter_ok"],
            "ssh_publickey_port_tcp_open": ssh_diagnosis["ssh_port_tcp_open"],
            "ssh_local_repair_receipt_seen": ssh_local["receipt_seen"],
            "ssh_local_repair_control_channel_metadata_ready": ssh_local[
                "control_channel_metadata_ready"
            ],
            "ssh_local_repair_account_metadata_ready": ssh_local[
                "account_metadata_ready"
            ],
            "ssh_local_repair_target_user_account_locked": ssh_local[
                "target_user_account_locked"
            ],
            "ssh_local_repair_target_user_shell_executable": ssh_local[
                "target_user_shell_executable"
            ],
            "ssh_local_repair_sshd_pubkeyauthentication": ssh_local[
                "sshd_pubkeyauthentication"
            ],
            "ssh_local_repair_sshd_authorized_keys_file_default": ssh_local[
                "sshd_authorized_keys_file_default"
            ],
            "watchdog_check_receipt_seen": watchdog_check["receipt_seen"],
            "watchdog_check_harbor_ready": watchdog_check["harbor_ready"],
            "watchdog_repair_receipt_seen": watchdog_repair["receipt_seen"],
            "watchdog_repair_harbor_ready": watchdog_repair["harbor_ready"],
            "controlled_cd_lane_readiness_receipt_seen": controlled_cd_lane[
                "receipt_seen"
            ],
            "controlled_cd_lane_ready": controlled_cd_lane[
                "controlled_cd_lane_ready"
            ],
            "controlled_cd_lane_blocker_count": controlled_cd_lane[
                "blocker_count"
            ],
            "controlled_cd_lane_safe_next_step": controlled_cd_lane[
                "safe_next_step"
            ],
            "non110_runner_readiness_receipt_seen": non110_runner[
                "receipt_seen"
            ],
            "non110_runner_ready": non110_runner["non110_runner_ready"],
            "non110_runner_blocker_count": non110_runner["blocker_count"],
            "non110_runner_safe_next_step": non110_runner["safe_next_step"],
            "non110_runner_ready_config_count": non110_runner[
                "ready_config_count"
            ],
            "non110_runner_ready_binary_count": non110_runner[
                "ready_binary_count"
            ],
            "non110_runner_ready_registration_count": non110_runner[
                "ready_registration_count"
            ],
            "non110_runner_ready_service_count": non110_runner[
                "ready_service_count"
            ],
            "non110_runner_ready_active_service_count": non110_runner[
                "ready_active_service_count"
            ],
            "non110_runner_ready_autostart_path_count": non110_runner[
                "ready_autostart_path_count"
            ],
            "post_apply_verifier_ready": verifier["registry_v2_ready"],
            "gitea_queue_readback_seen": gitea_queue["receipt_seen"],
            "gitea_queue_blocker_count": gitea_queue["blocker_count"],
            "gitea_queue_harbor_110_no_matching_runner": gitea_queue[
                "harbor_110_repair_no_matching_runner"
            ],
            "gitea_queue_harbor_110_jobs_stale_or_mismatched": gitea_queue[
                "harbor_110_repair_jobs_stale_or_mismatched"
            ],
            "gitea_queue_harbor_110_jobs_cross_workflow_mismatch": gitea_queue[
                "harbor_110_repair_jobs_cross_workflow_mismatch"
            ],
            "gitea_queue_harbor_110_jobs_payload_classifier": gitea_queue[
                "harbor_110_repair_jobs_payload_classifier"
            ],
            "gitea_queue_harbor_110_repair_visible_running_jobs_api_stale": (
                gitea_queue["harbor_110_repair_visible_running_jobs_api_stale"]
            ),
            "gitea_queue_current_cd_waiting_behind_harbor_110_repair_running": (
                gitea_queue["current_cd_waiting_behind_harbor_110_repair_running"]
            ),
            "gitea_queue_latest_cd_run_status": gitea_queue[
                "latest_visible_cd_run_status"
            ],
            "gitea_queue_current_cd_waiting_for_runner_or_queue": gitea_queue[
                "current_cd_waiting_for_runner_or_queue"
            ],
            "gitea_queue_current_cd_no_matching_runner_label": gitea_queue[
                "current_cd_no_matching_runner_label"
            ],
            "gitea_queue_controlled_profile_no_matching_runner_label_count": (
                gitea_queue["controlled_profile_no_matching_runner_label_count"]
            ),
            "gitea_queue_harbor_110_repair_run_status": gitea_queue[
                "latest_visible_harbor_110_repair_run_status"
            ],
            "gitea_queue_harbor_110_repair_failure_classifier": gitea_queue[
                "harbor_110_repair_failure_classifier"
            ],
            "gitea_queue_harbor_110_remote_control_channel_unavailable": (
                gitea_queue["harbor_110_repair_remote_control_channel_unavailable"]
            ),
            "gitea_queue_harbor_110_bounded_ssh_timeout_seen": gitea_queue[
                "harbor_110_repair_bounded_ssh_timeout_seen"
            ],
            "gitea_queue_current_cd_inflight_classifier": gitea_queue[
                "current_cd_inflight_classifier"
            ],
            "gitea_queue_current_cd_harbor_latest_registry_v2_status": (
                gitea_queue["current_cd_harbor_latest_registry_v2_status"]
            ),
            "gitea_queue_current_cd_harbor_retrying_unavailable": gitea_queue[
                "current_cd_harbor_retrying_unavailable"
            ],
            "gitea_queue_current_cd_harbor_repair_requires_110_controlled_lane": (
                gitea_queue["current_cd_harbor_repair_requires_110_controlled_lane"]
            ),
            "gitea_queue_current_cd_harbor_repair_blocked_by_awoooi_host": (
                gitea_queue[
                    "current_cd_harbor_repair_blocked_by_no_matching_awoooi_host"
                ]
            ),
            "gitea_queue_current_cd_harbor_repair_lane_classifier": gitea_queue[
                "current_cd_harbor_repair_lane_classifier"
            ],
            "gitea_queue_cd_jobs_stale_or_mismatched": gitea_queue[
                "cd_run_jobs_stale_or_mismatched"
            ],
            "gitea_queue_cd_jobs_payload_classifier": gitea_queue[
                "cd_run_jobs_payload_classifier"
            ],
            "gitea_queue_cd_jobs_head_sha_mismatch": gitea_queue[
                "cd_run_jobs_head_sha_mismatch"
            ],
            "gitea_queue_cd_jobs_run_id_mismatch": gitea_queue[
                "cd_run_jobs_run_id_mismatch"
            ],
            "gitea_queue_normalized_classifier_field_count": gitea_queue[
                "normalized_classifier_field_count"
            ],
            "gitea_queue_normalized_classifier_field_ids": gitea_queue[
                "normalized_classifier_field_ids"
            ],
            "deploy_marker_readback_seen": deploy_marker["receipt_seen"],
            "deploy_marker_verified": deploy_marker["deploy_marker_verified"],
            "deploy_marker_blocker_count": deploy_marker["blocker_count"],
            "deploy_marker_latest_cd_run_status": deploy_marker[
                "latest_visible_cd_run_status"
            ],
            "deploy_marker_production_image_matches_expected": deploy_marker[
                "production_image_matches_expected"
            ],
            "control_path_readiness_status": control_path_readiness["status"],
            "control_path_readiness_primary_blocker": control_path_readiness[
                "primary_blocker"
            ],
            "control_path_readiness_blocker_count": control_path_readiness[
                "blocker_count"
            ],
            "control_path_readiness_node_high_load": control_path_readiness[
                "node_high_load"
            ],
            "control_path_readiness_runner_systemctl_timeout": (
                control_path_readiness["runner_systemctl_show_timeout"]
            ),
            "control_path_readiness_awoooi_host_unavailable": (
                control_path_readiness["awoooi_host_runner_unavailable"]
            ),
            "control_path_readiness_non110_runner_unavailable": (
                control_path_readiness["non110_runner_unavailable"]
            ),
            "control_path_readiness_registry_v2_public_ready": (
                control_path_readiness["registry_v2_public_ready"]
            ),
            "control_path_readiness_registry_v2_internal_ready": (
                control_path_readiness["registry_v2_internal_ready"]
            ),
            "metadata_writeback_contract_ready": True,
            "local_console_phase_count": local_console_phase_readback["phase_count"],
            "local_console_completed_phase_count": local_console_phase_readback[
                "completed_phase_count"
            ],
            "local_console_blocked_phase_count": local_console_phase_readback[
                "blocked_phase_count"
            ],
        },
        "operation_boundaries": {
            "raw_output_returned": False,
            "ssh_used": False,
            "docker_command_performed": False,
            "docker_restart_performed": False,
            "docker_daemon_restart_performed": False,
            "host_reboot_performed": False,
            "node_drain_performed": False,
            "nginx_reload_or_restart_performed": False,
            "database_write_or_restore_performed": False,
            "secret_value_collection_allowed": False,
            "github_api_used": False,
            "workflow_trigger_performed": False,
            "runtime_write_allowed_by_this_validator": False,
        },
    }


def _control_path_readiness(
    *,
    ssh_diagnosis: dict[str, Any],
    non110_runner: dict[str, Any],
    verifier: dict[str, Any],
    gitea_queue: dict[str, Any],
) -> dict[str, Any]:
    public_registry_ready = bool(verifier["public_registry_v2_ready"])
    internal_registry_ready = bool(verifier["internal_registry_v2_ready"])
    command_path_ready = bool(ssh_diagnosis["wooo_command_path_ready"])
    harbor_110_repair_queue_superseded = (
        command_path_ready and public_registry_ready and internal_registry_ready
    )
    queue_no_matching_runner = bool(
        gitea_queue["harbor_110_repair_no_matching_runner"]
    ) and not harbor_110_repair_queue_superseded
    queue_jobs_stale = bool(
        gitea_queue["harbor_110_repair_jobs_stale_or_mismatched"]
    ) and not harbor_110_repair_queue_superseded
    queue_jobs_cross_workflow = bool(
        gitea_queue["harbor_110_repair_jobs_cross_workflow_mismatch"]
    ) and not harbor_110_repair_queue_superseded
    queue_harbor_running_jobs_api_stale = bool(
        gitea_queue["harbor_110_repair_visible_running_jobs_api_stale"]
    ) and not harbor_110_repair_queue_superseded
    current_cd_waiting_behind_harbor_running = bool(
        gitea_queue["current_cd_waiting_behind_harbor_110_repair_running"]
    ) and not harbor_110_repair_queue_superseded
    queue_remote_control_unavailable_raw = bool(
        gitea_queue["harbor_110_repair_remote_control_channel_unavailable"]
    )
    queue_remote_control_unavailable = (
        queue_remote_control_unavailable_raw
        and not harbor_110_repair_queue_superseded
    )
    cd_harbor_repair_requires_110_lane = bool(
        gitea_queue["current_cd_harbor_repair_requires_110_controlled_lane"]
    )
    cd_harbor_repair_blocked_by_awoooi_host = bool(
        gitea_queue["current_cd_harbor_repair_blocked_by_no_matching_awoooi_host"]
    )
    cd_jobs_stale = bool(gitea_queue["cd_run_jobs_stale_or_mismatched"])
    cd_jobs_head_sha_mismatch = bool(gitea_queue["cd_run_jobs_head_sha_mismatch"])
    current_cd_waiting = bool(gitea_queue["current_cd_waiting_for_runner_or_queue"])
    current_cd_no_matching_runner = bool(
        gitea_queue["current_cd_no_matching_runner_label"]
    )
    controlled_profile_no_matching_runner = (
        gitea_queue["controlled_profile_no_matching_runner_label_count"] > 0
    )
    non110_runner_not_ready = bool(
        non110_runner["receipt_seen"] and not non110_runner["non110_runner_ready"]
    )
    runner_timeout = bool(
        ssh_diagnosis["runner_systemctl_show_timeout_effective_seen"]
    )
    node_high_load = bool(ssh_diagnosis["node_high_load_seen"])
    publickey_offer_timeout = bool(
        ssh_diagnosis["publickey_offer_timeout_effective_seen"]
    )
    server_accepts_key_then_timeout = bool(
        ssh_diagnosis["server_accepts_key_then_timeout_effective_seen"]
    )
    awoooi_host_unavailable = (
        queue_no_matching_runner
        or queue_jobs_stale
        or queue_harbor_running_jobs_api_stale
        or current_cd_waiting_behind_harbor_running
        or current_cd_no_matching_runner
        or controlled_profile_no_matching_runner
    )
    non110_runner_unavailable = (
        non110_runner_not_ready
        or current_cd_no_matching_runner
        or controlled_profile_no_matching_runner
    )
    registry_v2_ready = public_registry_ready and internal_registry_ready
    signal_ids = _control_path_signal_ids(
        node_high_load=node_high_load,
        publickey_offer_timeout=publickey_offer_timeout,
        server_accepts_key_then_timeout=server_accepts_key_then_timeout,
        runner_timeout=runner_timeout,
        queue_no_matching_runner=queue_no_matching_runner,
        queue_jobs_stale=queue_jobs_stale,
        queue_jobs_cross_workflow=queue_jobs_cross_workflow,
        queue_harbor_running_jobs_api_stale=queue_harbor_running_jobs_api_stale,
        current_cd_waiting_behind_harbor_running=(
            current_cd_waiting_behind_harbor_running
        ),
        queue_remote_control_unavailable=queue_remote_control_unavailable,
        cd_harbor_repair_requires_110_lane=cd_harbor_repair_requires_110_lane,
        cd_harbor_repair_blocked_by_awoooi_host=(
            cd_harbor_repair_blocked_by_awoooi_host
        ),
        current_cd_waiting=current_cd_waiting,
        current_cd_no_matching_runner=current_cd_no_matching_runner,
        controlled_profile_no_matching_runner=controlled_profile_no_matching_runner,
        non110_runner_not_ready=non110_runner_not_ready,
        cd_jobs_head_sha_mismatch=cd_jobs_head_sha_mismatch,
        cd_jobs_stale=cd_jobs_stale,
        public_registry_ready=public_registry_ready,
        internal_registry_ready=internal_registry_ready,
    )
    status = _control_path_readiness_status(
        node_high_load=node_high_load,
        runner_timeout=runner_timeout,
        ssh_timeout=(
            publickey_offer_timeout
            or server_accepts_key_then_timeout
            or queue_remote_control_unavailable
        ),
        awoooi_host_unavailable=awoooi_host_unavailable,
        non110_runner_unavailable=non110_runner_unavailable,
        registry_v2_ready=registry_v2_ready,
        signal_ids=signal_ids,
    )
    return {
        "status": status,
        "primary_blocker": signal_ids[0] if signal_ids else "",
        "safe_next_action": _control_path_safe_next_action(status=status),
        "signal_ids": signal_ids,
        "blocker_count": len(signal_ids),
        "ssh_diagnosis_receipt_seen": ssh_diagnosis["receipt_seen"],
        "gitea_queue_readback_seen": gitea_queue["receipt_seen"],
        "node_high_load": node_high_load,
        "node_load_classifier": ssh_diagnosis["node_load_classifier"],
        "node_load1_per_cpu": ssh_diagnosis["node_load1_per_cpu"],
        "ssh_port_tcp_open": ssh_diagnosis["ssh_port_tcp_open"],
        "ssh_command_path_ready": command_path_ready,
        "ssh_command_path_classification": ssh_diagnosis[
            "wooo_command_path_classification"
        ],
        "ssh_publickey_offer_timeout": publickey_offer_timeout,
        "ssh_server_accepts_key_then_timeout": server_accepts_key_then_timeout,
        "ssh_publickey_offer_timeout_raw": ssh_diagnosis[
            "publickey_offer_timeout_seen"
        ],
        "ssh_server_accepts_key_then_timeout_raw": ssh_diagnosis[
            "server_accepts_key_then_timeout_seen"
        ],
        "runner_systemctl_show_timeout": runner_timeout,
        "runner_systemctl_show_timeout_raw": ssh_diagnosis[
            "runner_systemctl_show_timeout_seen"
        ],
        "awoooi_host_runner_unavailable": awoooi_host_unavailable,
        "non110_runner_unavailable": non110_runner_unavailable,
        "non110_runner_readiness_receipt_seen": non110_runner["receipt_seen"],
        "non110_runner_ready": non110_runner["non110_runner_ready"],
        "non110_runner_blocker_count": non110_runner["blocker_count"],
        "non110_runner_blockers": non110_runner["blockers"],
        "non110_runner_safe_next_step": non110_runner["safe_next_step"],
        "current_cd_no_matching_runner_label": gitea_queue[
            "current_cd_no_matching_runner_label"
        ],
        "current_cd_waiting_for_runner_or_queue": current_cd_waiting,
        "controlled_profile_no_matching_runner_label_count": gitea_queue[
            "controlled_profile_no_matching_runner_label_count"
        ],
        "harbor_110_repair_no_matching_runner": queue_no_matching_runner,
        "harbor_110_repair_no_matching_runner_label": gitea_queue[
            "harbor_110_repair_no_matching_runner_label"
        ],
        "harbor_110_repair_jobs_stale_or_mismatched": queue_jobs_stale,
        "harbor_110_repair_jobs_cross_workflow_mismatch": queue_jobs_cross_workflow,
        "harbor_110_repair_visible_running_jobs_api_stale": (
            queue_harbor_running_jobs_api_stale
        ),
        "current_cd_waiting_behind_harbor_110_repair_running": (
            current_cd_waiting_behind_harbor_running
        ),
        "harbor_110_repair_remote_control_channel_unavailable": (
            queue_remote_control_unavailable
        ),
        "harbor_110_repair_remote_control_channel_unavailable_raw": (
            queue_remote_control_unavailable_raw
        ),
        "harbor_110_repair_failure_classifier": gitea_queue[
            "harbor_110_repair_failure_classifier"
        ],
        "harbor_110_repair_remote_ssh_reachable": gitea_queue[
            "harbor_110_repair_remote_ssh_reachable"
        ],
        "harbor_110_repair_bounded_ssh_timeout_seen": gitea_queue[
            "harbor_110_repair_bounded_ssh_timeout_seen"
        ],
        "cd_harbor_repair_requires_110_controlled_lane": (
            cd_harbor_repair_requires_110_lane
        ),
        "cd_harbor_repair_blocked_by_no_matching_awoooi_host": (
            cd_harbor_repair_blocked_by_awoooi_host
        ),
        "cd_harbor_repair_lane_classifier": gitea_queue[
            "current_cd_harbor_repair_lane_classifier"
        ],
        "cd_jobs_stale_or_mismatched": cd_jobs_stale,
        "cd_jobs_head_sha_mismatch": cd_jobs_head_sha_mismatch,
        "registry_v2_public_http_status": verifier[
            "public_registry_v2_http_status"
        ],
        "registry_v2_internal_http_status": verifier[
            "internal_registry_v2_http_status"
        ],
        "registry_v2_public_ready": public_registry_ready,
        "registry_v2_internal_ready": internal_registry_ready,
        "registry_v2_ready": registry_v2_ready,
        "metadata_only": True,
        "raw_output_returned": False,
    }


def _control_path_signal_ids(
    *,
    node_high_load: bool,
    publickey_offer_timeout: bool,
    server_accepts_key_then_timeout: bool,
    runner_timeout: bool,
    queue_no_matching_runner: bool,
    queue_jobs_stale: bool,
    queue_jobs_cross_workflow: bool,
    queue_harbor_running_jobs_api_stale: bool,
    current_cd_waiting_behind_harbor_running: bool,
    queue_remote_control_unavailable: bool,
    cd_harbor_repair_requires_110_lane: bool,
    cd_harbor_repair_blocked_by_awoooi_host: bool,
    current_cd_waiting: bool,
    current_cd_no_matching_runner: bool,
    controlled_profile_no_matching_runner: bool,
    non110_runner_not_ready: bool,
    cd_jobs_head_sha_mismatch: bool,
    cd_jobs_stale: bool,
    public_registry_ready: bool,
    internal_registry_ready: bool,
) -> list[str]:
    signal_ids: list[str] = []
    if node_high_load:
        signal_ids.append("ssh_publickey_node_high_load_on_110")
    if publickey_offer_timeout:
        signal_ids.append("ssh_publickey_offer_timeout_on_wooo")
    if server_accepts_key_then_timeout:
        signal_ids.append("ssh_publickey_server_accepts_key_then_timeout_on_wooo")
    if runner_timeout:
        signal_ids.append("runner_systemctl_show_timeout_on_110")
    if queue_no_matching_runner:
        signal_ids.append("gitea_queue_harbor_110_repair_no_matching_runner")
    if queue_jobs_cross_workflow:
        signal_ids.append("gitea_queue_harbor_110_repair_jobs_cross_workflow_mismatch")
    if queue_jobs_stale:
        signal_ids.append("gitea_queue_harbor_110_repair_jobs_stale_or_mismatched")
    if queue_harbor_running_jobs_api_stale:
        signal_ids.append(
            "gitea_queue_harbor_110_repair_visible_running_jobs_api_stale"
        )
    if current_cd_waiting_behind_harbor_running:
        signal_ids.append(
            "gitea_queue_current_cd_waiting_behind_harbor_110_repair_running"
        )
    if queue_remote_control_unavailable:
        signal_ids.append("gitea_queue_harbor_110_remote_control_channel_unavailable")
    if cd_harbor_repair_blocked_by_awoooi_host:
        signal_ids.append(
            "gitea_queue_cd_harbor_repair_blocked_by_no_matching_awoooi_host"
        )
    if cd_harbor_repair_requires_110_lane:
        signal_ids.append("gitea_queue_cd_harbor_repair_requires_110_controlled_lane")
    if current_cd_no_matching_runner:
        signal_ids.append("gitea_queue_current_cd_no_matching_runner")
    if current_cd_waiting:
        signal_ids.append("gitea_queue_current_cd_waiting_for_runner_or_queue")
    if controlled_profile_no_matching_runner:
        signal_ids.append("gitea_queue_controlled_profile_no_matching_runner_labels")
    if non110_runner_not_ready:
        signal_ids.append("non110_runner_readiness_receipt_not_ready")
    if cd_jobs_head_sha_mismatch:
        signal_ids.append("gitea_queue_cd_jobs_head_sha_mismatch")
    if cd_jobs_stale:
        signal_ids.append("gitea_queue_cd_jobs_stale_or_mismatched")
    if not public_registry_ready:
        signal_ids.append("public_registry_v2_verifier_not_green")
    if not internal_registry_ready:
        signal_ids.append("internal_registry_v2_verifier_not_green")
    return signal_ids


def _control_path_readiness_status(
    *,
    node_high_load: bool,
    runner_timeout: bool,
    ssh_timeout: bool,
    awoooi_host_unavailable: bool,
    non110_runner_unavailable: bool,
    registry_v2_ready: bool,
    signal_ids: list[str],
) -> str:
    if not signal_ids:
        return "ready"
    if node_high_load and awoooi_host_unavailable:
        return "blocked_110_high_load_and_awoooi_host_control_path_unavailable"
    if non110_runner_unavailable:
        return "blocked_non110_controlled_runner_unavailable"
    if runner_timeout or ssh_timeout:
        return "blocked_110_ssh_or_runner_control_path_timeout"
    if awoooi_host_unavailable:
        return "blocked_awoooi_host_runner_queue_unavailable"
    if node_high_load:
        return "blocked_110_node_high_load"
    if any(item.startswith("gitea_queue_cd_jobs_") for item in signal_ids):
        return "blocked_gitea_cd_jobs_readback_stale_or_mismatched"
    if not registry_v2_ready:
        return "blocked_registry_v2_verifier_not_green"
    return "blocked_control_path_evidence_not_clear"


def _control_path_safe_next_action(*, status: str) -> str:
    if status == "ready":
        return "retry_gitea_cd_then_verify_deploy_marker_and_priority_readback"
    if status == "blocked_110_high_load_and_awoooi_host_control_path_unavailable":
        return "hold_110_capacity_protection_then_rerun_readonly_awoooi_host_and_registry_verifiers"
    if status == "blocked_awoooi_host_runner_queue_unavailable":
        return "run_awoooi_host_lane_readiness_verifier_before_controlled_lane_restore"
    if status == "blocked_non110_controlled_runner_unavailable":
        return "submit_non110_runner_readiness_receipt_then_restore_non110_controlled_lane"
    if status == "blocked_110_node_high_load":
        return "wait_for_110_load_to_normalize_then_rerun_readonly_control_path_probe"
    if status == "blocked_110_ssh_or_runner_control_path_timeout":
        return "rerun_non_secret_110_ssh_and_runner_control_path_diagnosis"
    if status == "blocked_registry_v2_verifier_not_green":
        return "rerun_public_and_internal_registry_v2_verifier_before_cd_retry"
    if status == "blocked_gitea_cd_jobs_readback_stale_or_mismatched":
        return "rerun_public_gitea_queue_and_deploy_marker_readback_before_cd_retry"
    return "normalize_control_path_evidence_then_retry_readback"


def _local_console_phase_readback(
    *,
    ssh_diagnosis: dict[str, Any],
    ssh_local: dict[str, Any],
    watchdog_check: dict[str, Any],
    watchdog_repair: dict[str, Any],
    controlled_cd_lane: dict[str, Any],
    non110_runner: dict[str, Any],
    verifier: dict[str, Any],
    gitea_queue: dict[str, Any],
) -> dict[str, Any]:
    phases = [
        _phase(
            "diagnose_ssh_publickey",
            _phase_status(
                ready=ssh_diagnosis["receipt_seen"] or ssh_local["receipt_seen"],
                blocked_status="blocked_waiting_ssh_publickey_diagnosis_receipt",
            ),
            "ssh_publickey_diagnosis",
        ),
        _phase(
            "preflight_control_path_and_harbor",
            _preflight_control_path_phase_status(
                ssh_diagnosis=ssh_diagnosis,
                watchdog_check=watchdog_check,
            ),
            "watchdog_check",
        ),
        _phase(
            "repair_ssh_metadata_if_check_confirms_metadata_drift",
            _ssh_metadata_phase_status(
                ssh_diagnosis=ssh_diagnosis,
                ssh_local=ssh_local,
                watchdog_check=watchdog_check,
            ),
            "ssh_local_repair",
        ),
        _phase(
            "repair_harbor_once_if_v2_still_502",
            _harbor_repair_once_phase_status(
                watchdog_check=watchdog_check,
                watchdog_repair=watchdog_repair,
            ),
            "watchdog_repair",
        ),
        _phase(
            "verify_controlled_cd_lane",
            _controlled_cd_lane_phase_status(
                controlled_cd_lane=controlled_cd_lane,
                verifier=verifier,
            ),
            "controlled_cd_lane_readiness",
        ),
        _phase(
            "verify_non110_runner_lane",
            _non110_runner_phase_status(
                non110_runner=non110_runner,
                verifier=verifier,
                gitea_queue=gitea_queue,
            ),
            "non110_runner_readiness",
        ),
    ]
    completed_statuses = {"ready", "skipped_not_required"}
    return {
        "phase_count": len(phases),
        "completed_phase_count": sum(
            1 for phase in phases if phase["status"] in completed_statuses
        ),
        "blocked_phase_count": sum(
            1 for phase in phases if phase["status"] not in completed_statuses
        ),
        "phase_ids": [phase["phase_id"] for phase in phases],
        "phases": phases,
        "metadata_only": True,
        "raw_output_returned": False,
    }


def _phase(phase_id: str, status: str, evidence_key: str) -> dict[str, Any]:
    return {
        "phase_id": phase_id,
        "status": status,
        "evidence_key": evidence_key,
        "raw_output_returned": False,
    }


def _phase_status(*, ready: bool, blocked_status: str) -> str:
    return "ready" if ready else blocked_status


def _preflight_control_path_phase_status(
    *,
    ssh_diagnosis: dict[str, Any],
    watchdog_check: dict[str, Any],
) -> str:
    if watchdog_check["receipt_seen"]:
        return "ready"
    if ssh_diagnosis["receipt_seen"] and ssh_diagnosis["node_high_load_seen"]:
        return "blocked_waiting_110_load_to_normalize_before_harbor_preflight"
    return "blocked_waiting_harbor_watchdog_check_receipt"


def _ssh_metadata_phase_status(
    *,
    ssh_diagnosis: dict[str, Any],
    ssh_local: dict[str, Any],
    watchdog_check: dict[str, Any],
) -> str:
    if ssh_local["receipt_seen"]:
        if ssh_local["control_channel_metadata_ready"]:
            return "ready"
        return "blocked_ssh_metadata_repair_receipt_not_ready"
    if ssh_diagnosis["wooo_command_path_ready"]:
        return "skipped_not_required"
    if ssh_diagnosis["publickey_offer_timeout_seen"]:
        return "blocked_waiting_ssh_metadata_repair_receipt_after_publickey_timeout"
    if ssh_diagnosis["server_accepts_key_then_timeout_seen"]:
        return "blocked_waiting_110_local_session_path_diagnosis_after_key_accept_timeout"
    if watchdog_check["receipt_seen"]:
        return "skipped_not_required"
    return "blocked_waiting_ssh_metadata_or_harbor_preflight_receipt"


def _harbor_repair_once_phase_status(
    *,
    watchdog_check: dict[str, Any],
    watchdog_repair: dict[str, Any],
) -> str:
    if watchdog_check["receipt_seen"] and watchdog_check["harbor_ready"]:
        return "skipped_not_required"
    if watchdog_repair["receipt_seen"] and watchdog_repair["harbor_ready"]:
        return "ready"
    if watchdog_repair["receipt_seen"]:
        return "blocked_harbor_repair_once_receipt_not_green"
    if watchdog_check["receipt_seen"]:
        return "blocked_waiting_harbor_repair_once_receipt"
    return "blocked_waiting_harbor_watchdog_check_receipt"


def _controlled_cd_lane_phase_status(
    *,
    controlled_cd_lane: dict[str, Any],
    verifier: dict[str, Any],
) -> str:
    if controlled_cd_lane["receipt_seen"]:
        if controlled_cd_lane["controlled_cd_lane_ready"]:
            return "ready"
        return "blocked_controlled_cd_lane_readiness_receipt_not_ready"
    return _phase_status(
        ready=verifier["registry_v2_ready"],
        blocked_status="blocked_waiting_controlled_cd_lane_readiness_receipt",
    )


def _non110_runner_phase_status(
    *,
    non110_runner: dict[str, Any],
    verifier: dict[str, Any],
    gitea_queue: dict[str, Any],
) -> str:
    if non110_runner["receipt_seen"]:
        if non110_runner["non110_runner_ready"]:
            return "ready"
        return "blocked_non110_runner_readiness_receipt_not_ready"
    if (
        gitea_queue["current_cd_waiting_for_runner_or_queue"]
        or gitea_queue["current_cd_no_matching_runner_label"]
        or gitea_queue["controlled_profile_no_matching_runner_label_count"] > 0
    ):
        return "blocked_waiting_non110_runner_readiness_receipt"
    if verifier["registry_v2_ready"]:
        return "skipped_not_required"
    return "blocked_waiting_non110_runner_or_registry_readiness_receipt"


def _parse_ssh_publickey_diagnosis_output(output: str) -> dict[str, Any]:
    fields = _parse_key_values(output)
    marker_seen = "AWOOOI_110_SSH_PUBLICKEY_AUTH_DIAGNOSIS" in output
    auth_attempts = _ssh_auth_attempts(output)
    command_path_attempts = _ssh_command_path_attempts(output)
    systemd_units = _systemd_unit_signals(output)
    wooo_publickey = _auth_classification(
        auth_attempts,
        user="wooo",
        mode="publickey",
    )
    wooo_command_path = _command_path_classification(
        command_path_attempts,
        user="wooo",
    )
    wooo_command_path_ready = wooo_command_path == "command_path_ready"
    publickey_offer_timeout_seen = any(
        item["mode"] == "publickey"
        and item["classification"] == "publickey_offer_timeout"
        for item in auth_attempts
    )
    server_accepts_key_then_timeout_seen = any(
        item["mode"] == "publickey"
        and item["classification"] == "server_accepts_key_then_timeout"
        for item in auth_attempts
    )
    preauth_timeout_count = sum(
        1 for item in auth_attempts if item["classification"] == "preauth_timeout"
    )
    permission_denied_count = sum(
        1 for item in auth_attempts if item["classification"] == "permission_denied"
    )
    runner_systemctl_show_timeout_seen = any(
        item["unit"].startswith("actions.runner")
        and item["classifier"] == "systemctl_show_timeout"
        for item in systemd_units
    )
    publickey_offer_timeout_effective_seen = bool(
        publickey_offer_timeout_seen and not wooo_command_path_ready
    )
    server_accepts_key_then_timeout_effective_seen = bool(
        server_accepts_key_then_timeout_seen and not wooo_command_path_ready
    )
    runner_systemctl_show_timeout_effective_seen = bool(
        runner_systemctl_show_timeout_seen and not wooo_command_path_ready
    )
    node_load_classifier = str(fields.get("NODE_LOAD_CLASSIFIER") or "")
    node_high_load_seen = node_load_classifier == "high_load"
    ssh_port_tcp_open = str(fields.get("SSH_PORT") or "") == "tcp_open"
    node_exporter_ok = str(fields.get("NODE_EXPORTER") or "") == "ok"
    return {
        "receipt_seen": marker_seen,
        "target": str(fields.get("TARGET") or ""),
        "node_exporter_ok": node_exporter_ok,
        "node_load_classifier": node_load_classifier,
        "node_load1_per_cpu": _float_or_none(fields.get("NODE_LOAD1_PER_CPU")),
        "node_high_load_seen": node_high_load_seen,
        "ssh_port_tcp_open": ssh_port_tcp_open,
        "ssh_banner_seen": "SSH_BANNER=SSH-" in output,
        "auth_attempt_count": len(auth_attempts),
        "auth_classifications": auth_attempts,
        "command_path_attempt_count": len(command_path_attempts),
        "command_path_classifications": command_path_attempts,
        "systemd_unit_signal_count": len(systemd_units),
        "systemd_unit_signals": systemd_units,
        "runner_systemctl_show_timeout_seen": runner_systemctl_show_timeout_seen,
        "runner_systemctl_show_timeout_effective_seen": (
            runner_systemctl_show_timeout_effective_seen
        ),
        "wooo_publickey_classification": wooo_publickey,
        "wooo_command_path_classification": wooo_command_path,
        "wooo_command_path_ready": wooo_command_path_ready,
        "publickey_offer_timeout_seen": publickey_offer_timeout_seen,
        "publickey_offer_timeout_effective_seen": (
            publickey_offer_timeout_effective_seen
        ),
        "server_accepts_key_then_timeout_seen": (
            server_accepts_key_then_timeout_seen
        ),
        "server_accepts_key_then_timeout_effective_seen": (
            server_accepts_key_then_timeout_effective_seen
        ),
        "preauth_timeout_count": preauth_timeout_count,
        "permission_denied_count": permission_denied_count,
        "diagnosis_ready": bool(
            marker_seen
            and (ssh_port_tcp_open or wooo_command_path_ready)
            and (auth_attempts or command_path_attempts)
        ),
        "metadata_only": True,
        "raw_output_returned": False,
    }


def _ssh_auth_attempts(output: str) -> list[dict[str, Any]]:
    attempts: list[dict[str, Any]] = []
    for line in output.splitlines():
        if not line.startswith("SSH_AUTH "):
            continue
        fields = _parse_key_values(line)
        attempts.append(
            {
                "user": str(fields.get("user") or ""),
                "mode": str(fields.get("mode") or ""),
                "rc": _int_or_none(fields.get("rc")),
                "classification": str(fields.get("classification") or ""),
            }
        )
    return attempts


def _ssh_command_path_attempts(output: str) -> list[dict[str, Any]]:
    attempts: list[dict[str, Any]] = []
    for line in output.splitlines():
        if not line.startswith("SSH_COMMAND_PATH "):
            continue
        fields = _parse_key_values(line)
        attempts.append(
            {
                "user": str(fields.get("user") or ""),
                "rc": _int_or_none(fields.get("rc")),
                "classification": str(fields.get("classification") or ""),
                "marker_seen": _bool_from_field(fields.get("marker_seen")),
                "remote_user_match": _bool_from_field(
                    fields.get("remote_user_match")
                ),
            }
        )
    return attempts


def _systemd_unit_signals(output: str) -> list[dict[str, str]]:
    signals: list[dict[str, str]] = []
    for line in output.splitlines():
        if not line.startswith("SYSTEMD_UNIT "):
            continue
        fields = _parse_key_values(line)
        signals.append(
            {
                "unit": str(fields.get("unit") or ""),
                "active_state": str(fields.get("active_state") or ""),
                "classifier": str(fields.get("classifier") or ""),
            }
        )
    return signals


def _auth_classification(
    attempts: list[dict[str, Any]],
    *,
    user: str,
    mode: str,
) -> str:
    for item in attempts:
        if item["user"] == user and item["mode"] == mode:
            return str(item["classification"] or "")
    return ""


def _command_path_classification(
    attempts: list[dict[str, Any]],
    *,
    user: str,
) -> str:
    for item in attempts:
        if item["user"] == user:
            return str(item["classification"] or "")
    return ""


def _parse_ssh_local_repair_output(output: str) -> dict[str, Any]:
    fields = _parse_key_values(output)
    marker_seen = "AWOOOI_110_SSH_PUBLICKEY_AUTH_LOCAL_REPAIR" in output
    sshd_unverified = "SSHD_CONFIG_SYNTAX=unverified_requires_root" in output
    sshd_after_unverified = (
        "SSHD_CONFIG_SYNTAX_AFTER_APPLY=unverified_requires_root" in output
    )
    sshd_ok = "SSHD_CONFIG_SYNTAX=ok" in output or sshd_unverified
    sshd_after_ok = (
        "SSHD_CONFIG_SYNTAX_AFTER_APPLY=ok" in output
        or sshd_after_unverified
        or not _bool_from_field(fields.get("APPLY"))
    )
    user_status = _first_key_value_line(output, prefix="USER_STATUS ")
    authorized_keys_status = _first_key_value_line(
        output,
        prefix="AUTHORIZED_KEYS_STATUS ",
    )
    user_exists = _bool_from_field(user_status.get("exists"))
    authorized_keys_exists = _bool_from_field(authorized_keys_status.get("exists"))
    account_locked = _bool_from_field(fields.get("account_locked"))
    shell_executable = _bool_from_field(fields.get("shell_executable"))
    sshd_effective_config_available = _bool_from_field(fields.get("available"))
    pubkey_authentication = str(fields.get("pubkeyauthentication") or "")
    password_authentication = str(fields.get("passwordauthentication") or "")
    kbdinteractive_authentication = str(
        fields.get("kbdinteractiveauthentication") or ""
    )
    usepam = str(fields.get("usepam") or "")
    maxstartups = str(fields.get("maxstartups") or "")
    authorized_keys_file_default = _bool_from_field(
        fields.get("authorized_keys_file_default")
    )
    account_metadata_ready = bool(
        user_exists
        and not account_locked
        and shell_executable
        and pubkey_authentication == "yes"
        and authorized_keys_file_default
    )
    permissions_applied = "APPLIED permissions" in output
    reload_done = "SSH_RELOAD=done" in output
    reload_skipped = "SSH_RELOAD=skipped" in output
    return {
        "receipt_seen": marker_seen,
        "mode": _mode_from_marker_line(output),
        "sshd_config_syntax_ok": sshd_ok,
        "sshd_config_syntax_unverified_requires_root": sshd_unverified,
        "sshd_config_syntax_after_apply_ok": sshd_after_ok,
        "sshd_config_syntax_after_apply_unverified_requires_root": (
            sshd_after_unverified
        ),
        "target_user_exists": user_exists,
        "target_user_account_locked": account_locked,
        "target_user_shell_executable": shell_executable,
        "sshd_effective_config_available": sshd_effective_config_available,
        "sshd_pubkeyauthentication": pubkey_authentication,
        "sshd_passwordauthentication": password_authentication,
        "sshd_kbdinteractiveauthentication": kbdinteractive_authentication,
        "sshd_usepam": usepam,
        "sshd_maxstartups": maxstartups,
        "sshd_authorized_keys_file_default": authorized_keys_file_default,
        "account_metadata_ready": account_metadata_ready,
        "authorized_keys_metadata_present": authorized_keys_exists,
        "permissions_applied": permissions_applied,
        "ssh_reload_done": reload_done,
        "ssh_reload_skipped": reload_skipped,
        "control_channel_metadata_ready": bool(
            marker_seen
            and sshd_ok
            and sshd_after_ok
            and user_exists
            and authorized_keys_exists
            and account_metadata_ready
        ),
    }


def _first_key_value_line(output: str, *, prefix: str) -> dict[str, str]:
    for line in output.splitlines():
        if line.startswith(prefix):
            return _parse_key_values(line)
    return {}


def _parse_watchdog_output(output: str) -> dict[str, Any]:
    fields = _parse_key_values(output)
    marker_seen = "AWOOOI_HARBOR_WATCHDOG_CHECK" in output
    harbor_ready = _bool_from_field(fields.get("harbor_ready"))
    local_status = _int_or_none(fields.get("harbor_local_v2_http_status"))
    docker_action = _bool_from_field(fields.get("docker_compose_action_performed"))
    service_restart = _bool_from_field(fields.get("service_restart_performed"))
    host_reboot = _bool_from_field(fields.get("host_reboot_performed"))
    expected_host = _bool_from_field(fields.get("expected_host_ip_present"))
    repair_attempt_seen = (
        "Harbor Watchdog" in output
        or "Harbor 修復" in output
        or "docker compose up -d" in output
    )
    return {
        "receipt_seen": marker_seen,
        "mode": str(fields.get("mode") or ""),
        "expected_host_ip_present": expected_host,
        "harbor_local_v2_http_status": local_status,
        "harbor_ready": harbor_ready,
        "check_only": _bool_from_field(fields.get("check_only")),
        "repair_attempt_seen": repair_attempt_seen,
        "docker_compose_action_performed": docker_action,
        "service_restart_performed": service_restart,
        "host_reboot_performed": host_reboot,
        "forbidden_action_seen": bool(service_restart or host_reboot),
    }


def _parse_controlled_cd_lane_readiness_output(output: str) -> dict[str, Any]:
    fields = _parse_key_values(output)
    marker_seen = "AWOOOI_110_CONTROLLED_CD_LANE_READY=" in output
    blockers = _prefixed_blockers(
        output,
        prefix="controlled_cd_lane_readiness:",
    )
    warning_count = _int_or_none(fields.get("WARNING_COUNT")) or 0
    blocker_count = _int_or_none(fields.get("BLOCKER_COUNT"))
    if blocker_count is None:
        blocker_count = len(blockers)
    ready = _bool_from_field(fields.get("AWOOOI_110_CONTROLLED_CD_LANE_READY"))
    return {
        "receipt_seen": marker_seen,
        "controlled_cd_lane_ready": bool(
            marker_seen and ready and blocker_count == 0 and not blockers
        ),
        "config_ready": _bool_from_field(fields.get("CONFIG_READY")),
        "binary_ready": _bool_from_field(fields.get("BINARY_READY")),
        "registration_ready": _bool_from_field(fields.get("REGISTRATION_READY")),
        "service_ready": _bool_from_field(fields.get("SERVICE_READY")),
        "legacy_failclosed": _bool_from_field(fields.get("LEGACY_FAILCLOSED")),
        "primary_lane_failclosed": _bool_from_field(
            fields.get("PRIMARY_LANE_FAILCLOSED")
        ),
        "warning_count": warning_count,
        "blocker_count": blocker_count,
        "blockers": blockers,
        "safe_next_step": str(fields.get("safe_next_step") or ""),
        "metadata_only": True,
        "raw_output_returned": False,
    }


def _parse_non110_runner_readiness_output(output: str) -> dict[str, Any]:
    fields = _parse_key_values(output)
    marker_seen = "AWOOOI_NON110_RUNNER_READY=" in output
    if not marker_seen:
        return {
            "receipt_seen": False,
            "non110_runner_ready": False,
            "ready_config_count": 0,
            "ready_binary_count": 0,
            "ready_registration_count": 0,
            "ready_service_count": 0,
            "ready_active_service_count": 0,
            "ready_autostart_path_count": 0,
            "warning_count": 0,
            "blocker_count": 0,
            "blockers": [],
            "safe_next_step": "",
            "raw_runner_registration_read": False,
            "runner_token_read": False,
            "metadata_only": True,
            "raw_output_returned": False,
        }
    blockers = _prefixed_blockers(
        output,
        prefix="non110_runner_readiness:",
    )
    warning_count = _int_or_none(fields.get("WARNING_COUNT")) or 0
    blocker_count = _int_or_none(fields.get("BLOCKER_COUNT"))
    if blocker_count is None:
        blocker_count = len(blockers)
    ready = _bool_from_field(fields.get("AWOOOI_NON110_RUNNER_READY"))
    raw_runner_registration_read = _bool_from_field(
        fields.get("raw_runner_registration_read")
    )
    runner_token_read = _bool_from_field(fields.get("runner_token_read"))
    boundary_blockers = []
    if raw_runner_registration_read:
        boundary_blockers.append(
            "non110_runner_readiness:raw_runner_registration_read"
        )
    if runner_token_read:
        boundary_blockers.append("non110_runner_readiness:runner_token_read")
    blockers = _unique_strings(blockers + boundary_blockers)
    if boundary_blockers and blocker_count == 0:
        blocker_count = len(blockers)
    return {
        "receipt_seen": marker_seen,
        "non110_runner_ready": bool(
            marker_seen
            and ready
            and blocker_count == 0
            and not blockers
            and not raw_runner_registration_read
            and not runner_token_read
        ),
        "ready_config_count": _int_or_none(fields.get("READY_CONFIG_COUNT")) or 0,
        "ready_binary_count": _int_or_none(fields.get("READY_BINARY_COUNT")) or 0,
        "ready_registration_count": (
            _int_or_none(fields.get("READY_REGISTRATION_COUNT")) or 0
        ),
        "ready_service_count": _int_or_none(fields.get("READY_SERVICE_COUNT")) or 0,
        "ready_active_service_count": (
            _int_or_none(fields.get("READY_ACTIVE_SERVICE_COUNT")) or 0
        ),
        "ready_autostart_path_count": (
            _int_or_none(fields.get("READY_AUTOSTART_PATH_COUNT")) or 0
        ),
        "warning_count": warning_count,
        "blocker_count": blocker_count,
        "blockers": blockers,
        "safe_next_step": str(fields.get("safe_next_step") or ""),
        "raw_runner_registration_read": raw_runner_registration_read,
        "runner_token_read": runner_token_read,
        "metadata_only": True,
        "raw_output_returned": False,
    }


def _post_apply_verifier(receipt_payload: dict[str, Any]) -> dict[str, Any]:
    public_status = _int_or_none(receipt_payload.get("public_registry_v2_http_status"))
    internal_status = _int_or_none(
        receipt_payload.get("internal_registry_v2_http_status")
    )
    public_ready = public_status in _READY_HTTP_STATUSES
    internal_ready = internal_status in _READY_HTTP_STATUSES
    return {
        "public_registry_v2_http_status": public_status,
        "internal_registry_v2_http_status": internal_status,
        "public_registry_v2_ready": public_ready,
        "internal_registry_v2_ready": internal_ready,
        "registry_v2_ready": public_ready and internal_ready,
    }


def _gitea_queue_readback(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        return {
            "receipt_seen": False,
            "schema_version": "",
            "status": "not_provided",
            "latest_visible_cd_run_id": "",
            "latest_visible_cd_run_status": "",
            "latest_visible_cd_run_waiting": False,
            "latest_visible_cd_run_commit_sha": "",
            "current_cd_no_matching_runner_label": "",
            "current_cd_waiting_for_runner_or_queue": False,
            "current_cd_inflight_classifier": "",
            "current_cd_harbor_latest_registry_v2_status": "",
            "current_cd_harbor_retrying_unavailable": False,
            "controlled_profile_no_matching_runner_labels": {},
            "controlled_profile_no_matching_runner_label_count": 0,
            "current_cd_harbor_repair_requires_110_controlled_lane": False,
            "current_cd_harbor_repair_blocked_by_no_matching_awoooi_host": False,
            "current_cd_harbor_repair_lane_classifier": "",
            "cd_run_jobs_stale_or_mismatched": False,
            "cd_run_jobs_payload_classifier": "",
            "cd_run_jobs_head_sha_mismatch": False,
            "cd_run_jobs_run_id_mismatch": False,
            "cd_run_jobs_expected_run_id": "",
            "cd_run_jobs_expected_head_sha": "",
            "cd_run_jobs_head_shas": [],
            "cd_run_jobs_run_ids": [],
            "latest_visible_harbor_110_repair_run_id": "",
            "latest_visible_harbor_110_repair_run_status": "",
            "harbor_110_repair_failure_classifier": "",
            "harbor_110_repair_remote_control_channel_unavailable": False,
            "harbor_110_repair_remote_ssh_reachable": None,
            "harbor_110_repair_bounded_ssh_timeout_seen": False,
            "harbor_110_repair_local_registry_v2_status": "",
            "harbor_110_repair_public_registry_v2_status": "",
            "harbor_110_repair_verified": False,
            "harbor_110_repair_no_matching_runner_label": "",
            "harbor_110_repair_no_matching_runner": False,
            "harbor_110_repair_waiting": False,
            "harbor_110_repair_blocked": False,
            "harbor_110_repair_jobs_stale_or_mismatched": False,
            "harbor_110_repair_jobs_cross_workflow_mismatch": False,
            "harbor_110_repair_jobs_payload_classifier": "",
            "harbor_110_repair_visible_running_jobs_api_stale": False,
            "current_cd_waiting_behind_harbor_110_repair_running": False,
            "harbor_110_repair_jobs_expected_names": [],
            "harbor_110_repair_jobs_unexpected_names": [],
            "harbor_110_repair_jobs_labels": [],
            "harbor_110_repair_jobs_runner_names": [],
            "normalized_classifier_fields": [],
            "normalized_classifier_field_count": 0,
            "normalized_classifier_field_ids": [],
            "operation_boundary_violation": False,
            "blockers": [],
            "blocker_count": 0,
            "metadata_only": True,
            "raw_output_returned": False,
            "top_visible_runs_returned": False,
        }

    readback = _dict(value.get("readback"))
    rollups = _dict(value.get("rollups"))
    operation_boundaries = _dict(value.get("operation_boundaries"))
    no_matching_label = str(
        rollups.get("harbor_110_repair_no_matching_runner_label")
        or readback.get("latest_visible_harbor_110_repair_no_matching_runner_label")
        or ""
    )
    jobs_stale = bool(
        rollups.get("harbor_110_repair_jobs_stale_or_mismatched") is True
        or readback.get("harbor_110_repair_jobs_stale_or_mismatched") is True
    )
    jobs_cross_workflow = bool(
        rollups.get("harbor_110_repair_jobs_cross_workflow_mismatch") is True
        or readback.get("harbor_110_repair_jobs_cross_workflow_mismatch") is True
    )
    waiting = bool(
        rollups.get("harbor_110_repair_waiting") is True
        or readback.get("latest_visible_harbor_110_repair_waiting") is True
    )
    blocked = bool(
        rollups.get("harbor_110_repair_blocked") is True
        or readback.get("latest_visible_harbor_110_repair_blocked") is True
    )
    harbor_repair_failure_classifier = str(
        rollups.get("harbor_110_repair_failure_classifier")
        or readback.get("latest_visible_harbor_110_repair_failure_classifier")
        or ""
    )
    harbor_repair_remote_control_unavailable = bool(
        rollups.get("harbor_110_repair_remote_control_channel_unavailable") is True
        or readback.get(
            "latest_visible_harbor_110_repair_remote_control_channel_unavailable"
        )
        is True
        or harbor_repair_failure_classifier
        == "harbor_110_remote_control_channel_unavailable"
    )
    harbor_repair_bounded_ssh_timeout_seen = bool(
        rollups.get("harbor_110_repair_bounded_ssh_timeout_seen") is True
        or readback.get("latest_visible_harbor_110_repair_bounded_ssh_timeout_seen")
        is True
    )
    harbor_repair_remote_ssh_reachable = (
        readback.get("latest_visible_harbor_110_repair_remote_ssh_reachable")
        if "latest_visible_harbor_110_repair_remote_ssh_reachable" in readback
        else rollups.get("harbor_110_repair_remote_ssh_reachable")
    )
    harbor_repair_local_registry_status = str(
        rollups.get("harbor_110_repair_local_registry_v2_status")
        or readback.get("latest_visible_harbor_110_repair_local_registry_v2_status")
        or ""
    )
    harbor_repair_public_registry_status = str(
        rollups.get("harbor_110_repair_public_registry_v2_status")
        or readback.get("latest_visible_harbor_110_repair_public_registry_v2_status")
        or ""
    )
    harbor_repair_verified = bool(
        rollups.get("harbor_110_repair_verified") is True
        or readback.get("latest_visible_harbor_110_repair_verified") is True
    )
    harbor_repair_local_v2_unavailable = (
        harbor_repair_failure_classifier
        == "harbor_110_remote_local_registry_v2_unavailable"
    )
    harbor_repair_public_v2_unavailable = (
        harbor_repair_failure_classifier == "harbor_public_registry_v2_unavailable"
    )
    current_cd_inflight_classifier = str(
        rollups.get("current_main_cd_inflight_classifier")
        or readback.get("latest_visible_cd_inflight_classifier")
        or ""
    )
    current_cd_waiting = bool(
        rollups.get("current_main_cd_run_waiting") is True
        or readback.get("latest_visible_cd_run_waiting") is True
    )
    current_cd_no_matching_label = str(
        rollups.get("current_main_cd_no_matching_runner_label")
        or readback.get("latest_visible_cd_no_matching_runner_label")
        or ""
    )
    controlled_profile_no_matching_labels = {
        str(key): str(label)
        for key, label in _dict(
            readback.get("controlled_profile_no_matching_runner_labels")
            or rollups.get("controlled_profile_no_matching_runner_labels")
        ).items()
        if label
    }
    current_cd_latest_registry_status = str(
        rollups.get("current_main_cd_harbor_latest_registry_v2_status")
        or readback.get("latest_visible_cd_harbor_latest_registry_v2_status")
        or ""
    )
    current_cd_harbor_retrying = bool(
        rollups.get("current_main_cd_harbor_public_route_retrying_unavailable")
        is True
        or readback.get("latest_visible_cd_harbor_public_route_retrying_unavailable")
        is True
        or current_cd_inflight_classifier
        == "harbor_registry_public_route_unavailable_pending_retry"
    )
    current_cd_harbor_repair_requires_110_lane = bool(
        rollups.get("current_main_cd_harbor_repair_requires_110_controlled_lane")
        is True
        or readback.get("latest_visible_cd_harbor_repair_requires_110_controlled_lane")
        is True
    )
    current_cd_harbor_repair_blocked_by_awoooi_host = bool(
        rollups.get("current_main_cd_harbor_repair_blocked_by_no_matching_awoooi_host")
        is True
        or readback.get(
            "latest_visible_cd_harbor_repair_blocked_by_no_matching_awoooi_host"
        )
        is True
    )
    current_cd_harbor_repair_lane_classifier = str(
        rollups.get("current_main_cd_harbor_repair_lane_classifier")
        or readback.get("latest_visible_cd_harbor_repair_lane_classifier")
        or ""
    )
    controlled_profile_no_matching = bool(controlled_profile_no_matching_labels)
    cd_jobs_stale = bool(
        rollups.get("cd_run_jobs_stale_or_mismatched") is True
        or readback.get("cd_run_jobs_stale_or_mismatched") is True
    )
    cd_jobs_head_sha_mismatch = bool(
        rollups.get("cd_run_jobs_head_sha_mismatch") is True
        or readback.get("cd_run_jobs_head_sha_mismatch") is True
    )
    cd_jobs_run_id_mismatch = bool(
        rollups.get("cd_run_jobs_run_id_mismatch") is True
        or readback.get("cd_run_jobs_run_id_mismatch") is True
    )
    cd_jobs_payload_classifier = str(
        rollups.get("cd_run_jobs_payload_classifier")
        or readback.get("cd_run_jobs_payload_classifier")
        or ""
    )
    harbor_jobs_payload_classifier = str(
        rollups.get("harbor_110_repair_jobs_payload_classifier")
        or readback.get("harbor_110_repair_jobs_payload_classifier")
        or ""
    )
    harbor_running_jobs_api_stale = bool(
        rollups.get("harbor_110_repair_visible_running_jobs_api_stale") is True
        or readback.get("harbor_110_repair_visible_running_jobs_api_stale") is True
    )
    current_cd_waiting_behind_harbor_running = bool(
        rollups.get("current_cd_waiting_behind_harbor_110_repair_running") is True
        or readback.get("current_cd_waiting_behind_harbor_110_repair_running") is True
    )
    boundary_violation = any(
        operation_boundaries.get(flag) is True
        for flag in (
            "github_api_used",
            "host_write_performed",
            "runner_registration_performed",
            "runner_service_start_performed",
            "secret_or_runner_token_read",
            "workflow_dispatch_performed",
        )
    )
    blockers = _gitea_queue_blockers(
        no_matching_runner=bool(no_matching_label),
        jobs_stale=jobs_stale,
        jobs_cross_workflow=jobs_cross_workflow,
        current_cd_harbor_retrying=current_cd_harbor_retrying,
        current_cd_waiting=current_cd_waiting,
        current_cd_no_matching_runner=bool(current_cd_no_matching_label),
        current_cd_harbor_repair_requires_110_lane=(
            current_cd_harbor_repair_requires_110_lane
        ),
        current_cd_harbor_repair_blocked_by_awoooi_host=(
            current_cd_harbor_repair_blocked_by_awoooi_host
        ),
        controlled_profile_no_matching=controlled_profile_no_matching,
        harbor_running_jobs_api_stale=harbor_running_jobs_api_stale,
        current_cd_waiting_behind_harbor_running=(
            current_cd_waiting_behind_harbor_running
        ),
        cd_jobs_stale=cd_jobs_stale,
        cd_jobs_head_sha_mismatch=cd_jobs_head_sha_mismatch,
        cd_jobs_run_id_mismatch=cd_jobs_run_id_mismatch,
        harbor_repair_remote_control_unavailable=(
            harbor_repair_remote_control_unavailable
        ),
        harbor_repair_local_v2_unavailable=harbor_repair_local_v2_unavailable,
        harbor_repair_public_v2_unavailable=harbor_repair_public_v2_unavailable,
        blocked=blocked,
        boundary_violation=boundary_violation,
    )
    normalized_classifier_fields = _gitea_queue_normalized_classifier_fields(
        cd_jobs_payload_classifier=cd_jobs_payload_classifier,
        harbor_jobs_payload_classifier=harbor_jobs_payload_classifier,
        harbor_repair_failure_classifier=harbor_repair_failure_classifier,
        no_matching_label=no_matching_label,
        current_cd_no_matching_label=current_cd_no_matching_label,
        current_cd_waiting=current_cd_waiting,
        controlled_profile_no_matching_labels=controlled_profile_no_matching_labels,
        harbor_running_jobs_api_stale=harbor_running_jobs_api_stale,
        current_cd_waiting_behind_harbor_running=(
            current_cd_waiting_behind_harbor_running
        ),
        blockers=blockers,
    )
    return {
        "receipt_seen": True,
        "schema_version": str(value.get("schema_version") or ""),
        "status": str(value.get("status") or "unknown"),
        "latest_visible_cd_run_id": str(readback.get("latest_visible_cd_run_id") or ""),
        "latest_visible_cd_run_status": str(
            readback.get("latest_visible_cd_run_status") or ""
        ),
        "latest_visible_cd_run_waiting": current_cd_waiting,
        "latest_visible_cd_run_commit_sha": str(
            readback.get("latest_visible_cd_run_commit_sha") or ""
        ),
        "current_cd_no_matching_runner_label": current_cd_no_matching_label,
        "current_cd_waiting_for_runner_or_queue": current_cd_waiting,
        "current_cd_inflight_classifier": current_cd_inflight_classifier,
        "current_cd_harbor_latest_registry_v2_status": (
            current_cd_latest_registry_status
        ),
        "current_cd_harbor_retrying_unavailable": current_cd_harbor_retrying,
        "controlled_profile_no_matching_runner_labels": (
            controlled_profile_no_matching_labels
        ),
        "controlled_profile_no_matching_runner_label_count": len(
            controlled_profile_no_matching_labels
        ),
        "current_cd_harbor_repair_requires_110_controlled_lane": (
            current_cd_harbor_repair_requires_110_lane
        ),
        "current_cd_harbor_repair_blocked_by_no_matching_awoooi_host": (
            current_cd_harbor_repair_blocked_by_awoooi_host
        ),
        "current_cd_harbor_repair_lane_classifier": (
            current_cd_harbor_repair_lane_classifier
        ),
        "cd_run_jobs_stale_or_mismatched": cd_jobs_stale,
        "cd_run_jobs_payload_classifier": cd_jobs_payload_classifier,
        "cd_run_jobs_head_sha_mismatch": cd_jobs_head_sha_mismatch,
        "cd_run_jobs_run_id_mismatch": cd_jobs_run_id_mismatch,
        "cd_run_jobs_expected_run_id": str(
            readback.get("cd_run_jobs_expected_run_id") or ""
        ),
        "cd_run_jobs_expected_head_sha": str(
            readback.get("cd_run_jobs_expected_head_sha") or ""
        ),
        "cd_run_jobs_head_shas": _strings(readback.get("cd_run_jobs_head_shas")),
        "cd_run_jobs_run_ids": _strings(readback.get("cd_run_jobs_run_ids")),
        "latest_visible_harbor_110_repair_run_id": str(
            readback.get("latest_visible_harbor_110_repair_run_id") or ""
        ),
        "latest_visible_harbor_110_repair_run_status": str(
            readback.get("latest_visible_harbor_110_repair_run_status") or ""
        ),
        "harbor_110_repair_failure_classifier": harbor_repair_failure_classifier,
        "harbor_110_repair_remote_control_channel_unavailable": (
            harbor_repair_remote_control_unavailable
        ),
        "harbor_110_repair_remote_ssh_reachable": (
            harbor_repair_remote_ssh_reachable
        ),
        "harbor_110_repair_bounded_ssh_timeout_seen": (
            harbor_repair_bounded_ssh_timeout_seen
        ),
        "harbor_110_repair_local_registry_v2_status": (
            harbor_repair_local_registry_status
        ),
        "harbor_110_repair_public_registry_v2_status": (
            harbor_repair_public_registry_status
        ),
        "harbor_110_repair_verified": harbor_repair_verified,
        "harbor_110_repair_no_matching_runner_label": no_matching_label,
        "harbor_110_repair_no_matching_runner": bool(no_matching_label),
        "harbor_110_repair_waiting": waiting,
        "harbor_110_repair_blocked": blocked,
        "harbor_110_repair_jobs_stale_or_mismatched": jobs_stale,
        "harbor_110_repair_jobs_cross_workflow_mismatch": jobs_cross_workflow,
        "harbor_110_repair_jobs_payload_classifier": harbor_jobs_payload_classifier,
        "harbor_110_repair_visible_running_jobs_api_stale": (
            harbor_running_jobs_api_stale
        ),
        "current_cd_waiting_behind_harbor_110_repair_running": (
            current_cd_waiting_behind_harbor_running
        ),
        "harbor_110_repair_jobs_expected_names": _strings(
            readback.get("harbor_110_repair_jobs_expected_names")
        ),
        "harbor_110_repair_jobs_unexpected_names": _strings(
            readback.get("harbor_110_repair_jobs_unexpected_names")
        ),
        "harbor_110_repair_jobs_labels": _strings(
            readback.get("harbor_110_repair_jobs_labels")
        ),
        "harbor_110_repair_jobs_runner_names": _strings(
            readback.get("harbor_110_repair_jobs_runner_names")
        ),
        "normalized_classifier_fields": normalized_classifier_fields,
        "normalized_classifier_field_count": len(normalized_classifier_fields),
        "normalized_classifier_field_ids": [
            item["field_id"] for item in normalized_classifier_fields
        ],
        "operation_boundary_violation": boundary_violation,
        "blockers": blockers,
        "blocker_count": len(blockers),
        "metadata_only": True,
        "raw_output_returned": False,
        "top_visible_runs_returned": False,
    }


def _gitea_queue_normalized_classifier_fields(
    *,
    cd_jobs_payload_classifier: str,
    harbor_jobs_payload_classifier: str,
    harbor_repair_failure_classifier: str,
    no_matching_label: str,
    current_cd_no_matching_label: str,
    current_cd_waiting: bool,
    controlled_profile_no_matching_labels: dict[str, str],
    harbor_running_jobs_api_stale: bool,
    current_cd_waiting_behind_harbor_running: bool,
    blockers: list[str],
) -> list[dict[str, Any]]:
    fields = [
        {
            "field_id": "cd_run_jobs_payload_classifier",
            "value": cd_jobs_payload_classifier,
            "blockers": [
                item for item in blockers if item.startswith("gitea_queue_cd_jobs_")
            ],
            "metadata_only": True,
            "raw_output_returned": False,
        },
        {
            "field_id": "harbor_110_repair_jobs_payload_classifier",
            "value": harbor_jobs_payload_classifier,
            "blockers": [
                item
                for item in blockers
                if item.startswith("gitea_queue_harbor_110_repair_jobs_")
            ],
            "metadata_only": True,
            "raw_output_returned": False,
        },
        {
            "field_id": "latest_visible_harbor_110_repair_no_matching_runner_label",
            "value": no_matching_label,
            "blockers": [
                item
                for item in blockers
                if item == "gitea_queue_harbor_110_repair_no_matching_runner"
            ],
            "metadata_only": True,
            "raw_output_returned": False,
        },
    ]
    if harbor_repair_failure_classifier:
        fields.append(
            {
                "field_id": "harbor_110_repair_failure_classifier",
                "value": harbor_repair_failure_classifier,
                "blockers": [
                    item
                    for item in blockers
                    if item
                    in {
                        "gitea_queue_harbor_110_remote_control_channel_unavailable",
                        (
                            "gitea_queue_harbor_110_remote_local_registry_v2_"
                            "unavailable"
                        ),
                        (
                            "gitea_queue_harbor_public_registry_v2_unavailable_"
                            "after_remote_repair"
                        ),
                    }
                ],
                "metadata_only": True,
                "raw_output_returned": False,
            }
        )
    if current_cd_no_matching_label or current_cd_waiting:
        fields.append(
            {
                "field_id": "current_cd_workflow_runner_readiness",
                "value": current_cd_no_matching_label,
                "waiting": current_cd_waiting,
                "blockers": [
                    item
                    for item in blockers
                    if item.startswith("gitea_queue_current_cd_")
                ],
                "metadata_only": True,
                "raw_output_returned": False,
            }
        )
    if harbor_running_jobs_api_stale:
        fields.append(
            {
                "field_id": "harbor_110_repair_visible_running_jobs_api_stale",
                "value": True,
                "blockers": [
                    item
                    for item in blockers
                    if item
                    == "gitea_queue_harbor_110_repair_visible_running_jobs_api_stale"
                ],
                "metadata_only": True,
                "raw_output_returned": False,
            }
        )
    if current_cd_waiting_behind_harbor_running:
        fields.append(
            {
                "field_id": "current_cd_waiting_behind_harbor_110_repair_running",
                "value": True,
                "blockers": [
                    item
                    for item in blockers
                    if item
                    == (
                        "gitea_queue_current_cd_waiting_behind_harbor_110_"
                        "repair_running"
                    )
                ],
                "metadata_only": True,
                "raw_output_returned": False,
            }
        )
    if controlled_profile_no_matching_labels:
        fields.append(
            {
                "field_id": "controlled_profile_no_matching_runner_labels",
                "value": controlled_profile_no_matching_labels,
                "blockers": [
                    item
                    for item in blockers
                    if item == "gitea_queue_controlled_profile_no_matching_runner_labels"
                ],
                "metadata_only": True,
                "raw_output_returned": False,
            }
        )
    return fields


def _gitea_queue_blockers(
    *,
    no_matching_runner: bool,
    jobs_stale: bool,
    jobs_cross_workflow: bool,
    current_cd_harbor_retrying: bool,
    current_cd_waiting: bool,
    current_cd_no_matching_runner: bool,
    current_cd_harbor_repair_requires_110_lane: bool,
    current_cd_harbor_repair_blocked_by_awoooi_host: bool,
    controlled_profile_no_matching: bool,
    harbor_running_jobs_api_stale: bool,
    current_cd_waiting_behind_harbor_running: bool,
    cd_jobs_stale: bool,
    cd_jobs_head_sha_mismatch: bool,
    cd_jobs_run_id_mismatch: bool,
    harbor_repair_remote_control_unavailable: bool,
    harbor_repair_local_v2_unavailable: bool,
    harbor_repair_public_v2_unavailable: bool,
    blocked: bool,
    boundary_violation: bool,
) -> list[str]:
    blockers: list[str] = []
    if current_cd_harbor_repair_blocked_by_awoooi_host:
        blockers.append(
            "gitea_queue_cd_harbor_repair_blocked_by_no_matching_awoooi_host"
        )
    if current_cd_harbor_repair_requires_110_lane:
        blockers.append("gitea_queue_cd_harbor_repair_requires_110_controlled_lane")
    if current_cd_harbor_retrying:
        blockers.append("gitea_queue_current_cd_harbor_retrying_unavailable")
    if current_cd_no_matching_runner:
        blockers.append("gitea_queue_current_cd_no_matching_runner")
    if current_cd_waiting:
        blockers.append("gitea_queue_current_cd_waiting_for_runner_or_queue")
    if controlled_profile_no_matching:
        blockers.append("gitea_queue_controlled_profile_no_matching_runner_labels")
    if cd_jobs_head_sha_mismatch:
        blockers.append("gitea_queue_cd_jobs_head_sha_mismatch")
    if cd_jobs_run_id_mismatch:
        blockers.append("gitea_queue_cd_jobs_run_id_mismatch")
    if cd_jobs_stale:
        blockers.append("gitea_queue_cd_jobs_stale_or_mismatched")
    if no_matching_runner:
        blockers.append("gitea_queue_harbor_110_repair_no_matching_runner")
    if harbor_repair_remote_control_unavailable:
        blockers.append("gitea_queue_harbor_110_remote_control_channel_unavailable")
    if harbor_repair_local_v2_unavailable:
        blockers.append("gitea_queue_harbor_110_remote_local_registry_v2_unavailable")
    if harbor_repair_public_v2_unavailable:
        blockers.append(
            "gitea_queue_harbor_public_registry_v2_unavailable_after_remote_repair"
        )
    if (
        blocked
        and not no_matching_runner
        and not harbor_repair_remote_control_unavailable
        and not harbor_repair_local_v2_unavailable
        and not harbor_repair_public_v2_unavailable
        and not harbor_running_jobs_api_stale
        and not current_cd_waiting_behind_harbor_running
    ):
        blockers.append("gitea_queue_harbor_110_repair_blocked")
    if jobs_cross_workflow:
        blockers.append("gitea_queue_harbor_110_repair_jobs_cross_workflow_mismatch")
    if jobs_stale:
        blockers.append("gitea_queue_harbor_110_repair_jobs_stale_or_mismatched")
    if harbor_running_jobs_api_stale:
        blockers.append(
            "gitea_queue_harbor_110_repair_visible_running_jobs_api_stale"
        )
    if current_cd_waiting_behind_harbor_running:
        blockers.append(
            "gitea_queue_current_cd_waiting_behind_harbor_110_repair_running"
        )
    if boundary_violation:
        blockers.append("gitea_queue_readback_boundary_violation")
    return blockers


def _deploy_marker_readback(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        return {
            "receipt_seen": False,
            "schema_version": "",
            "status": "not_provided",
            "expected_commit_sha": "",
            "deploy_marker_commit_sha": "",
            "production_image_tag_sha": "",
            "latest_visible_cd_run_id": "",
            "latest_visible_cd_run_status": "",
            "production_preflight_http_status": None,
            "production_workbench_http_status": None,
            "production_image_matches_expected": False,
            "production_readback_http_green": False,
            "deploy_marker_verified": False,
            "operation_boundary_violation": False,
            "blockers": [],
            "blocker_count": 0,
            "metadata_only": True,
            "raw_output_returned": False,
        }

    readback = _dict(value.get("readback"))
    rollups = _dict(value.get("rollups"))
    operation_boundaries = _dict(value.get("operation_boundaries"))
    expected_sha = str(
        value.get("expected_commit_sha")
        or readback.get("expected_commit_sha")
        or readback.get("observed_source_control_main_sha")
        or readback.get("latest_visible_cd_run_commit_sha")
        or ""
    )
    production_image_sha = str(
        readback.get("production_image_tag_sha")
        or readback.get("runtime_build_commit_sha")
        or ""
    )
    production_image_matches = bool(
        expected_sha
        and production_image_sha
        and production_image_sha == expected_sha
    ) or bool(rollups.get("production_image_tag_matches_main") is True)
    preflight_status = _int_or_none(readback.get("production_preflight_http_status"))
    workbench_status = _int_or_none(readback.get("production_workbench_http_status"))
    http_green = preflight_status == 200 and workbench_status == 200
    latest_cd_status = str(readback.get("latest_visible_cd_run_status") or "")
    cd_success = latest_cd_status in {"", "Success", "success"}
    boundary_violation = any(
        operation_boundaries.get(flag) is True
        for flag in (
            "deploy_trigger_allowed",
            "workflow_modification_allowed",
            "gitea_api_write_allowed",
            "host_write_allowed",
            "k8s_write_allowed",
            "secret_read_allowed",
            "github_api_allowed",
            "raw_session_or_sqlite_read_allowed",
        )
    )
    status = str(value.get("status") or "unknown")
    verified = bool(
        status == "closure_verified"
        and production_image_matches
        and http_green
        and cd_success
        and not boundary_violation
    )
    blockers = _deploy_marker_blockers(
        verified=verified,
        status=status,
        production_image_matches=production_image_matches,
        http_green=http_green,
        cd_success=cd_success,
        boundary_violation=boundary_violation,
    )
    return {
        "receipt_seen": True,
        "schema_version": str(value.get("schema_version") or ""),
        "status": status,
        "expected_commit_sha": expected_sha,
        "deploy_marker_commit_sha": str(
            readback.get("deploy_marker_commit_sha") or ""
        ),
        "production_image_tag_sha": production_image_sha,
        "latest_visible_cd_run_id": str(
            readback.get("latest_visible_cd_run_id") or ""
        ),
        "latest_visible_cd_run_status": latest_cd_status,
        "production_preflight_http_status": preflight_status,
        "production_workbench_http_status": workbench_status,
        "production_image_matches_expected": production_image_matches,
        "production_readback_http_green": http_green,
        "deploy_marker_verified": verified,
        "operation_boundary_violation": boundary_violation,
        "blockers": blockers,
        "blocker_count": len(blockers),
        "metadata_only": True,
        "raw_output_returned": False,
    }


def _deploy_marker_blockers(
    *,
    verified: bool,
    status: str,
    production_image_matches: bool,
    http_green: bool,
    cd_success: bool,
    boundary_violation: bool,
) -> list[str]:
    if verified:
        return []
    blockers: list[str] = []
    if status != "closure_verified":
        blockers.append("deploy_marker_closure_not_verified")
    if not production_image_matches:
        blockers.append("deploy_marker_production_image_not_current_main")
    if not http_green:
        blockers.append("deploy_marker_production_readback_http_not_green")
    if not cd_success:
        blockers.append("deploy_marker_latest_cd_run_not_success")
    if boundary_violation:
        blockers.append("deploy_marker_readback_boundary_violation")
    return blockers


def _active_blockers(
    *,
    ssh_diagnosis: dict[str, Any],
    ssh_local: dict[str, Any],
    watchdog_check: dict[str, Any],
    watchdog_repair: dict[str, Any],
    controlled_cd_lane: dict[str, Any],
    non110_runner: dict[str, Any],
    verifier: dict[str, Any],
    gitea_queue: dict[str, Any],
    deploy_marker: dict[str, Any],
) -> list[str]:
    blockers: list[str] = []
    command_path_ready = bool(ssh_diagnosis["wooo_command_path_ready"])
    if ssh_diagnosis["receipt_seen"] and not ssh_diagnosis["diagnosis_ready"]:
        blockers.append("ssh_publickey_diagnosis_receipt_not_ready")
    if (
        ssh_diagnosis["publickey_offer_timeout_effective_seen"]
        and not ssh_local["control_channel_metadata_ready"]
    ):
        blockers.append("ssh_publickey_offer_timeout_on_wooo")
    if (
        ssh_diagnosis["server_accepts_key_then_timeout_effective_seen"]
        and not ssh_local["control_channel_metadata_ready"]
    ):
        blockers.append("ssh_publickey_server_accepts_key_then_timeout_on_wooo")
    if ssh_diagnosis["runner_systemctl_show_timeout_effective_seen"]:
        blockers.append("runner_systemctl_show_timeout_on_110")
    if ssh_diagnosis["node_high_load_seen"]:
        blockers.append("ssh_publickey_node_high_load_on_110")
    if (
        ssh_local["receipt_seen"]
        and not ssh_local["control_channel_metadata_ready"]
        and not command_path_ready
    ):
        blockers.extend(_ssh_local_metadata_blockers(ssh_local))
        blockers.append("ssh_local_repair_receipt_metadata_not_ready")
    if not watchdog_check["receipt_seen"]:
        blockers.append("harbor_watchdog_check_receipt_missing")
    elif watchdog_check["forbidden_action_seen"]:
        blockers.append("harbor_watchdog_check_forbidden_action_seen")
    elif not watchdog_check["harbor_ready"] and not watchdog_repair["receipt_seen"]:
        blockers.append("harbor_watchdog_repair_once_receipt_missing")
    if watchdog_repair["receipt_seen"] and watchdog_repair["forbidden_action_seen"]:
        blockers.append("harbor_watchdog_repair_forbidden_action_seen")
    if watchdog_repair["receipt_seen"] and not watchdog_repair["harbor_ready"]:
        blockers.append("harbor_watchdog_repair_did_not_restore_local_v2")
    blockers.extend(_strings(controlled_cd_lane.get("blockers")))
    blockers.extend(_strings(non110_runner.get("blockers")))
    if not verifier["public_registry_v2_ready"]:
        blockers.append("public_registry_v2_verifier_not_green")
    if not verifier["internal_registry_v2_ready"]:
        blockers.append("internal_registry_v2_verifier_not_green")
    blockers.extend(
        _effective_gitea_queue_blockers(
            ssh_diagnosis=ssh_diagnosis,
            ssh_local=ssh_local,
            watchdog_check=watchdog_check,
            controlled_cd_lane=controlled_cd_lane,
            verifier=verifier,
            gitea_queue=gitea_queue,
        )
    )
    blockers.extend(_strings(deploy_marker.get("blockers")))
    return _unique_strings(blockers)


def _effective_gitea_queue_blockers(
    *,
    ssh_diagnosis: dict[str, Any],
    ssh_local: dict[str, Any],
    watchdog_check: dict[str, Any],
    controlled_cd_lane: dict[str, Any],
    verifier: dict[str, Any],
    gitea_queue: dict[str, Any],
) -> list[str]:
    blockers = _strings(gitea_queue.get("blockers"))
    superseded = _gitea_queue_blockers_superseded_by_local_receipt(
        ssh_diagnosis=ssh_diagnosis,
        ssh_local=ssh_local,
        watchdog_check=watchdog_check,
        controlled_cd_lane=controlled_cd_lane,
        verifier=verifier,
    )
    if not superseded:
        return blockers
    return [blocker for blocker in blockers if blocker not in superseded]


def _gitea_queue_blockers_superseded_by_local_receipt(
    *,
    ssh_diagnosis: dict[str, Any],
    ssh_local: dict[str, Any],
    watchdog_check: dict[str, Any],
    controlled_cd_lane: dict[str, Any],
    verifier: dict[str, Any],
) -> set[str]:
    control_channel_ready = bool(
        ssh_diagnosis["wooo_command_path_ready"]
        or ssh_local["control_channel_metadata_ready"]
    )
    harbor_ready = bool(watchdog_check["harbor_ready"] or verifier["registry_v2_ready"])
    if not control_channel_ready or not harbor_ready:
        return set()

    superseded = {
        "gitea_queue_harbor_110_remote_control_channel_unavailable",
        "gitea_queue_harbor_110_repair_blocked",
        "gitea_queue_harbor_110_repair_visible_running_jobs_api_stale",
        "gitea_queue_current_cd_waiting_behind_harbor_110_repair_running",
    }
    if verifier["registry_v2_ready"]:
        superseded.update(
            {
                "gitea_queue_current_cd_harbor_retrying_unavailable",
                "gitea_queue_harbor_110_remote_local_registry_v2_unavailable",
                "gitea_queue_harbor_public_registry_v2_unavailable_after_remote_repair",
            }
        )
    if controlled_cd_lane["receipt_seen"]:
        superseded.update(
            {
                "gitea_queue_harbor_110_repair_no_matching_runner",
                "gitea_queue_harbor_110_repair_jobs_cross_workflow_mismatch",
                "gitea_queue_harbor_110_repair_jobs_stale_or_mismatched",
                "gitea_queue_cd_harbor_repair_blocked_by_no_matching_awoooi_host",
                "gitea_queue_cd_harbor_repair_requires_110_controlled_lane",
            }
        )
    return superseded


def _ssh_local_metadata_blockers(ssh_local: dict[str, Any]) -> list[str]:
    blockers: list[str] = []
    if not ssh_local["sshd_config_syntax_ok"]:
        blockers.append("ssh_local_repair_sshd_config_syntax_not_ok")
    if not ssh_local["sshd_config_syntax_after_apply_ok"]:
        blockers.append("ssh_local_repair_sshd_config_syntax_after_apply_not_ok")
    if not ssh_local["target_user_exists"]:
        blockers.append("ssh_local_repair_target_user_missing")
    if ssh_local["target_user_account_locked"]:
        blockers.append("ssh_local_repair_target_user_account_locked")
    if not ssh_local["target_user_shell_executable"]:
        blockers.append("ssh_local_repair_target_user_shell_not_executable")
    if not ssh_local["authorized_keys_metadata_present"]:
        blockers.append("ssh_local_repair_authorized_keys_missing")
    if not ssh_local["sshd_effective_config_available"]:
        blockers.append("ssh_local_repair_sshd_effective_config_unavailable")
    if ssh_local["sshd_pubkeyauthentication"] != "yes":
        blockers.append("ssh_local_repair_sshd_pubkeyauthentication_not_enabled")
    if not ssh_local["sshd_authorized_keys_file_default"]:
        blockers.append("ssh_local_repair_sshd_authorized_keys_file_drift")
    return blockers


def _status(
    *,
    ssh_diagnosis: dict[str, Any],
    ssh_local: dict[str, Any],
    watchdog_check: dict[str, Any],
    watchdog_repair: dict[str, Any],
    controlled_cd_lane: dict[str, Any],
    non110_runner: dict[str, Any],
    verifier: dict[str, Any],
    gitea_queue: dict[str, Any],
    deploy_marker: dict[str, Any],
    active_blockers: list[str],
) -> str:
    if verifier["registry_v2_ready"] and not active_blockers:
        return "harbor_registry_recovery_receipt_verified"
    if controlled_cd_lane["receipt_seen"] and not controlled_cd_lane[
        "controlled_cd_lane_ready"
    ]:
        return "controlled_cd_lane_readiness_receipt_blocked"
    if non110_runner["receipt_seen"] and not non110_runner["non110_runner_ready"]:
        return "non110_runner_readiness_receipt_blocked"
    if (
        verifier["registry_v2_ready"]
        and gitea_queue["receipt_seen"]
        and any(blocker.startswith("gitea_queue_") for blocker in active_blockers)
    ):
        return "harbor_registry_recovery_receipt_verified_waiting_gitea_queue_clearance"
    if (
        verifier["registry_v2_ready"]
        and deploy_marker["receipt_seen"]
        and deploy_marker["blocker_count"] > 0
    ):
        return "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback"
    if watchdog_repair["receipt_seen"]:
        return "harbor_registry_repair_receipt_waiting_registry_v2_verifier"
    if watchdog_check["receipt_seen"] and watchdog_check["harbor_ready"]:
        return "harbor_local_registry_ready_waiting_public_registry_v2_verifier"
    if watchdog_check["receipt_seen"]:
        return "harbor_watchdog_check_unhealthy_waiting_repair_once_receipt"
    if ssh_local["receipt_seen"]:
        return "ssh_local_repair_receipt_waiting_harbor_watchdog_check"
    if ssh_diagnosis["receipt_seen"]:
        return "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check"
    return "blocked_waiting_harbor_controlled_recovery_receipt"


def _safe_next_step(*, status: str) -> str:
    if status == "harbor_registry_recovery_receipt_verified":
        return "retry_gitea_cd_then_verify_deploy_marker_and_priority_readback"
    if status == "harbor_registry_recovery_receipt_verified_waiting_gitea_queue_clearance":
        return "clear_harbor_110_runner_queue_then_retry_cd_marker_readback"
    if status == "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback":
        return "rerun_gitea_cd_then_verify_deploy_marker_and_priority_readback"
    if status == "controlled_cd_lane_readiness_receipt_blocked":
        return "fix_controlled_cd_lane_guardrail_blockers_then_rerun_readiness_verifier"
    if status == "non110_runner_readiness_receipt_blocked":
        return "fix_or_register_non110_runner_then_rerun_readiness_and_queue_verifiers"
    if status == "harbor_registry_repair_receipt_waiting_registry_v2_verifier":
        return "rerun_public_and_internal_registry_v2_verifier_before_cd_retry"
    if status == "harbor_local_registry_ready_waiting_public_registry_v2_verifier":
        return "verify_public_registry_v2_route_then_retry_gitea_cd"
    if status == "harbor_watchdog_check_unhealthy_waiting_repair_once_receipt":
        return "run_harbor_watchdog_repair_once_then_submit_receipt_with_verifier"
    if status == "ssh_local_repair_receipt_waiting_harbor_watchdog_check":
        return "rerun_ssh_control_channel_probe_then_harbor_watchdog_check_mode"
    if (
        status
        == "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check"
    ):
        return "run_110_local_ssh_metadata_check_then_harbor_watchdog_check_mode"
    return "submit_non_secret_ssh_local_or_harbor_watchdog_receipt"


def _current_apply_blocker(*, status: str) -> str:
    if status == "harbor_registry_recovery_receipt_verified":
        return "receipt_verified_waiting_cd_marker_readback"
    if status == "harbor_registry_recovery_receipt_verified_waiting_gitea_queue_clearance":
        return "gitea_queue_clearance_required_after_registry_receipt"
    if status == "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback":
        return "deploy_marker_readback_required_after_registry_receipt"
    if status == "controlled_cd_lane_readiness_receipt_blocked":
        return "controlled_cd_lane_readiness_required_for_awoooi_host_queue"
    if status == "non110_runner_readiness_receipt_blocked":
        return "non110_runner_readiness_required_for_current_cd_queue"
    if status == "harbor_watchdog_check_unhealthy_waiting_repair_once_receipt":
        return "repair_once_receipt_required_after_unhealthy_check"
    if status == "ssh_local_repair_receipt_waiting_harbor_watchdog_check":
        return "harbor_watchdog_check_receipt_required_after_ssh_local_repair"
    if (
        status
        == "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check"
    ):
        return "110_local_ssh_metadata_or_harbor_watchdog_check_receipt_required"
    return "registry_v2_verifier_or_controlled_receipt_required"


def _learning_writeback_contracts(*, status: str) -> list[dict[str, Any]]:
    targets = [
        "timeline_event",
        "knowledge_entry",
        "rag_embedding",
        "mcp_tool_registry_signal",
        "playbook_trust_score",
        "controlled_recovery_report",
    ]
    return [
        {
            "target": target,
            "writeback_mode": "metadata_only",
            "raw_log_allowed": False,
            "secret_value_allowed": False,
            "receipt_key": f"harbor_registry_recovery_receipt::{status}::{target}",
        }
        for target in targets
    ]


def _parse_key_values(output: str) -> dict[str, str]:
    fields: dict[str, str] = {}
    for raw_line in output.splitlines():
        for token in raw_line.strip().split():
            if "=" not in token:
                continue
            key, value = token.split("=", 1)
            if key:
                fields[key] = value
    return fields


def _prefixed_blockers(output: str, *, prefix: str) -> list[str]:
    blockers: list[str] = []
    for raw_line in output.splitlines():
        line = raw_line.strip()
        if not line.startswith("BLOCKER "):
            continue
        blocker = line.split(" ", 1)[1].strip()
        if blocker:
            blockers.append(f"{prefix}{blocker}")
    return _unique_strings(blockers)


def _mode_from_marker_line(output: str) -> str:
    for raw_line in output.splitlines():
        if "AWOOOI_110_SSH_PUBLICKEY_AUTH_LOCAL_REPAIR" not in raw_line:
            continue
        fields = _parse_key_values(raw_line)
        return str(fields.get("mode") or "")
    return ""


def _text(value: Any) -> str:
    return value if isinstance(value, str) else ""


def _receipt_output(specific_output: str, local_console_output: str) -> str:
    if specific_output.strip():
        return specific_output
    return local_console_output


def _text_stats(value: str) -> dict[str, Any]:
    return {
        "provided": bool(value),
        "byte_count": len(value.encode("utf-8")),
        "line_count": len(value.splitlines()) if value else 0,
    }


def _bool_from_field(value: Any) -> bool:
    return str(value).strip().lower() in {"1", "true", "yes", "ok", "done"}


def _int_or_none(value: Any) -> int | None:
    try:
        return int(str(value))
    except (TypeError, ValueError):
        return None


def _float_or_none(value: Any) -> float | None:
    try:
        return float(str(value))
    except (TypeError, ValueError):
        return None


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


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


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
