content-find-agent-new/content_agent/business_modules/m6_acceptance_report.py

from __future__ import annotations

from collections import Counter
from typing import Any, Iterable

from content_agent.dashboard_service import _timeline_summary
from content_agent.interfaces import RuntimeStore


SCHEMA_VERSION = "v4_m6_acceptance_report.v1"

REPORT_INPUT_FILES = [
    "run_events.jsonl",
    "pattern_recall_evidence.jsonl",
    "rule_decisions.jsonl",
    "walk_actions.jsonl",
    "final_output.json",
    "strategy_review.json",
]


def build_report(
    run_id: str,
    runtime: RuntimeStore,
    *,
    platform: str | None = None,
    platform_mode: str | None = None,
) -> dict[str, Any]:
    run_events = _read_jsonl(runtime, run_id, "run_events.jsonl")
    evidence_rows = _read_jsonl(runtime, run_id, "pattern_recall_evidence.jsonl")
    decisions = _read_jsonl(runtime, run_id, "rule_decisions.jsonl")
    walk_actions = _read_jsonl(runtime, run_id, "walk_actions.jsonl")
    source_paths = _read_jsonl(runtime, run_id, "source_path_records.jsonl")
    final_output = _read_json(runtime, run_id, "final_output.json")
    strategy_review = _read_json(runtime, run_id, "strategy_review.json")

    return {
        "schema_version": SCHEMA_VERSION,
        "run_id": run_id,
        "policy_run_id": _policy_run_id(final_output, decisions, run_events),
        "platform": platform or _first_value(decisions, "platform"),
        "platform_mode": platform_mode,
        "generated_from": {
            "runtime_files": {
                filename: bool(_file_exists(runtime, run_id, filename))
                for filename in REPORT_INPUT_FILES
            }
        },
        "gemini_summary": _gemini_summary(run_events, evidence_rows),
        "performance": _performance_summary(run_events, walk_actions, source_paths),
        "score_calibration": _score_calibration(decisions),
        "observability": _observability_summary(decisions, walk_actions, final_output),
        "strategy_review_summary": _strategy_review_summary(strategy_review),
        "known_gaps": [
            {
                "gap_id": "tag_relevance_not_production",
                "description": (
                    "V4 plan mentions tag Gemini relevance, but current M4/M6 scope does "
                    "not implement it as production semantics."
                ),
                "severity": "planning_gap",
                "blocking_m6_acceptance": False,
            }
        ],
    }


def _read_json(runtime: RuntimeStore, run_id: str, filename: str) -> dict[str, Any]:
    try:
        return runtime.read_json(run_id, filename)
    except FileNotFoundError:
        return {}


def _read_jsonl(runtime: RuntimeStore, run_id: str, filename: str) -> list[dict[str, Any]]:
    try:
        return runtime.read_jsonl(run_id, filename)
    except FileNotFoundError:
        return []


def _file_exists(runtime: RuntimeStore, run_id: str, filename: str) -> bool:
    try:
        return bool(runtime.file_status(run_id).get(filename))
    except Exception:
        try:
            if filename.endswith(".jsonl"):
                return bool(runtime.read_jsonl(run_id, filename))
            return bool(runtime.read_json(run_id, filename))
        except Exception:
            return False


def _policy_run_id(
    final_output: dict[str, Any],
    decisions: list[dict[str, Any]],
    run_events: list[dict[str, Any]],
) -> str | None:
    return (
        final_output.get("policy_run_id")
        or _first_value(decisions, "policy_run_id")
        or _first_value(run_events, "policy_run_id")
    )


def _first_value(rows: list[dict[str, Any]], key: str) -> Any:
    for row in rows:
        value = row.get(key)
        if value not in (None, ""):
            return value
    return None


def _gemini_summary(
    run_events: list[dict[str, Any]],
    evidence_rows: list[dict[str, Any]],
) -> dict[str, Any]:
    summaries = [
        row.get("evidence_summary") or row.get("raw_payload") or {}
        for row in evidence_rows
    ]
    failure_type_counts: Counter[str] = Counter()
    http_status_counts: Counter[str] = Counter()
    retry_rescued_count = 0
    final_failed_count = 0
    successful_count = 0

    for summary in summaries:
        final_status = str(summary.get("final_status") or "")
        failure_type = summary.get("failure_type")
        if failure_type:
            failure_type_counts[str(failure_type)] += 1
        http_status = summary.get("http_status_code")
        if http_status not in (None, ""):
            http_status_counts[str(http_status)] += 1
        retry_count = _int_value(summary.get("retry_count"))
        if final_status in {"ok", "success"}:
            successful_count += 1
            if retry_count > 0:
                retry_rescued_count += 1
        elif final_status:
            final_failed_count += 1

    return {
        "used": len(summaries),
        "cap": None,
        "quota_exhausted": False,
        "successful_count": successful_count,
        "failure_type_counts": dict(failure_type_counts),
        "http_status_counts": dict(http_status_counts),
        "retry_rescued_count": retry_rescued_count,
        "final_failed_count": final_failed_count,
    }


def _performance_summary(
    run_events: list[dict[str, Any]],
    walk_actions: list[dict[str, Any]],
    source_paths: list[dict[str, Any]],
) -> dict[str, Any]:
    timeline = _timeline_summary(run_events, walk_actions, source_paths)
    return {
        "total_duration_ms": timeline.get("total_duration_ms"),
        "stage_duration_ms": timeline.get("stage_duration_ms", {}),
        "platform_query_failure_count": _platform_query_failure_count(run_events),
        "platform_rate_limited_count": timeline.get("platform_rate_limited_count", 0),
        "error_counts": timeline.get("error_counts", {}),
        "walk_status_counts": timeline.get("walk_status_counts", {}),
    }


def _platform_query_failure_count(run_events: list[dict[str, Any]]) -> int:
    return sum(1 for row in run_events if row.get("event_type") == "platform_query_failed")


def _score_calibration(decisions: list[dict[str, Any]]) -> dict[str, Any]:
    return {
        "decision_count": len(decisions),
        "score_ranges": {
            "query_relevance_score": _range_summary(
                (row.get("scorecard") or {}).get("query_relevance_score")
                for row in decisions
            ),
            "platform_performance_score": _range_summary(
                (row.get("scorecard") or {}).get("platform_performance_score")
                for row in decisions
            ),
            "score": _range_summary(row.get("score") for row in decisions),
        },
        "threshold_edge_samples": _threshold_edge_samples(decisions),
        "decision_action_counts": dict(Counter(row.get("decision_action") for row in decisions)),
    }


def _range_summary(values: Iterable[Any]) -> dict[str, float | None]:
    numeric = [
        float(value)
        for value in values
        if isinstance(value, (int, float)) and not isinstance(value, bool)
    ]
    if not numeric:
        return {"min": None, "max": None, "avg": None}
    return {
        "min": min(numeric),
        "max": max(numeric),
        "avg": round(sum(numeric) / len(numeric), 4),
    }


def _threshold_edge_samples(decisions: list[dict[str, Any]]) -> list[dict[str, Any]]:
    samples = []
    for row in decisions:
        score = row.get("score")
        if not isinstance(score, (int, float)) or isinstance(score, bool):
            continue
        if 53 <= score <= 57 or 68 <= score <= 72:
            samples.append(
                {
                    "decision_id": row.get("decision_id"),
                    "decision_target_id": row.get("decision_target_id"),
                    "score": score,
                    "decision_action": row.get("decision_action"),
                    "decision_reason_code": row.get("decision_reason_code"),
                }
            )
    return samples[:20]


def _observability_summary(
    decisions: list[dict[str, Any]],
    walk_actions: list[dict[str, Any]],
    final_output: dict[str, Any],
) -> dict[str, Any]:
    return {
        "missing_observable_fields": _missing_observable_fields(decisions),
        "allow_walk_distribution": _allow_walk_distribution(decisions),
        "walk_deny_reasons": _walk_deny_reasons(walk_actions),
        "author_asset_count": len(final_output.get("author_assets") or []),
    }


def _missing_observable_fields(decisions: list[dict[str, Any]]) -> list[dict[str, Any]]:
    counts: Counter[tuple[str, str]] = Counter()
    for row in decisions:
        for item in (row.get("scorecard") or {}).get("missing_observable_fields") or []:
            if isinstance(item, dict):
                field = str(item.get("field") or "unknown")
                reason = str(item.get("reason") or item.get("missing_type") or "unknown")
            else:
                field = str(item)
                reason = "unknown"
            counts[(field, reason)] += 1
    return [
        {"field": field, "reason": reason, "count": count}
        for (field, reason), count in sorted(counts.items())
    ]


def _allow_walk_distribution(decisions: list[dict[str, Any]]) -> dict[str, int]:
    counts = Counter()
    for row in decisions:
        replay_data = row.get("decision_replay_data") or {}
        if replay_data.get("allow_walk") is True:
            counts["allowed"] += 1
        elif replay_data.get("allow_walk") is False:
            counts["denied"] += 1
        else:
            counts["missing"] += 1
    return {
        "allowed": counts["allowed"],
        "denied": counts["denied"],
        "missing": counts["missing"],
    }


def _walk_deny_reasons(walk_actions: list[dict[str, Any]]) -> list[dict[str, Any]]:
    counts: Counter[str] = Counter()
    for row in walk_actions:
        if row.get("walk_status") not in {"skipped", "failed", "blocked"}:
            continue
        reason = (
            row.get("reason_code")
            or (row.get("raw_payload") or {}).get("walk_gate_reason_code")
            or "unknown"
        )
        counts[str(reason)] += 1
    return [
        {"reason_code": reason, "count": count}
        for reason, count in counts.most_common()
    ]


def _strategy_review_summary(strategy_review: dict[str, Any]) -> dict[str, Any]:
    v4_summary = strategy_review.get("v4_summary") or {}
    if not v4_summary and isinstance(strategy_review.get("raw_payload"), dict):
        v4_summary = strategy_review["raw_payload"].get("v4_summary") or {}
    return {
        "review_status": strategy_review.get("review_status", "not_generated"),
        "v4_summary_schema_version": v4_summary.get("schema_version"),
        "score_buckets": v4_summary.get("score_buckets", {}),
        "allow_walk_distribution": v4_summary.get("allow_walk_distribution", {}),
        "walk_gate_review": v4_summary.get("walk_gate_review", {}),
    }


def _int_value(value: Any) -> int:
    if isinstance(value, bool):
        return 0
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    try:
        return int(str(value))
    except (TypeError, ValueError):
        return 0