content-find-agent-new/content_agent/business_modules/rule_judgment/evaluator.py

from __future__ import annotations

from typing import Any

from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.record_payload import with_raw_payload


HARD_GATE_REASON_CATEGORY = "hard_gate"
SCORE_REASON_CATEGORY = "score_or_data_failed"
REVIEW_REASON_CATEGORY = "review_needed"
SUCCESS_REASON_CATEGORY = "score_pass"
TECHNICAL_REASON_CATEGORY = "technical_error"
V4_SCORECARD_SCHEMA_VERSION = "v4_scorecard.v1"
V4_POOL_REASON = "v4_query_and_platform_pass"
V4_REVIEW_REASON = "v4_score_review_needed"
V4_REJECT_REASON = "v4_query_or_score_below_threshold"
V4_TECHNICAL_RETRY_REASON = "v4_technical_retry_needed"
V4_TECHNICAL_RETRY_ACTION = "TECHNICAL_RETRY_REQUIRED"
RULE_DECISION_RAW_PAYLOAD_EXCLUDE_KEYS = {
    "run_id",
    "policy_run_id",
    "decision_id",
    "policy_bundle_id",
    "rule_pack_id",
    "rule_pack_version",
    "strategy_version",
    "decision_target_type",
    "decision_target_id",
    "triggered_blocking_rules",
    "scorecard",
    "score",
    "decision_action",
    "decision_reason_code",
    "search_query_effect_status",
    "source_evidence",
    "decision_replay_data",
}
# V5-M2:V4 权重从规则包 score_weight_profiles 读取;缺配置时 fallback 保持历史行为。
V4_PORTRAIT_UNAVAILABLE_REASON = "portrait_unavailable"
V4_PORTRAIT_INCOMPLETE_REASON = "portrait_incomplete"


def decide(
    run_id: str,
    policy_run_id: str,
    index: int,
    bundle: dict[str, Any],
    policy_bundle: dict[str, Any],
) -> dict[str, Any]:
    rule_pack = policy_bundle["rule_pack"]
    matched_gates = _evaluate_hard_gates(bundle, rule_pack.get("hard_gates", []))
    blocking_gates = [
        gate for gate in matched_gates if gate.get("severity") == "fatal" or gate.get("stop_scoring")
    ]
    triggered_blocking_rules = [gate["gate_id"] for gate in blocking_gates]
    if blocking_gates:
        primary_gate = _primary_hard_gate(blocking_gates)
        replay_marker = {
            "matched_gates": [_gate_replay(gate) for gate in matched_gates],
            "primary_gate_id": primary_gate["gate_id"],
            "primary_reason_code": primary_gate["decision_reason_code"],
            "primary_gate_priority": primary_gate.get("priority"),
            "reason_category": HARD_GATE_REASON_CATEGORY,
        }
        scorecard = _scorecard(None, rule_pack.get("scorecard", {}), missing=True)
        if _is_v4_policy(policy_bundle):
            replay_marker.update(_v4_walk_gate(False, primary_gate["decision_reason_code"]))
            if primary_gate["decision_reason_code"] == V4_TECHNICAL_RETRY_REASON:
                scorecard = _v4_scorecard_total(bundle, policy_bundle)
        effect = _effect_status_for_decision(
            policy_bundle,
            primary_gate["decision_action"],
            HARD_GATE_REASON_CATEGORY,
            is_hard_gate=True,
        )
        return _build_decision(
            run_id=run_id,
            policy_run_id=policy_run_id,
            index=index,
            bundle=bundle,
            policy_bundle=policy_bundle,
            decision_action=primary_gate["decision_action"],
            decision_reason_code=primary_gate["decision_reason_code"],
            search_query_effect_status=effect["content_effect_status"],
            triggered_blocking_rules=triggered_blocking_rules,
            scorecard=scorecard,
            score=None,
            replay_marker={**replay_marker, "effect_mapping_id": effect.get("mapping_id")},
        )

    if _is_v4_policy(policy_bundle):
        return _decide_v4(
            run_id,
            policy_run_id,
            index,
            bundle,
            policy_bundle,
            matched_gates,
            triggered_blocking_rules,
        )

    scorecard = _scorecard_total(bundle, rule_pack.get("scorecard", {}))
    score = scorecard.get("total_score")
    if score is None or _scorecard_all_dimensions_missing(scorecard):
        return _missing_score_decision(
            run_id,
            policy_run_id,
            index,
            bundle,
            policy_bundle,
            matched_gates,
            scorecard,
        )

    threshold = _match_threshold(score, rule_pack.get("thresholds", []))
    if threshold is None:
        raise ValueError(f"no threshold matched score: {score}")
    decision_action = threshold["decision_action"]
    decision_reason_code = threshold["decision_reason_code"]
    reason_category = _reason_category_for_threshold(decision_action, decision_reason_code)
    effect = _effect_status_for_decision(
        policy_bundle, decision_action, reason_category, is_hard_gate=False
    )
    return _build_decision(
        run_id=run_id,
        policy_run_id=policy_run_id,
        index=index,
        bundle=bundle,
        policy_bundle=policy_bundle,
        decision_action=decision_action,
        decision_reason_code=decision_reason_code,
        search_query_effect_status=effect["content_effect_status"],
        triggered_blocking_rules=triggered_blocking_rules,
        scorecard=scorecard,
        score=score,
        replay_marker={
            "matched_gates": [_gate_replay(gate) for gate in matched_gates],
            "matched_threshold": _threshold_label(threshold),
            "matched_threshold_priority": threshold.get("priority"),
            "effect_mapping_id": effect.get("mapping_id"),
            "reason_category": reason_category,
            "matched_scoring_rules": scorecard.get("matched_scoring_rules", []),
        },
    )


def _is_v4_policy(policy_bundle: dict[str, Any]) -> bool:
    return (
        policy_bundle.get("strategy_version") == "V4"
        or policy_bundle.get("rule_pack_version") == "4.0.0"
        or policy_bundle.get("rule_pack", {}).get("scorecard", {}).get("schema_version")
        == V4_SCORECARD_SCHEMA_VERSION
    )


def _decide_v4(
    run_id: str,
    policy_run_id: str,
    index: int,
    bundle: dict[str, Any],
    policy_bundle: dict[str, Any],
    matched_gates: list[dict[str, Any]],
    triggered_blocking_rules: list[str],
) -> dict[str, Any]:
    scorecard = _v4_scorecard_total(bundle, policy_bundle)
    score = scorecard.get("total_score")
    thresholds = _v4_thresholds(policy_bundle, _get_path(bundle, "content.platform"))
    scorecard["score_thresholds"] = thresholds  # 随决策落库:供前端展示实际生效门槛(单一真源)
    if score is None:
        decision_action = V4_TECHNICAL_RETRY_ACTION
        decision_reason_code = _v4_technical_retry_reason(scorecard)
    else:
        decision_action, decision_reason_code = _v4_decision(scorecard, thresholds)
    reason_category = _reason_category_for_threshold(decision_action, decision_reason_code)
    effect = _effect_status_for_decision(
        policy_bundle, decision_action, reason_category, is_hard_gate=False
    )
    allow_walk = _v4_allow_walk(scorecard, score, thresholds)
    return _build_decision(
        run_id=run_id,
        policy_run_id=policy_run_id,
        index=index,
        bundle=bundle,
        policy_bundle=policy_bundle,
        decision_action=decision_action,
        decision_reason_code=decision_reason_code,
        search_query_effect_status=effect["content_effect_status"],
        triggered_blocking_rules=triggered_blocking_rules,
        scorecard=scorecard,
        score=score,
        replay_marker={
            "matched_gates": [_gate_replay(gate) for gate in matched_gates],
            "effect_mapping_id": effect.get("mapping_id"),
            "reason_category": reason_category,
            **_v4_walk_gate(
                allow_walk,
                decision_reason_code,
                scorecard=scorecard,
                score=score,
                thresholds=thresholds,
            ),
        },
    )


def _v4_scorecard_total(bundle: dict[str, Any], policy_bundle: dict[str, Any]) -> dict[str, Any]:
    query_score = _query_relevance_from_bundle(bundle)
    platform = _platform_performance_from_bundle(bundle)
    platform_score = platform.get("platform_performance_score") if isinstance(platform, dict) else None
    missing = platform.get("missing_observable_fields", []) if isinstance(platform, dict) else []
    components = platform.get("platform_performance_components", []) if isinstance(platform, dict) else []
    scorecard = {
        "schema_version": V4_SCORECARD_SCHEMA_VERSION,
        "query_relevance_score": query_score,
        "platform_performance_score": platform_score,
        "platform_performance_components": components,
        "missing_observable_fields": missing if isinstance(missing, list) else [],
    }
    fifty = _fifty_plus_from_bundle(bundle)
    selected_profile, weights = _v4_score_weight_profile(policy_bundle, bundle)
    scorecard["score_weights"] = weights
    missing_failure = _v4_missing_critical_dimension_failure(selected_profile, bundle)
    if missing_failure:
        scorecard.update(missing_failure)
        scorecard["total_score"] = None
        scorecard["score_missing"] = True
        scorecard.update(_technical_failure_fields(bundle))
        return scorecard
    if fifty is None:
        total = _v4_weighted_total(
            {"query_relevance": query_score, "platform_performance": platform_score},
            weights,
        )
    else:
        status = fifty.get("status")
        scorecard["fifty_plus_status"] = status
        fifty_score = fifty.get("score")
        if status == "ok" and _is_number(query_score) and _is_number(platform_score) and _is_number(fifty_score):
            scorecard["fifty_plus_score"] = fifty_score
            scorecard["fifty_plus_components"] = fifty.get("components", {})
            total = _v4_weighted_total(
                {
                    "query_relevance": query_score,
                    "platform_performance": platform_score,
                    "fifty_plus": fifty_score,
                },
                weights,
            )
        elif status == "not_attempted" and _is_number(query_score) and _is_number(platform_score):
            total = _v4_weighted_total(
                {"query_relevance": query_score, "platform_performance": platform_score},
                weights,
            )
        else:
            # unavailable / incomplete / 缺分 → 技术失败(停游走)。把失败类型+原因写进 scorecard,供技术详情展示。
            total = None
            scorecard["failure_type"] = (
                V4_PORTRAIT_UNAVAILABLE_REASON
                if status == "unavailable"
                else V4_PORTRAIT_INCOMPLETE_REASON
                if status == "incomplete"
                else "fifty_plus_missing"
            )
            fifty_reason = fifty.get("failure_reason")
            if fifty_reason:
                scorecard["fifty_plus_failure_reason"] = fifty_reason
    scorecard["total_score"] = total
    scorecard["score_missing"] = total is None
    scorecard.update(_technical_failure_fields(bundle))
    return scorecard


_V4_SCORE_WEIGHT_FALLBACK_PROFILES = {
    "default_two_dimension": {"query_relevance": 50, "platform_performance": 50},
    "douyin_fifty_plus_ok": {"query_relevance": 35, "platform_performance": 35, "fifty_plus": 30},
    "douyin_fifty_plus_not_attempted": {"query_relevance": 35, "platform_performance": 35},
}


def _v4_score_weights(policy_bundle: dict[str, Any], bundle: dict[str, Any]) -> dict[str, float]:
    _, weights = _v4_score_weight_profile(policy_bundle, bundle)
    return weights


def _v4_score_weight_profile(policy_bundle: dict[str, Any], bundle: dict[str, Any]) -> tuple[dict[str, Any], dict[str, float]]:
    profiles = (
        policy_bundle.get("rule_pack", {})
        .get("scorecard", {})
        .get("score_weight_profiles")
    )
    generic_profile = _select_generic_score_weight_profile(profiles, bundle)
    if generic_profile:
        return generic_profile, _decimal_weights(generic_profile.get("weights"))
    if not isinstance(profiles, dict) or "profiles" in profiles:
        profiles = _V4_SCORE_WEIGHT_FALLBACK_PROFILES
    fifty = _fifty_plus_from_bundle(bundle)
    if fifty is None:
        profile_name = "default_two_dimension"
    elif fifty.get("status") == "not_attempted":
        profile_name = "douyin_fifty_plus_not_attempted"
    else:
        profile_name = "douyin_fifty_plus_ok"
    raw_weights = profiles.get(profile_name)
    if not isinstance(raw_weights, dict) or not raw_weights:
        raw_weights = _V4_SCORE_WEIGHT_FALLBACK_PROFILES[profile_name]
    return {"profile_id": profile_name, "weights": raw_weights}, _decimal_weights(raw_weights)


def _decimal_weights(raw_weights: Any) -> dict[str, float]:
    if not isinstance(raw_weights, dict):
        return {}
    return {
        str(key): float(value) / 100
        for key, value in raw_weights.items()
        if _is_number(value)
    }


def _select_generic_score_weight_profile(profiles: Any, bundle: dict[str, Any]) -> dict[str, Any] | None:
    if not isinstance(profiles, dict) or not isinstance(profiles.get("profiles"), list):
        return None
    platform = str(_get_path(bundle, "content.platform") or "")
    content_format = str(_get_path(bundle, "content.platform_content_format") or "video")
    matches = [
        profile
        for profile in profiles["profiles"]
        if isinstance(profile, dict)
        and profile.get("approval_status") == "approved"
        and profile.get("runtime_enabled") is True
        and _scope_matches(profile.get("platform_scope"), platform)
        and _content_format_matches(profile.get("content_format"), content_format)
        and _applies_when_matches(str(profile.get("applies_when") or "always"), bundle)
    ]
    if not matches:
        return None
    matches.sort(key=lambda row: int(row.get("priority") or 0), reverse=True)
    if len(matches) > 1 and int(matches[0].get("priority") or 0) == int(matches[1].get("priority") or 0):
        raise ValueError(f"score_weight_profile_priority_conflict: {matches[0].get('profile_id')}, {matches[1].get('profile_id')}")
    return matches[0]


def _scope_matches(scope: Any, platform: str) -> bool:
    values = scope if isinstance(scope, list) else [scope]
    return "*" in values or platform in values


def _content_format_matches(scope: Any, content_format: str) -> bool:
    if scope in {None, "", "*"}:
        return True
    values = scope if isinstance(scope, list) else [scope]
    return "*" in values or content_format in values


def _applies_when_matches(expr: str, bundle: dict[str, Any]) -> bool:
    if expr in {"", "always"}:
        return True
    if expr == "no_extra_dimension_available":
        return _fifty_plus_from_bundle(bundle) is None
    if expr.startswith("field_exists:"):
        return _get_path(bundle, expr.removeprefix("field_exists:")) is not None
    if expr.startswith("field_missing:"):
        return _get_path(bundle, expr.removeprefix("field_missing:")) is None
    if expr.startswith("field_equals:"):
        _, path, expected = expr.split(":", 2)
        return str(_get_path(bundle, path)) == expected
    raise ValueError(f"unsupported score_weight_profile applies_when: {expr}")


def _v4_missing_critical_dimension_failure(profile: dict[str, Any], bundle: dict[str, Any]) -> dict[str, Any] | None:
    if profile.get("missing_dimension_policy") != "technical_retry":
        return None
    paths = profile.get("score_source_paths") if isinstance(profile.get("score_source_paths"), dict) else {}
    for dimension in profile.get("critical_dimensions") or []:
        source_path = paths.get(dimension)
        if source_path and _get_path(bundle, source_path) is None:
            return {
                "failure_type": "critical_dimension_missing",
                "final_status": "failed",
                "missing_dimension_key": dimension,
                "missing_dimension_score_source_path": source_path,
                "missing_dimension_failure_reason": f"missing critical dimension {dimension} at {source_path}",
            }
    return None


def _v4_weighted_total(scores: dict[str, Any], weights: dict[str, float]) -> float | None:
    if not weights:
        return None
    total = 0.0
    for key, weight in weights.items():
        score = scores.get(key)
        if not _is_number(score):
            return None
        total += float(score) * weight
    return round(total, 2)


def _fifty_plus_from_bundle(bundle: dict[str, Any]) -> dict[str, Any] | None:
    block = bundle.get("content_audience_50plus")
    return block if isinstance(block, dict) else None


def _v4_technical_retry_reason(scorecard: dict[str, Any]) -> str:
    status = scorecard.get("fifty_plus_status")
    if status == "unavailable":
        return V4_PORTRAIT_UNAVAILABLE_REASON
    if status == "incomplete":
        return V4_PORTRAIT_INCOMPLETE_REASON
    return V4_TECHNICAL_RETRY_REASON


def _query_relevance_from_bundle(bundle: dict[str, Any]) -> Any:
    return _get_path(bundle, "pattern_match_result.query_relevance_score")


def _platform_performance_from_bundle(bundle: dict[str, Any]) -> dict[str, Any]:
    value = _get_path(bundle, "content_engagement_metrics.platform_performance")
    return value if isinstance(value, dict) else {}


_V4_THRESHOLDS_FALLBACK = {
    "pool_total": 70, "pool_query": 70,
    "review_total": 55, "review_query": 55,
    "walk_query": 70, "walk_platform": 65, "walk_total": 70,
}


def _v4_thresholds(policy_bundle: dict[str, Any], platform: Any) -> dict[str, Any]:
    """按平台解析 V4 判定/游走门槛:score_thresholds.json 的 default 合并 per-platform 覆盖。
    缺配置时回退到内置字面量(向后兼容)。改门槛只动 tech_documents/数据接口与来源/score_thresholds.json。"""
    cfg = policy_bundle.get("score_thresholds")
    if not isinstance(cfg, dict):
        return dict(_V4_THRESHOLDS_FALLBACK)
    default = cfg.get("default") if isinstance(cfg.get("default"), dict) else {}
    override = cfg.get(str(platform)) if isinstance(cfg.get(str(platform)), dict) else {}
    return {**_V4_THRESHOLDS_FALLBACK, **default, **override}


def _v4_decision(scorecard: dict[str, Any], thresholds: dict[str, Any]) -> tuple[str, str]:
    query = scorecard.get("query_relevance_score")
    score = scorecard.get("total_score")
    if _is_number(query) and _is_number(score) and query >= thresholds["pool_query"] and score >= thresholds["pool_total"]:
        return "ADD_TO_CONTENT_POOL", V4_POOL_REASON
    if _is_number(query) and _is_number(score) and query >= thresholds["review_query"] and score >= thresholds["review_total"]:
        return "KEEP_CONTENT_FOR_REVIEW", V4_REVIEW_REASON
    return "REJECT_CONTENT", V4_REJECT_REASON


def _technical_failure_fields(bundle: dict[str, Any]) -> dict[str, Any]:
    pattern = bundle.get("pattern_match_result")
    if not isinstance(pattern, dict):
        return {}
    fields = {}
    for field in ["failure_type", "exception_type", "http_status_code", "retry_count", "final_status"]:
        if field in pattern:
            fields[field] = pattern.get(field)
    return fields


def _v4_allow_walk(scorecard: dict[str, Any], score: Any, thresholds: dict[str, Any]) -> bool:
    query = scorecard.get("query_relevance_score")
    platform = scorecard.get("platform_performance_score")
    return (
        _is_number(query)
        and _is_number(platform)
        and _is_number(score)
        and query >= thresholds["walk_query"]
        and platform >= thresholds["walk_platform"]
        and score >= thresholds["walk_total"]
    )


def _v4_walk_gate(
    allow_walk: bool,
    reason: str,
    *,
    scorecard: dict[str, Any] | None = None,
    score: Any = None,
    thresholds: dict[str, Any] | None = None,
) -> dict[str, Any]:
    scorecard = scorecard or {}
    thresholds = thresholds or _V4_THRESHOLDS_FALLBACK
    snapshot = {
        "query_relevance_score": scorecard.get("query_relevance_score"),
        "platform_performance_score": scorecard.get("platform_performance_score"),
        "score": score,
    }
    return {
        "allow_walk": allow_walk,
        "allow_walk_reason": (
            f"query>={thresholds['walk_query']}/platform>={thresholds['walk_platform']}/score>={thresholds['walk_total']}"
            if allow_walk
            else reason
        ),
        "walk_gate_snapshot": snapshot,
    }


def _build_decision(
    run_id: str,
    policy_run_id: str,
    index: int,
    bundle: dict[str, Any],
    policy_bundle: dict[str, Any],
    decision_action: str,
    decision_reason_code: str,
    search_query_effect_status: str,
    triggered_blocking_rules: list[str],
    scorecard: dict[str, Any],
    score: Any,
    replay_marker: dict[str, Any],
) -> dict[str, Any]:
    content = bundle["content"]
    decision = {
        "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "decision_id": f"d_{index:03d}",
        "policy_bundle_id": policy_bundle["policy_bundle_id"],
        "rule_pack_id": policy_bundle["rule_pack_id"],
        "rule_pack_version": policy_bundle["rule_pack_version"],
        "strategy_id": policy_bundle["strategy_id"],
        "strategy_version": policy_bundle["strategy_version"],
        "policy_bundle_hash": policy_bundle.get("policy_bundle_hash"),
        "dispatch_id": policy_bundle.get("dispatch_id"),
        "decision_target_type": content["decision_target_type"],
        "decision_target_id": content["decision_target_id"],
        "triggered_blocking_rules": triggered_blocking_rules,
        "scorecard": scorecard,
        "score": score,
        "decision_action": decision_action,
        "decision_reason_code": decision_reason_code,
        "search_query_effect_status": search_query_effect_status,
        "source_evidence": bundle["source_evidence"],
        "decision_input_snapshot_id": bundle["run_context"]["decision_input_snapshot_id"],
        "decision_evidence_refs": _evidence_refs(policy_bundle["rule_pack"]),
        "decision_replay_data": {
            "policy_run_id": policy_run_id,
            "policy_bundle_id": policy_bundle["policy_bundle_id"],
            "policy_bundle_hash": policy_bundle.get("policy_bundle_hash"),
            "rule_package_id": policy_bundle.get("rule_package_id"),
            "rule_pack_id": policy_bundle["rule_pack_id"],
            "rule_pack_version": policy_bundle["rule_pack_version"],
            "dispatch_id": policy_bundle.get("dispatch_id"),
            "runtime_stage": policy_bundle.get("runtime_stage"),
            "strategy_version": policy_bundle["strategy_version"],
            "runtime_record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
            "decision_evidence_refs": _evidence_refs(policy_bundle["rule_pack"]),
            "missing_dimensions": [
                row["key"] for row in scorecard.get("dimensions", []) if row.get("score_missing")
            ],
            **replay_marker,
        },
    }
    return with_raw_payload(decision, exclude_keys=RULE_DECISION_RAW_PAYLOAD_EXCLUDE_KEYS)


def _evaluate_hard_gates(bundle: dict[str, Any], gates: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [gate for gate in gates if _condition_matches(bundle, gate.get("when", {}))]


def _primary_hard_gate(gates: list[dict[str, Any]]) -> dict[str, Any]:
    return sorted(gates, key=lambda gate: (gate.get("priority", 999), gate.get("gate_id", "")))[0]


def _condition_matches(bundle: dict[str, Any], condition: dict[str, Any]) -> bool:
    value = _get_path(bundle, condition.get("field", ""))
    op = condition.get("op")
    expected = condition.get("value")
    if op == "is_empty":
        return _is_empty(value)
    if op == "in":
        return value in (expected or [])
    if op == "not_in":
        return value not in (expected or [])
    if op == "eq":
        return value == expected
    if op == "gte":
        return value is not None and value >= expected
    if op == "lt":
        return value is not None and value < expected
    if op == "lte":
        return value is not None and value <= expected
    raise ValueError(f"unsupported rule operator: {op}")


def _match_threshold(score: int | float | None, thresholds: list[dict[str, Any]]) -> dict[str, Any] | None:
    if score is None:
        return None
    for threshold in thresholds:
        min_score = threshold.get("min_score")
        max_score = threshold.get("max_score")
        if min_score is not None and score < min_score:
            continue
        if max_score is not None and score > max_score:
            continue
        return threshold
    return None


def _get_path(data: dict[str, Any], path: str) -> Any:
    current: Any = data
    for part in path.split("."):
        if not part:
            return None
        if not isinstance(current, dict):
            return None
        current = current.get(part)
        if current is None:
            return None
    return current


def _is_empty(value: Any) -> bool:
    return value is None or value == "" or value == [] or value == {}


def _is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _threshold_label(threshold: dict[str, Any] | None) -> str:
    if not threshold:
        return "unknown"
    min_score = threshold.get("min_score")
    max_score = threshold.get("max_score")
    if min_score is not None and max_score is not None:
        return f"{min_score}<=score<={max_score}"
    if min_score is not None:
        return f"score>={min_score}"
    if max_score is not None:
        return f"score<={max_score}"
    return "unknown"


def _scorecard(score: Any, config: dict[str, Any], missing: bool = False) -> dict[str, Any]:
    return {
        "total_score": score,
        "dimensions": [
            {
                "key": dimension.get("key"),
                "max_score": dimension.get("max_score"),
                "score": None,
                "evidence_paths": dimension.get("evidence_paths", []),
            }
            for dimension in config.get("dimensions", [])
            if dimension.get("runtime_status") == "active"
        ],
        "matched_scoring_rules": [],
        "score_missing": missing,
    }


def _scorecard_total(bundle: dict[str, Any], config: dict[str, Any]) -> dict[str, Any]:
    dimensions = [
        dimension for dimension in config.get("dimensions", []) if dimension.get("runtime_status") == "active"
    ]
    rules = [rule for rule in config.get("scoring_rules", []) if rule.get("enabled", True)]
    if dimensions and not rules:
        raise ValueError("active scorecard dimensions require scorecard.scoring_rules")

    dimension_rows = []
    matched_rules = []
    total_score = 0
    for dimension in dimensions:
        score, rule = _score_dimension(bundle, dimension, rules)
        total_score += score
        if rule:
            matched_rules.append(rule["scoring_rule_id"])
        dimension_rows.append(
            {
                "key": dimension.get("key"),
                "max_score": dimension.get("max_score"),
                "score": score,
                "matched_rule_id": rule.get("scoring_rule_id") if rule else None,
                "score_missing": rule is None,
                "evidence_paths": dimension.get("evidence_paths", []),
            }
        )
    return {
        "total_score": total_score if dimensions else None,
        "dimensions": dimension_rows,
        "matched_scoring_rules": matched_rules,
        "score_missing": False,
    }


def _scorecard_has_any_evidence(scorecard: dict[str, Any]) -> bool:
    return any(not row.get("score_missing") for row in scorecard.get("dimensions", []))


def _scorecard_all_dimensions_missing(scorecard: dict[str, Any]) -> bool:
    return not _scorecard_has_any_evidence(scorecard)


def _score_dimension(
    bundle: dict[str, Any], dimension: dict[str, Any], rules: list[dict[str, Any]]
) -> tuple[int | float, dict[str, Any] | None]:
    dimension_rules = sorted(
        [rule for rule in rules if rule.get("dimension_key") == dimension.get("key")],
        key=lambda rule: (rule.get("priority", 999), rule.get("scoring_rule_id", "")),
    )
    if not dimension_rules:
        raise ValueError(f"active scorecard dimension lacks scoring rules: {dimension.get('key')}")
    for rule in dimension_rules:
        condition = {
            "field": rule.get("field_path", ""),
            "op": rule.get("operator"),
            "value": rule.get("expected_value"),
        }
        if _condition_matches(bundle, condition):
            return rule.get("score_value", 0), rule
    return 0, None


def _missing_score_decision(
    run_id: str,
    policy_run_id: str,
    index: int,
    bundle: dict[str, Any],
    policy_bundle: dict[str, Any],
    matched_gates: list[dict[str, Any]],
    scorecard: dict[str, Any],
) -> dict[str, Any]:
    missing_policy = policy_bundle["rule_pack"].get("scorecard", {}).get("score_missing_policy", {})
    decision_action = missing_policy.get("decision_action", "REJECT_CONTENT")
    decision_reason_code = missing_policy.get("decision_reason_code", "missing_score")
    effect = _effect_status_for_decision(
        policy_bundle, decision_action, SCORE_REASON_CATEGORY, is_hard_gate=False
    )
    return _build_decision(
        run_id=run_id,
        policy_run_id=policy_run_id,
        index=index,
        bundle=bundle,
        policy_bundle=policy_bundle,
        decision_action=decision_action,
        decision_reason_code=decision_reason_code,
        search_query_effect_status=effect["content_effect_status"],
        triggered_blocking_rules=[],
        scorecard={**scorecard, "score_missing": True},
        score=None,
        replay_marker={
            "matched_gates": [_gate_replay(gate) for gate in matched_gates],
            "score_missing_policy": missing_policy,
            "effect_mapping_id": effect.get("mapping_id"),
            "reason_category": SCORE_REASON_CATEGORY,
        },
    )


def _effect_status_for_decision(
    policy_bundle: dict[str, Any],
    decision_action: str,
    reason_category: str,
    is_hard_gate: bool,
) -> dict[str, Any]:
    candidates = [
        mapping
        for mapping in policy_bundle.get("effect_status_mapping", [])
        if mapping.get("enabled", True)
        and mapping.get("target_level") == "content"
        and mapping.get("decision_action") == decision_action
        and mapping.get("is_hard_gate") is is_hard_gate
    ]
    exact = [mapping for mapping in candidates if mapping.get("reason_category") == reason_category]
    matches = exact or candidates
    if (
        not matches
        and decision_action == V4_TECHNICAL_RETRY_ACTION
        and reason_category == TECHNICAL_REASON_CATEGORY
        and not is_hard_gate
    ):
        return {
            "mapping_id": "fallback_technical_retry_failed",
            "content_effect_status": "failed",
            "query_effect_status": "failed",
        }
    if not matches:
        raise ValueError(f"effect mapping not found for {decision_action}/{reason_category}")
    return sorted(matches, key=lambda mapping: (mapping.get("priority", 999), mapping.get("mapping_id", "")))[0]


def _reason_category_for_threshold(decision_action: str, decision_reason_code: str) -> str:
    if decision_reason_code == V4_TECHNICAL_RETRY_REASON:
        return TECHNICAL_REASON_CATEGORY
    if decision_action == "ADD_TO_CONTENT_POOL":
        return SUCCESS_REASON_CATEGORY
    if decision_action == "KEEP_CONTENT_FOR_REVIEW":
        return REVIEW_REASON_CATEGORY
    if decision_reason_code == "missing_score":
        return SCORE_REASON_CATEGORY
    return SCORE_REASON_CATEGORY


def _gate_replay(gate: dict[str, Any]) -> dict[str, Any]:
    return {
        "gate_id": gate.get("gate_id"),
        "decision_reason_code": gate.get("decision_reason_code"),
        "priority": gate.get("priority"),
        "severity": gate.get("severity"),
        "stop_scoring": gate.get("stop_scoring"),
    }


def _evidence_refs(rule_pack: dict[str, Any]) -> list[str]:
    refs = ["source_evidence"]
    for field in rule_pack.get("input_contract", {}).get("required_fields", []):
        if field not in refs:
            refs.append(field)
    return refs