#!/usr/bin/env python3
from __future__ import annotations

import argparse
import glob
import hashlib
import json
import math
import re
import shlex
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import parse_qs, unquote, urlsplit
from zoneinfo import ZoneInfo


DEFAULT_BACKEND_LOG = "/home/ubuntu/nexuscrm-deployment/online/envs/multi_prod/current/logs/backend.stderr.log"
REPORT_WINDOW_END_TIME = time(hour=15)
ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
TIMESTAMP_RE = re.compile(r"^(?P<ts>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)\b")
FIELD_RE = re.compile(
    r"\b(?P<key>ip|latency|method|path|request_id|status|trace_id)="
    r"(?P<value>.*?)(?=\s+(?:ip|latency|method|path|request_id|status|trace_id)=|$)"
)
STATIC_EXT_RE = re.compile(
    r"\.(?:css|js|map|png|jpg|jpeg|gif|svg|ico|webp|woff|woff2|ttf|eot)(?:\?|$)",
    re.IGNORECASE,
)
SALESFORCE_ID_RE = re.compile(r"\b[0-9A-Za-z]{15,18}\b")
QUOTED_STRING_RE = re.compile(r"'(?:''|[^'])*'")
NUMBER_RE = re.compile(r"\b\d+(?:\.\d+)?\b")
IN_LIST_RE = re.compile(r"\bIN\s*\((?:\s*\?,?\s*){4,}\)", re.IGNORECASE)

# The backend logs every search query it builds as a `DEBUG_SEARCH_SQL` line that
# carries both the user's raw terms (`params=[...]`) and the generated SQL
# (`sql="..."`). This is the only place the actual typed search term is visible,
# so the search-term profile parses these lines directly — the /query HTTP route
# only shows the resulting SOQL, not what the user typed.
SEARCH_SQL_LINE = "DEBUG_SEARCH_SQL"
PERMISSION_DIAGNOSTICS_LINE = "Permission diagnostics"
PERMISSION_PHASE_LINE = "Permission read phase"
PERMISSION_DURATION_FIELDS = (
    "perm_system_ms",
    "perm_custom_ms",
    "perm_recordtype_ms",
    "perm_object_db_ms",
    "perm_field_db_ms",
    "perm_object_bulk_ms",
    "perm_field_bulk_ms",
    "perm_effective_access_ms",
)
PERMISSION_REQUIRED_SURFACES = ("notifications", "permissions_effective", "ui_api_apps")
# 150 notification samples cover one five-minute window even at the low-end
# steady rate of 30 requests/minute; the admin/UI surfaces are intentionally sparse.
PERMISSION_MIN_SAMPLES = {"notifications": 150, "permissions_effective": 5, "ui_api_apps": 5}
PERMISSION_EFFECTIVE_PHASES = (
    "total",
    "system_permissions",
    "custom_permissions",
    "object_permissions",
    "field_permissions",
    "projection",
)
SEARCH_PARAMS_RE = re.compile(r"params=\s*\[(?P<params>.*?)\]\s+sql=")
SEARCH_SQL_RE = re.compile(r'sql=\s*"(?P<sql>.*)"\s*$')
PARAM_STRING_RE = re.compile(r'"((?:[^"\\]|\\.)*)"')
# Fields named in `<field> LIKE ?` clauses (backtick-quoted or bare).
LIKE_FIELD_RE = re.compile(r"`?(\w+)`?\s+LIKE", re.IGNORECASE)
# Salesforce 15/18-char record id: alphanumeric AND mixed (has a digit and a letter),
# which separates real ids from ordinary all-letter words of the same length.
SF_ID_TERM_RE = re.compile(r"[A-Za-z0-9]{15,18}")
EMAILISH_RE = re.compile(r"[^@\s]+@[^@\s]+")
PHONEISH_RE = re.compile(r"[\d\s+\-()]{2,}")
NON_ASCII_RE = re.compile(r"[^\x00-\x7f]")


def classify_search_term(term: str) -> str:
    """Bucket a raw search term by its *shape*, to reveal which list-view fields a
    multi-field OR-LIKE query can never match. A name typed into the box will never
    hit a Phone or OwnerId column, so those OR branches are pure full-scan waste —
    this classification is the evidence for trimming them."""
    term = term.strip().strip("%").strip()
    if not term:
        return "empty"
    if EMAILISH_RE.fullmatch(term):
        return "email"
    if PHONEISH_RE.fullmatch(term):
        return "phone/number"
    if SF_ID_TERM_RE.fullmatch(term) and any(c.isdigit() for c in term) and any(c.isalpha() for c in term):
        return "sf-id-like"
    if NON_ASCII_RE.search(term):
        return "non-ascii-text(CJK)"
    return "ascii-text"


def parse_search_sql_line(clean_line: str) -> dict[str, Any] | None:
    """Extract (first term, its type, queried field-set, wildcard shape) from a
    DEBUG_SEARCH_SQL line. Returns None when the line carries no usable term."""
    pm = SEARCH_PARAMS_RE.search(clean_line)
    if not pm:
        return None
    params = PARAM_STRING_RE.findall(pm.group("params"))
    if not params:
        return None
    raw = params[0]
    term_type = classify_search_term(raw)
    if term_type in ("empty",):
        return None
    sm = SEARCH_SQL_RE.search(clean_line)
    sql = sm.group("sql") if sm else ""
    fields = tuple(sorted({f for f in LIKE_FIELD_RE.findall(sql)}))
    # Leading wildcard (`%term`) defeats any index → full scan; trailing-only
    # (`term%`) can use one. The param value still has its wildcards here.
    leading_wildcard = raw.startswith("%")
    return {
        "term_type": term_type,
        "fields": fields,
        "leading_wildcard": leading_wildcard,
    }



@dataclass(frozen=True)
class ReportWindow:
    report_date: date
    timezone_name: str
    local_start: datetime
    local_end: datetime
    utc_start: datetime
    utc_end: datetime
    window_label: str


@dataclass(frozen=True)
class HttpRequest:
    timestamp: datetime
    method: str
    path: str
    status: int
    latency_ms: float | None
    ip: str
    request_id: str
    trace_id: str


@dataclass(frozen=True)
class PermissionDiagnostic:
    timestamp: datetime
    request_id: str
    route: str
    path: str
    status: int
    permission_ms: float
    field_bulk_calls: int
    field_bulk_rows: int
    field_bulk_ms: float
    field_bulk_rejected: int
    observed_fields: frozenset[str]


@dataclass(frozen=True)
class PermissionPhaseDiagnostic:
    timestamp: datetime
    correlation: str
    operation: str
    phase: str
    state: str
    outcome: str
    target_scope: str
    rows: int
    elapsed_ms: float
    observed_fields: frozenset[str]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate daily production backend slow request report from access logs.")
    parser.add_argument("--log-file", action="append", default=[], help="Explicit backend log file. Can be passed multiple times. Not mtime-filtered.")
    parser.add_argument(
        "--log-glob",
        action="append",
        default=[],
        help=(
            "Glob for per-release backend logs (e.g. '.../releases/*/logs/backend.stderr.log'). "
            "Backend logs are split per release, so a single day's window can span several release "
            "directories. Matches whose mtime predates the window start are skipped as stale; the rest "
            "are scanned and filtered line-by-line. Can be passed multiple times."
        ),
    )
    parser.add_argument("--source", choices=("backend", "prod-nginx"), default="backend")
    parser.add_argument("--nginx-log-glob", action="append", default=[])
    parser.add_argument("--prod-root", default="/home/ubuntu/nexuscrm-deployment/online/envs/multi_prod")
    parser.add_argument("--skip-container-diagnostics", action="store_true",
                        help="Offline fixture analysis only; marks diagnostic coverage unavailable.")
    parser.add_argument("--timezone", default="Asia/Shanghai")
    parser.add_argument(
        "--report-date",
        default="",
        help="Asia/Shanghai report window end date in YYYY-MM-DD. Defaults to the latest completed 15:00 window.",
    )
    parser.add_argument(
        "--window-start",
        default="",
        help="Explicit window start as local-time ISO 8601 (e.g. '2026-06-18T11:00' or "
             "'2026-06-18 11:00'), interpreted in --timezone. Use with --window-end for an "
             "arbitrary (hour-level) window instead of a full calendar day. Mutually exclusive "
             "with --report-date; both --window-start and --window-end must be given together.",
    )
    parser.add_argument(
        "--window-end",
        default="",
        help="Explicit window end as local-time ISO 8601, interpreted in --timezone. Exclusive "
             "upper bound (a request at exactly --window-end is not counted). See --window-start.",
    )
    parser.add_argument("--slow-threshold-ms", type=float, default=10000.0)
    parser.add_argument("--output-json", required=True)
    parser.add_argument("--output-markdown", required=True)
    parser.add_argument("--output-feishu-text", required=True)
    return parser.parse_args()


def _parse_local_dt(raw: str, tz: ZoneInfo) -> datetime:
    """Parse a local-time ISO 8601 string (date+time, 'T' or space separated).

    A bare date ('2026-06-18') is rejected here — callers that want a full day
    use --report-date. A naive datetime is attached to tz; an explicit offset is
    converted into tz so downstream UTC math stays correct.
    """
    text = raw.strip().replace(" ", "T")
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError as exc:
        raise SystemExit(
            f"invalid window timestamp {raw!r}: expected local ISO 8601 like "
            f"'2026-06-18T11:00' ({exc})"
        )
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=tz)
    return parsed.astimezone(tz)


def resolve_report_window(
    report_date_raw: str,
    timezone_name: str,
    window_start_raw: str = "",
    window_end_raw: str = "",
) -> ReportWindow:
    tz = ZoneInfo(timezone_name)

    has_window = bool(window_start_raw or window_end_raw)
    if has_window:
        if report_date_raw:
            raise SystemExit("--report-date is mutually exclusive with --window-start/--window-end")
        if not (window_start_raw and window_end_raw):
            raise SystemExit("--window-start and --window-end must be provided together")
        local_start = _parse_local_dt(window_start_raw, tz)
        local_end = _parse_local_dt(window_end_raw, tz)
        if local_end <= local_start:
            raise SystemExit(
                f"--window-end ({window_end_raw}) must be strictly after --window-start ({window_start_raw})"
            )
        # Label both windows so a comparison of two windows on the same calendar
        # day stays distinguishable; report_date carries the start date for any
        # date-keyed consumers.
        window_label = f"{local_start.isoformat()}→{local_end.isoformat()}"
        return ReportWindow(
            report_date=local_start.date(),
            timezone_name=timezone_name,
            local_start=local_start,
            local_end=local_end,
            utc_start=local_start.astimezone(timezone.utc),
            utc_end=local_end.astimezone(timezone.utc),
            window_label=window_label,
        )

    if report_date_raw:
        report_date = date.fromisoformat(report_date_raw)
    else:
        now = datetime.now(tz)
        report_date = now.date() if now.timetz() >= REPORT_WINDOW_END_TIME else (now - timedelta(days=1)).date()
    local_end = datetime.combine(report_date, REPORT_WINDOW_END_TIME, tzinfo=tz)
    local_start = local_end - timedelta(days=1)
    return ReportWindow(
        report_date=report_date,
        timezone_name=timezone_name,
        local_start=local_start,
        local_end=local_end,
        utc_start=local_start.astimezone(timezone.utc),
        utc_end=local_end.astimezone(timezone.utc),
        window_label=report_date.isoformat(),
    )


def strip_ansi(value: str) -> str:
    return ANSI_RE.sub("", value)


def parse_timestamp(line: str) -> datetime | None:
    match = TIMESTAMP_RE.search(line)
    if not match:
        return None
    raw = match.group("ts")
    return datetime.fromisoformat(raw.replace("Z", "+00:00"))


def parse_http_request_line(line: str) -> HttpRequest | None:
    clean = strip_ansi(line).strip()
    if "HTTP Request" not in clean:
        return None
    timestamp = parse_timestamp(clean)
    if timestamp is None:
        return None
    fields = {match.group("key"): match.group("value").strip() for match in FIELD_RE.finditer(clean)}
    required = ("latency", "method", "path", "status")
    if any(not fields.get(key) for key in required):
        return None
    try:
        latency_ms = float(fields["latency"])
        status = int(fields["status"])
    except ValueError:
        return None
    return HttpRequest(
        timestamp=timestamp,
        method=fields["method"],
        path=fields["path"],
        status=status,
        latency_ms=latency_ms,
        ip=fields.get("ip", ""),
        request_id=fields.get("request_id", ""),
        trace_id=fields.get("trace_id", ""),
    )


def _structured_log_fields(clean_line: str) -> dict[str, Any]:
    if clean_line.startswith("{"):
        try:
            value = json.loads(clean_line)
        except json.JSONDecodeError:
            return {}
        return value if isinstance(value, dict) else {}

    fields: dict[str, Any] = {}
    try:
        tokens = shlex.split(clean_line)
    except ValueError:
        return fields
    for token in tokens:
        if "=" not in token:
            continue
        key, value = token.split("=", 1)
        fields[key] = value
    return fields


def _log_int(fields: dict[str, Any], key: str) -> tuple[int, bool]:
    if key not in fields:
        return 0, False
    try:
        return int(fields[key]), True
    except (TypeError, ValueError):
        return 0, False


def _log_float(fields: dict[str, Any], key: str) -> tuple[float, bool]:
    if key not in fields:
        return 0.0, False
    try:
        value = float(fields[key])
        return value, math.isfinite(value)
    except (TypeError, ValueError):
        return 0.0, False


def permission_phase_correlation(request_id: str) -> str:
    bounded = request_id.strip()[:128]
    if not bounded:
        return ""
    return hashlib.sha256(bounded.encode("utf-8")).hexdigest()[:16]


def parse_permission_diagnostic_line(line: str) -> PermissionDiagnostic | None:
    clean = strip_ansi(line).strip()
    if PERMISSION_DIAGNOSTICS_LINE not in clean:
        return None
    fields = _structured_log_fields(clean)
    timestamp = parse_timestamp(clean)
    if timestamp is None:
        raw_timestamp = str(fields.get("time", fields.get("timestamp", ""))).strip()
        try:
            timestamp = datetime.fromisoformat(raw_timestamp.replace("Z", "+00:00"))
        except ValueError:
            return None
    observed_fields: set[str] = set()
    duration_values: list[float] = []
    for key in PERMISSION_DURATION_FIELDS:
        value, valid = _log_float(fields, key)
        if valid and value >= 0:
            observed_fields.add(key)
            duration_values.append(value)
    status, status_valid = _log_int(fields, "status")
    field_bulk_calls, calls_valid = _log_int(fields, "perm_field_bulk_calls")
    field_bulk_rows, rows_valid = _log_int(fields, "perm_field_bulk_rows")
    field_bulk_ms, bulk_ms_valid = _log_float(fields, "perm_field_bulk_ms")
    field_bulk_rejected, rejected_valid = _log_int(fields, "perm_field_bulk_rejected")
    for key, value, valid in (
        ("status", status, status_valid),
        ("perm_field_bulk_calls", field_bulk_calls, calls_valid),
        ("perm_field_bulk_rows", field_bulk_rows, rows_valid),
        ("perm_field_bulk_ms", field_bulk_ms, bulk_ms_valid),
        ("perm_field_bulk_rejected", field_bulk_rejected, rejected_valid),
    ):
        if valid and value >= 0:
            observed_fields.add(key)
    # Some diagnostics wrap others, so summing them can double-count time. The
    # longest observed component is only a peak-component signal; HTTP and
    # explicit effective-permission total timings are evaluated separately.
    permission_ms = max(duration_values, default=0.0)
    return PermissionDiagnostic(
        timestamp=timestamp,
        request_id=str(fields.get("request_id", "")),
        route=str(fields.get("route", "")),
        path=str(fields.get("path", "")),
        status=status,
        permission_ms=permission_ms,
        field_bulk_calls=field_bulk_calls,
        field_bulk_rows=field_bulk_rows,
        field_bulk_ms=field_bulk_ms,
        field_bulk_rejected=field_bulk_rejected,
        observed_fields=frozenset(observed_fields),
    )


def parse_permission_phase_line(line: str) -> PermissionPhaseDiagnostic | None:
    clean = strip_ansi(line).strip()
    if PERMISSION_PHASE_LINE not in clean:
        return None
    fields = _structured_log_fields(clean)
    timestamp = parse_timestamp(clean)
    if timestamp is None:
        raw_timestamp = str(fields.get("time", fields.get("timestamp", ""))).strip()
        try:
            timestamp = datetime.fromisoformat(raw_timestamp.replace("Z", "+00:00"))
        except ValueError:
            return None
    rows, rows_valid = _log_int(fields, "rows")
    elapsed_ms, elapsed_valid = _log_float(fields, "elapsed_ms")
    observed_fields = {
        key
        for key in ("correlation", "operation", "phase", "state", "outcome", "target_scope")
        if str(fields.get(key, "")).strip()
    }
    if rows_valid and rows >= 0:
        observed_fields.add("rows")
    if elapsed_valid and elapsed_ms >= 0:
        observed_fields.add("elapsed_ms")
    return PermissionPhaseDiagnostic(
        timestamp=timestamp,
        correlation=str(fields.get("correlation", "")).strip(),
        operation=str(fields.get("operation", "")).strip(),
        phase=str(fields.get("phase", "")).strip(),
        state=str(fields.get("state", "")).strip(),
        outcome=str(fields.get("outcome", "")).strip(),
        target_scope=str(fields.get("target_scope", "")).strip(),
        rows=rows,
        elapsed_ms=elapsed_ms,
        observed_fields=frozenset(observed_fields),
    )


def iter_log_requests(log_files: Iterable[Path]) -> Iterable[tuple[HttpRequest | None, bool, str]]:
    for log_file in log_files:
        with log_file.open("r", encoding="utf-8", errors="ignore") as handle:
            for line in handle:
                is_http_line = "HTTP Request" in line
                request = parse_http_request_line(line) if is_http_line else None
                yield request, is_http_line, line


def route_path(path: str) -> str:
    return urlsplit(path).path or path


def normalize_route(path: str) -> str:
    clean_path = route_path(path)
    parts = [part for part in clean_path.split("/") if part]
    normalized: list[str] = []
    for index, part in enumerate(parts):
        previous = parts[index - 1] if index else ""
        if SALESFORCE_ID_RE.fullmatch(part):
            normalized.append(":id")
        elif previous == "sobjects" and index + 1 < len(parts):
            normalized.append(part)
        elif previous not in {"services", "data"} and index > 0 and SALESFORCE_ID_RE.search(part):
            normalized.append(":id")
        else:
            normalized.append(part)
    route = "/" + "/".join(normalized)
    route = re.sub(r"(/services/data/v[0-9.]+/sobjects/[^/]+)/[^/]+$", r"\1/:id", route)
    route = re.sub(r"(/services/data/v[0-9.]+/ui-api/(?:object-info|layouts|record-ui)/[^/]+)/[^/]+", r"\1/:id", route)
    return route


def request_category(path: str) -> str:
    clean_path = route_path(path)
    if clean_path in {"/health", "/healthz", "/ready", "/readyz", "/livez", "/version.json", "/__invalid_request__"}:
        return "ignored"
    if STATIC_EXT_RE.search(clean_path):
        return "ignored"
    if clean_path == "/product/notifications" or (clean_path.startswith("/services/data/") and "/connect/notifications" in clean_path):
        return "background"
    if clean_path.startswith(("/webhooks/", "/hubspot/workflow-actions/")):
        return "integration"
    return "interactive"


def permission_surface(path: str) -> str:
    clean_path = route_path(path)
    if clean_path.endswith("/product/notifications") or clean_path.endswith("/connect/notifications"):
        return "notifications"
    if clean_path.endswith("/permissions/effective"):
        return "permissions_effective"
    if clean_path.endswith("/ui-api/apps"):
        return "ui_api_apps"
    return "other"


def extract_soql(path: str) -> str:
    split = urlsplit(path)
    if not split.path.endswith("/query"):
        return ""
    values = parse_qs(split.query, keep_blank_values=True).get("q", [])
    if not values:
        return ""
    return unquote(values[0]).strip()


def normalize_soql(soql: str) -> str:
    normalized = unquote(soql)
    normalized = QUOTED_STRING_RE.sub("'?'", normalized)
    normalized = SALESFORCE_ID_RE.sub("?", normalized)
    normalized = NUMBER_RE.sub("?", normalized)
    normalized = re.sub(r"\s+", " ", normalized).strip()
    normalized = IN_LIST_RE.sub("IN (...)", normalized)
    return normalized


def compute_percentile(values: list[float], ratio: float) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    index = max(0, math.ceil(ratio * len(ordered)) - 1)
    return float(ordered[index])


def status_bucket(status: int) -> str:
    return f"{status // 100}xx" if status > 0 else "unknown"


# Latency distribution buckets. Lower bound is 5s (below the 10s slow threshold)
# so the card can also surface near-threshold requests as an early warning.
LATENCY_BUCKET_ORDER = ("5-10s", "10-30s", "30s+")


def latency_bucket(latency_ms: float) -> str | None:
    if latency_ms >= 30000:
        return "30s+"
    if latency_ms >= 10000:
        return "10-30s"
    if latency_ms >= 5000:
        return "5-10s"
    return None


def add_latency_metric(target: dict[str, Any], latencies: list[float]) -> None:
    target["avg_latency_ms"] = round(sum(latencies) / len(latencies), 2) if latencies else 0.0
    target["p50_latency_ms"] = round(compute_percentile(latencies, 0.50), 2)
    target["p95_latency_ms"] = round(compute_percentile(latencies, 0.95), 2)
    target["p99_latency_ms"] = round(compute_percentile(latencies, 0.99), 2)
    target["max_latency_ms"] = round(max(latencies), 2) if latencies else 0.0


def build_route_latency_profile(
    route_latencies: dict[str, list[float]],
    route_slow_counts: dict[str, int],
    route_methods: dict[str, Counter[str]],
    route_status_buckets: dict[str, Counter[str]],
    route_category: dict[str, str],
) -> list[dict[str, Any]]:
    """Build a per-route profile over ALL in-window traffic, not only slow requests.

    slow_rate here is the route's own slow share (route_slow / route_total), i.e. the
    per-route timeout rate. This is distinct from a route's share of all slow requests,
    and is what downstream comparison reports need to judge optimization effectiveness.
    """
    rows: list[dict[str, Any]] = []
    for route, latencies in route_latencies.items():
        request_count = sum(route_methods.get(route, Counter()).values())
        slow_count = int(route_slow_counts.get(route, 0))
        row: dict[str, Any] = {
            "route": route,
            "category": route_category.get(route, "interactive"),
            "request_count": request_count,
            "slow_count": slow_count,
            "slow_rate": round((slow_count / request_count) * 100, 2) if request_count else 0.0,
            "methods": dict(sorted(route_methods.get(route, Counter()).items())),
            "status_buckets": dict(sorted(route_status_buckets.get(route, Counter()).items())),
        }
        add_latency_metric(row, latencies)
        row["latency_sample_count"] = len(latencies)
        row["missing_latency_count"] = request_count - len(latencies)
        if row["missing_latency_count"]:
            for key in list(row):
                if key.endswith("_latency_ms") or key in {"slow_count", "slow_rate"}:
                    row[key] = None
        rows.append(row)
    return sorted(rows, key=lambda item: (-item["request_count"], -(item["p95_latency_ms"] or 0), item["route"]))


def build_search_term_profile(
    total: int,
    term_types: Counter[str],
    term_samples: dict[str, list[str]],
    fieldsets: Counter[tuple[str, ...]],
    leading_wildcard_count: int,
) -> dict[str, Any]:
    """Summarize what users actually type into search boxes, and how the queries
    scan. `by_input_type` answers "are people searching text, or emails/phones/ids?"
    — which is the evidence for trimming OR-LIKE branches that can never match a
    text term. `leading_wildcard_rate` is the share of searches whose `%term%`
    pattern defeats the index and forces a full scan."""
    by_input_type = [
        {
            "type": term_type,
            "count": count,
            "pct": round((count / total) * 100, 2) if total else 0.0,
            "samples": term_samples.get(term_type, [])[:5],
        }
        for term_type, count in term_types.most_common()
    ]
    top_fieldsets = [
        {"fields": list(fields), "field_count": len(fields), "count": count}
        for fields, count in fieldsets.most_common(12)
    ]
    return {
        "total_search_queries": total,
        "leading_wildcard_count": leading_wildcard_count,
        "leading_wildcard_rate": round((leading_wildcard_count / total) * 100, 2) if total else 0.0,
        "by_input_type": by_input_type,
        "top_field_sets": top_fieldsets,
    }


def build_group_rows(groups: dict[str, list[HttpRequest]], limit: int) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for name, requests in groups.items():
        latencies = [request.latency_ms for request in requests]
        row: dict[str, Any] = {
            "name": name,
            "count": len(requests),
            "status_counts": dict(sorted(Counter(str(request.status) for request in requests).items())),
            "methods": dict(sorted(Counter(request.method for request in requests).items())),
            "sample_request_id": next((request.request_id for request in requests if request.request_id), ""),
            "sample_trace_id": next((request.trace_id for request in requests if request.trace_id), ""),
        }
        add_latency_metric(row, latencies)
        rows.append(row)
    return sorted(rows, key=lambda item: (-item["count"], -item["p95_latency_ms"], item["name"]))[:limit]


def build_effective_permission_phase_profile(
    http_requests: list[HttpRequest],
    phase_diagnostics: list[PermissionPhaseDiagnostic],
) -> dict[str, Any]:
    effective_requests = [
        request
        for request in http_requests
        if request.method.upper() == "GET"
        and permission_surface(request.path) == "permissions_effective"
        and 200 <= request.status < 300
    ]
    known_effective_correlations = {
        permission_phase_correlation(request.request_id)
        for request in http_requests
        if request.method.upper() == "GET"
        and permission_surface(request.path) == "permissions_effective"
        and request.request_id
    }
    phases_by_correlation: dict[str, list[PermissionPhaseDiagnostic]] = defaultdict(list)
    for diagnostic in phase_diagnostics:
        if diagnostic.operation == "effective_permissions" and diagnostic.correlation:
            phases_by_correlation[diagnostic.correlation].append(diagnostic)

    missing_requests: list[str] = []
    incomplete_requests: list[str] = []
    non_ok: Counter[str] = Counter()
    phase_elapsed: dict[str, list[float]] = defaultdict(list)
    total_elapsed: list[float] = []
    matched_correlations: set[str] = set()

    for request in effective_requests:
        correlation = permission_phase_correlation(request.request_id)
        events = phases_by_correlation.get(correlation, [])
        if not correlation or not events:
            missing_requests.append(correlation or "missing-correlation")
            continue
        matched_correlations.add(correlation)
        target_scope = next((event.target_scope for event in events if event.target_scope), "")
        required_phases = list(PERMISSION_EFFECTIVE_PHASES)
        if target_scope == "other":
            required_phases[1:1] = ["target_authorization", "target_profile"]
        request_incomplete = target_scope not in {"self", "other"}

        for phase in required_phases:
            starts = [event for event in events if event.phase == phase and event.state == "start"]
            completes = [event for event in events if event.phase == phase and event.state == "complete"]
            if len(starts) != 1 or len(completes) != 1:
                request_incomplete = True
                continue
            complete = completes[0]
            if not {"outcome", "rows", "elapsed_ms"}.issubset(complete.observed_fields):
                request_incomplete = True
                continue
            phase_elapsed[phase].append(complete.elapsed_ms)
            if complete.outcome != "ok":
                non_ok[complete.outcome or "missing"] += 1
            if phase == "total":
                total_elapsed.append(complete.elapsed_ms)
        if request_incomplete:
            incomplete_requests.append(correlation)

    phase_rows: list[dict[str, Any]] = []
    for phase, values in sorted(phase_elapsed.items()):
        phase_rows.append(
            {
                "phase": phase,
                "sample_count": len(values),
                "p50_ms": round(compute_percentile(values, 0.50), 2),
                "p95_ms": round(compute_percentile(values, 0.95), 2),
                "p99_ms": round(compute_percentile(values, 0.99), 2),
                "max_ms": round(max(values), 2) if values else 0.0,
            }
        )

    # A 4xx effective-permission request can legitimately emit error phases after
    # authorization rejects it. It is known evidence, not an orphan and not part
    # of the successful-read latency cohort.
    orphan_correlations = set(phases_by_correlation) - known_effective_correlations
    orphan_non_ok: Counter[str] = Counter()
    for correlation in orphan_correlations:
        for event in phases_by_correlation[correlation]:
            if event.state == "complete" and event.outcome and event.outcome != "ok":
                orphan_non_ok[event.outcome] += 1

    return {
        "expected_request_count": len(effective_requests),
        "matched_request_count": len(matched_correlations),
        "missing_request_count": len(missing_requests),
        "incomplete_request_count": len(incomplete_requests),
        "non_ok_outcomes": dict(sorted(non_ok.items())),
        "orphan_non_ok_outcomes": dict(sorted(orphan_non_ok.items())),
        "total_p50_ms": round(compute_percentile(total_elapsed, 0.50), 2),
        "total_p95_ms": round(compute_percentile(total_elapsed, 0.95), 2),
        "total_p99_ms": round(compute_percentile(total_elapsed, 0.99), 2),
        "total_max_ms": round(max(total_elapsed), 2) if total_elapsed else 0.0,
        "phases": phase_rows,
        "orphan_correlation_count": len(orphan_correlations),
    }


def build_permission_latency_profile(
    diagnostics: list[PermissionDiagnostic],
    http_requests: list[HttpRequest],
    phase_diagnostics: list[PermissionPhaseDiagnostic],
) -> dict[str, Any]:
    requests_by_id = {request.request_id: request for request in http_requests if request.request_id}
    http_by_surface: dict[str, list[HttpRequest]] = defaultdict(list)
    for request in http_requests:
        surface = permission_surface(request.path)
        if request.method.upper() == "GET" and surface in PERMISSION_REQUIRED_SURFACES:
            http_by_surface[surface].append(request)

    all_diagnostics_by_surface: dict[str, list[PermissionDiagnostic]] = defaultdict(list)
    successful_diagnostics_by_surface: dict[str, list[PermissionDiagnostic]] = defaultdict(list)
    diagnostic_without_http_by_surface: Counter[str] = Counter()
    correlations: list[dict[str, Any]] = []

    for diagnostic in diagnostics:
        route = diagnostic.route or diagnostic.path
        surface = permission_surface(route)
        all_diagnostics_by_surface[surface].append(diagnostic)
        request = requests_by_id.get(diagnostic.request_id) if diagnostic.request_id else None
        if request is None or permission_surface(request.path) != surface:
            diagnostic_without_http_by_surface[surface] += 1
            continue
        if not 200 <= request.status < 300:
            continue
        successful_diagnostics_by_surface[surface].append(diagnostic)
        correlations.append(
            {
                "request_id": diagnostic.request_id,
                "phase_correlation": permission_phase_correlation(diagnostic.request_id),
                "surface": surface,
                "route": normalize_route(request.path),
                "status": request.status,
                "permission_ms": round(diagnostic.permission_ms, 2),
                "field_bulk_ms": round(diagnostic.field_bulk_ms, 2),
                "http_latency_ms": round(request.latency_ms, 2),
                "http_minus_peak_permission_component_ms": round(max(0.0, request.latency_ms - diagnostic.permission_ms), 2),
            }
        )

    surface_rows: list[dict[str, Any]] = []
    missing_diagnostic_fields: dict[str, Counter[str]] = defaultdict(Counter)
    missing_diagnostic_for_http: Counter[str] = Counter()
    duplicate_diagnostics_for_http: Counter[str] = Counter()
    rejection_metric_available = any(
        "perm_field_bulk_rejected" in sample.observed_fields
        for surface_samples in successful_diagnostics_by_surface.values()
        for sample in surface_samples
    )
    surfaces = set(successful_diagnostics_by_surface) | set(http_by_surface)
    for surface in sorted(surfaces):
        samples = successful_diagnostics_by_surface.get(surface, [])
        surface_http = http_by_surface.get(surface, [])
        successful_http = [request for request in surface_http if 200 <= request.status < 300]
        diagnostic_counts = Counter(sample.request_id for sample in samples if sample.request_id)
        for request in successful_http:
            count = diagnostic_counts.get(request.request_id, 0) if request.request_id else 0
            if count == 0:
                missing_diagnostic_for_http[surface] += 1
            elif count > 1:
                duplicate_diagnostics_for_http[surface] += count - 1
        permission_values = [sample.permission_ms for sample in samples]
        field_values = [
            sample.field_bulk_ms
            for sample in samples
            if sample.field_bulk_calls > 0 and "perm_field_bulk_ms" in sample.observed_fields
        ]
        row_values = [
            float(sample.field_bulk_rows)
            for sample in samples
            if sample.field_bulk_calls > 0 and "perm_field_bulk_rows" in sample.observed_fields
        ]
        required_fields = set(PERMISSION_DURATION_FIELDS)
        if rejection_metric_available:
            required_fields.add("perm_field_bulk_rejected")
        if surface == "notifications":
            required_fields.update({"perm_field_bulk_calls", "perm_field_bulk_rows"})
        for sample in samples:
            for field in sorted(required_fields - set(sample.observed_fields)):
                missing_diagnostic_fields[surface][field] += 1
        http_values = [request.latency_ms for request in successful_http]
        surface_rows.append(
            {
                "surface": surface,
                "sample_count": len(samples),
                "projection_sample_count": len(field_values),
                "http_request_count": len(surface_http),
                "successful_http_count": len(successful_http),
                "correlated_http_count": len(samples),
                "http_without_diagnostic_count": missing_diagnostic_for_http[surface],
                "excluded_non_success_http_count": len(surface_http) - len(successful_http),
                "field_bulk_calls": sum(sample.field_bulk_calls for sample in samples),
                "field_bulk_rejected": sum(sample.field_bulk_rejected for sample in samples),
                "field_bulk_rejected_available": rejection_metric_available,
                "field_bulk_rows_p99": round(compute_percentile(row_values, 0.99), 2),
                "field_bulk_rows_max": int(max(row_values)) if row_values else 0,
                "permission_p50_ms": round(compute_percentile(permission_values, 0.50), 2),
                "permission_p95_ms": round(compute_percentile(permission_values, 0.95), 2),
                "permission_p99_ms": round(compute_percentile(permission_values, 0.99), 2),
                "permission_max_ms": round(max(permission_values), 2) if permission_values else 0.0,
                "permission_30s_count": sum(value >= 30000 for value in permission_values),
                "http_p50_ms": round(compute_percentile(http_values, 0.50), 2),
                "http_p95_ms": round(compute_percentile(http_values, 0.95), 2),
                "http_p99_ms": round(compute_percentile(http_values, 0.99), 2),
                "http_max_ms": round(max(http_values), 2) if http_values else 0.0,
                "http_499_or_5xx_count": sum(request.status == 499 or request.status >= 500 for request in surface_http),
                "http_30s_count": sum(request.latency_ms >= 30000 for request in surface_http),
            }
        )

    rows_by_surface = {row["surface"]: row for row in surface_rows}
    successful_diagnostics = [
        diagnostic
        for surface_samples in successful_diagnostics_by_surface.values()
        for diagnostic in surface_samples
    ]
    phase_profile = build_effective_permission_phase_profile(http_requests, phase_diagnostics)
    failures: list[str] = []
    for surface in PERMISSION_REQUIRED_SURFACES:
        row = rows_by_surface.get(surface)
        if row is None:
            continue
        if row["permission_p95_ms"] >= 500:
            failures.append(f"{surface}: peak permission component p95 is not below 500ms")
        if row["permission_p99_ms"] >= 1000:
            failures.append(f"{surface}: peak permission component p99 is not below 1000ms")
        if row["http_p95_ms"] >= 500:
            failures.append(f"{surface}: successful HTTP p95 is not below 500ms")
        if row["http_p99_ms"] >= 1000:
            failures.append(f"{surface}: successful HTTP p99 is not below 1000ms")
        if row["field_bulk_rejected_available"] and row["field_bulk_rejected"] > 0:
            failures.append(f"{surface}: bounded field projection rejection observed")
        if row["permission_30s_count"] > 0:
            failures.append(f"{surface}: permission read reached 30s")
        if row["http_499_or_5xx_count"] > 0 or row["http_30s_count"] > 0:
            failures.append(f"{surface}: required-route HTTP 499, 5xx, or 30s request observed")

    if phase_profile["non_ok_outcomes"] or phase_profile["orphan_non_ok_outcomes"]:
        failures.append("permissions_effective: non-ok permission phase outcome observed")
    if phase_profile["total_p95_ms"] >= 500:
        failures.append("permissions_effective: phase total p95 is not below 500ms")
    if phase_profile["total_p99_ms"] >= 1000:
        failures.append("permissions_effective: phase total p99 is not below 1000ms")

    missing_surfaces = [surface for surface in PERMISSION_REQUIRED_SURFACES if not successful_diagnostics_by_surface.get(surface)]
    missing_correlation = [
        surface
        for surface in PERMISSION_REQUIRED_SURFACES
        if diagnostic_without_http_by_surface[surface] > 0
        or duplicate_diagnostics_for_http[surface] > 0
        or missing_diagnostic_for_http[surface] > 0
    ]
    insufficient_samples = {
        surface: {"observed": len(successful_diagnostics_by_surface.get(surface, [])), "required": minimum}
        for surface, minimum in PERMISSION_MIN_SAMPLES.items()
        if len(successful_diagnostics_by_surface.get(surface, [])) < minimum
    }
    missing_fields = {
        surface: dict(sorted(counts.items()))
        for surface, counts in sorted(missing_diagnostic_fields.items())
        if counts
    }
    phase_evidence_incomplete = (
        phase_profile["orphan_correlation_count"] > 0
        or (
            phase_profile["expected_request_count"] > 0
            and (
                phase_profile["missing_request_count"] > 0
                or phase_profile["incomplete_request_count"] > 0
                or phase_profile["matched_request_count"] != phase_profile["expected_request_count"]
            )
        )
    )
    if failures:
        gate_status = "FAIL"
    elif missing_surfaces or missing_correlation or insufficient_samples or missing_fields or phase_evidence_incomplete:
        gate_status = "INCONCLUSIVE"
    else:
        gate_status = "PASS"

    return {
        "sample_count": len(diagnostics),
        "evaluated_sample_count": len(successful_diagnostics),
        "projection_sample_count": sum(sample.field_bulk_calls > 0 for sample in successful_diagnostics),
        "rejection_metric_available": rejection_metric_available,
        "correlated_http_count": len(correlations),
        "uncorrelated_count": sum(diagnostic_without_http_by_surface.values()),
        "surfaces": surface_rows,
        "effective_permission_phases": phase_profile,
        "http_correlation": sorted(correlations, key=lambda item: (-item["permission_ms"], item["request_id"]))[:30],
        "gate": {
            "status": gate_status,
            "scope": "backend_window_signal_only",
            "release_acceptance": False,
            "thresholds": {
                "peak_permission_component_p95_ms_lt": 500,
                "peak_permission_component_p99_ms_lt": 1000,
                "successful_http_p95_ms_lt": 500,
                "successful_http_p99_ms_lt": 1000,
                "effective_permission_phase_total_p95_ms_lt": 500,
                "effective_permission_phase_total_p99_ms_lt": 1000,
                "permission_or_http_30s_count": 0,
                "http_499_or_5xx_count": 0,
                "field_bulk_rejected": 0,
            },
            "required_surfaces": list(PERMISSION_REQUIRED_SURFACES),
            "minimum_samples": dict(PERMISSION_MIN_SAMPLES),
            "missing_surfaces": missing_surfaces,
            "missing_http_correlation": missing_correlation,
            "missing_permission_diagnostic_for_http": {
                surface: count for surface, count in sorted(missing_diagnostic_for_http.items()) if count
            },
            "diagnostic_without_http": {
                surface: count for surface, count in sorted(diagnostic_without_http_by_surface.items()) if count
            },
            "duplicate_permission_diagnostics": {
                surface: count for surface, count in sorted(duplicate_diagnostics_for_http.items()) if count
            },
            "insufficient_samples": insufficient_samples,
            "missing_diagnostic_fields": missing_fields,
            "phase_evidence_incomplete": phase_evidence_incomplete,
            "failures": failures,
            "zero_samples_are_inconclusive": True,
            "external_release_evidence_required": [
                "exact deployed SHA and UTC window",
                "two or three consecutive controlled five-minute windows including forced cold load",
                "nginx notification and UI 499/502 counts",
                "TiDB statement history with Execute >=10s count",
            ],
        },
    }


def truncate(value: str, limit: int = 180) -> str:
    if len(value) <= limit:
        return value
    return value[: limit - 3] + "..."


def build_report(log_files: list[Path], window: ReportWindow, slow_threshold_ms: float,
                 *, entries: Iterable[tuple[HttpRequest | None, bool, str]] | None = None) -> dict[str, Any]:
    scanned_lines = 0
    http_lines = 0
    parsed_requests = 0
    malformed_http_lines = 0
    total_by_category: Counter[str] = Counter()
    slow_by_category: Counter[str] = Counter()
    slow_methods: Counter[str] = Counter()
    slow_statuses: Counter[str] = Counter()
    slow_status_buckets: Counter[str] = Counter()
    slow_latency_buckets: Counter[str] = Counter()
    slow_routes: dict[str, list[HttpRequest]] = defaultdict(list)
    slow_soql: dict[str, list[tuple[HttpRequest, str]]] = defaultdict(list)
    slow_samples: list[HttpRequest] = []
    # Full-traffic per-route accumulators (every in-window request, not just slow ones).
    # These power route_latency_profile, which downstream comparison reports consume to
    # compute per-route timeout/slow rates against the route's own total volume.
    route_latencies: dict[str, list[float]] = defaultdict(list)
    route_slow_counts: Counter[str] = Counter()
    route_methods: dict[str, Counter[str]] = defaultdict(Counter)
    route_status_buckets: dict[str, Counter[str]] = defaultdict(Counter)
    route_category: dict[str, str] = {}
    all_request_latencies: list[float] = []
    # Search-term accumulators (DEBUG_SEARCH_SQL lines, in-window). Keyed by term
    # shape and by the field-set the query OR-LIKEs over, so the report can show
    # both "what users actually type" and "which field-sets are full-scanning".
    search_term_types: Counter[str] = Counter()
    search_term_samples: dict[str, list[str]] = defaultdict(list)
    search_fieldsets: Counter[tuple[str, ...]] = Counter()
    search_leading_wildcard = 0
    search_total = 0
    permission_diagnostics: list[PermissionDiagnostic] = []
    permission_phase_diagnostics: list[PermissionPhaseDiagnostic] = []
    http_requests: list[HttpRequest] = []

    for request, is_http_line, raw_line in (iter_log_requests(log_files) if entries is None else entries):
        scanned_lines += 1
        if not is_http_line and PERMISSION_DIAGNOSTICS_LINE in raw_line:
            diagnostic = parse_permission_diagnostic_line(raw_line)
            if diagnostic is not None and window.utc_start <= diagnostic.timestamp < window.utc_end:
                permission_diagnostics.append(diagnostic)
            continue
        if not is_http_line and PERMISSION_PHASE_LINE in raw_line:
            phase_diagnostic = parse_permission_phase_line(raw_line)
            if phase_diagnostic is not None and window.utc_start <= phase_diagnostic.timestamp < window.utc_end:
                permission_phase_diagnostics.append(phase_diagnostic)
            continue
        if not is_http_line and SEARCH_SQL_LINE in raw_line:
            clean = strip_ansi(raw_line).strip()
            ts = parse_timestamp(clean)
            if ts is not None and window.utc_start <= ts < window.utc_end:
                parsed = parse_search_sql_line(clean)
                if parsed is not None:
                    search_total += 1
                    search_term_types[parsed["term_type"]] += 1
                    if len(search_term_samples[parsed["term_type"]]) < 8:
                        pm = SEARCH_PARAMS_RE.search(clean)
                        terms = PARAM_STRING_RE.findall(pm.group("params")) if pm else []
                        if terms:
                            search_term_samples[parsed["term_type"]].append(terms[0].strip("%"))
                    if parsed["fields"]:
                        search_fieldsets[parsed["fields"]] += 1
                    if parsed["leading_wildcard"]:
                        search_leading_wildcard += 1
            continue
        if is_http_line:
            http_lines += 1
        if request is None:
            if is_http_line:
                malformed_http_lines += 1
            continue
        if not (window.utc_start <= request.timestamp < window.utc_end):
            continue

        parsed_requests += 1
        http_requests.append(request)
        category = request_category(request.path)
        total_by_category[category] += 1
        route = normalize_route(request.path)
        route_latencies.setdefault(route, [])
        route_methods[route][request.method] += 1
        route_status_buckets[route][status_bucket(request.status)] += 1
        route_category.setdefault(route, category)
        if request.latency_ms is None:
            continue
        all_request_latencies.append(request.latency_ms)
        route_latencies[route].append(request.latency_ms)
        bucket = latency_bucket(request.latency_ms)
        if bucket is not None:
            slow_latency_buckets[bucket] += 1
        if request.latency_ms < slow_threshold_ms:
            continue

        slow_by_category[category] += 1
        slow_methods[request.method] += 1
        slow_statuses[str(request.status)] += 1
        slow_status_buckets[status_bucket(request.status)] += 1
        route_slow_counts[route] += 1
        slow_routes[route].append(request)
        soql = extract_soql(request.path)
        if soql:
            slow_soql[normalize_soql(soql)].append((request, soql))
        slow_samples.append(request)

    all_slow_latencies = [request.latency_ms for requests in slow_routes.values() for request in requests]
    interactive_slow_count = slow_by_category["interactive"]
    parsed_total = sum(total_by_category.values())
    slow_total = sum(slow_by_category.values())
    slow_rate = round((slow_total / parsed_total) * 100, 2) if parsed_total else 0.0
    interactive_total = total_by_category["interactive"]
    interactive_slow_rate = round((interactive_slow_count / interactive_total) * 100, 2) if interactive_total else 0.0

    slow_soql_rows: list[dict[str, Any]] = []
    for normalized, items in slow_soql.items():
        requests = [item[0] for item in items]
        raw_soql = next((item[1] for item in items if item[1]), "")
        row: dict[str, Any] = {
            "normalized_soql": normalized,
            "sample_soql": raw_soql,
            "count": len(items),
            "status_counts": dict(sorted(Counter(str(request.status) for request in requests).items())),
            "sample_request_id": next((request.request_id for request in requests if request.request_id), ""),
            "sample_trace_id": next((request.trace_id for request in requests if request.trace_id), ""),
        }
        add_latency_metric(row, [request.latency_ms for request in requests])
        slow_soql_rows.append(row)

    slow_samples = sorted(slow_samples, key=lambda item: item.latency_ms, reverse=True)[:20]
    summary: dict[str, Any] = {
        "scanned_lines": scanned_lines,
        "http_lines": http_lines,
        "malformed_http_lines": malformed_http_lines,
        "parsed_requests_in_window": parsed_requests,
        "slow_threshold_ms": slow_threshold_ms,
        "slow_request_count": slow_total,
        "slow_request_rate": slow_rate,
        "interactive_request_count": interactive_total,
        "interactive_slow_request_count": interactive_slow_count,
        "interactive_slow_request_rate": interactive_slow_rate,
        "request_categories": dict(sorted(total_by_category.items())),
        "slow_request_categories": dict(sorted(slow_by_category.items())),
        "slow_methods": dict(sorted(slow_methods.items())),
        "slow_statuses": dict(sorted(slow_statuses.items())),
        "slow_status_buckets": dict(sorted(slow_status_buckets.items())),
        "slow_latency_buckets": {label: slow_latency_buckets.get(label, 0) for label in LATENCY_BUCKET_ORDER},
    }
    # Ignored health/static traffic cannot establish business-route coverage.
    # Empty/unrecognized inputs and missing durations are also inconclusive.
    summary["window_data_warning"] = (
        total_by_category["interactive"] + total_by_category["integration"] == 0
        or malformed_http_lines > 0
    )
    summary["latency_sample_count"] = len(all_request_latencies)
    summary["missing_latency_count"] = parsed_requests - len(all_request_latencies)
    summary["latency_available"] = parsed_requests > 0 and summary["missing_latency_count"] == 0
    summary["first_request_at"] = min((r.timestamp for r in http_requests), default=None)
    summary["last_request_at"] = max((r.timestamp for r in http_requests), default=None)
    for key in ("first_request_at", "last_request_at"):
        if summary[key] is not None:
            summary[key] = summary[key].isoformat()
    summary["status_buckets"] = dict(sorted(Counter(status_bucket(r.status) for r in http_requests).items()))
    summary["status_counts"] = dict(sorted(Counter(str(r.status) for r in http_requests).items()))
    add_latency_metric(summary, all_slow_latencies)
    # Full-traffic latency percentiles (all in-window requests, slow or not). These are the
    # baseline the comparison report diffs day-over-day; keep keys distinct from the slow-only
    # avg/p50/p95/p99/max above.
    summary["all_request_avg_ms"] = round(sum(all_request_latencies) / len(all_request_latencies), 2) if all_request_latencies else 0.0
    summary["all_request_p50_ms"] = round(compute_percentile(all_request_latencies, 0.50), 2)
    summary["all_request_p95_ms"] = round(compute_percentile(all_request_latencies, 0.95), 2)
    summary["all_request_p99_ms"] = round(compute_percentile(all_request_latencies, 0.99), 2)
    summary["all_request_max_ms"] = round(max(all_request_latencies), 2) if all_request_latencies else 0.0

    route_latency_profile = build_route_latency_profile(
        route_latencies,
        route_slow_counts,
        route_methods,
        route_status_buckets,
        route_category,
    )

    search_term_profile = build_search_term_profile(
        search_total,
        search_term_types,
        search_term_samples,
        search_fieldsets,
        search_leading_wildcard,
    )
    permission_latency_profile = build_permission_latency_profile(
        permission_diagnostics,
        [r for r in http_requests if r.latency_ms is not None],
        permission_phase_diagnostics,
    )

    if not summary["latency_available"]:
        summary["window_data_warning"] = True
        summary["observed_slow_request_count"] = slow_total
        for key in list(summary):
            if (key.startswith("all_request_") or key.endswith("_latency_ms")
                    or key in {"slow_request_count", "slow_request_rate",
                               "interactive_slow_request_count", "interactive_slow_request_rate"}):
                summary[key] = None
        if permission_latency_profile["gate"]["status"] == "PASS":
            permission_latency_profile["gate"]["status"] = "INCONCLUSIVE"
        permission_latency_profile["gate"]["http_latency_incomplete"] = True

    return {
        "report_date": window.report_date.isoformat(),
        "window_label": window.window_label,
        "timezone": window.timezone_name,
        "local_window_start": window.local_start.isoformat(),
        "local_window_end": window.local_end.isoformat(),
        "utc_window_start": window.utc_start.isoformat().replace("+00:00", "Z"),
        "utc_window_end": window.utc_end.isoformat().replace("+00:00", "Z"),
        "log_files": [str(path) for path in log_files],
        "summary": summary,
        "route_latency_profile": route_latency_profile,
        "search_term_profile": search_term_profile,
        "permission_latency_profile": permission_latency_profile,
        "slow_routes": build_group_rows(slow_routes, 30),
        "slow_soql": sorted(slow_soql_rows, key=lambda item: (-item["count"], -item["p95_latency_ms"], item["normalized_soql"]))[:30],
        "slow_samples": [
            {
                "timestamp": request.timestamp.isoformat().replace("+00:00", "Z"),
                "method": request.method,
                "status": request.status,
                "latency_ms": round(request.latency_ms, 2),
                "route": normalize_route(request.path),
                "path": request.path,
                "category": request_category(request.path),
                "request_id": request.request_id,
                "trace_id": request.trace_id,
            }
            for request in slow_samples
        ],
        "notes": [
            "Source is backend business access log, not nginx or TiDB slow log.",
            "Current backend access log does not include user id or request body.",
            "latency is interpreted as milliseconds based on current production samples.",
            "Permission latency is the longest recorded permission component, not a sum, because component timings can overlap.",
            "Permission HTTP correlations include the same bounded request hash emitted as correlation by effective-permission phase logs.",
            "Permission budgets are component, effective-phase total, and successful required-route HTTP P95 < 500 ms and P99 < 1 s, with zero 30 s reads or HTTP 499/5xx; a produced bounded-field rejection is also a failure.",
            "Permission backend-window PASS requires the configured minimum samples, bidirectional request-id correlation for successful requests, complete diagnostic fields, and closed effective-permission phase evidence; zero or incomplete evidence is INCONCLUSIVE.",
            "field_bulk_rows/calls are request-level aggregate trend signals and do not prove the loaded object or an exact Notification field contract.",
            "perm_field_bulk_rejected is optional for runtimes that do not produce it; availability is explicit and missing values render as n/a rather than zero.",
            "Permission backend-window PASS is not release acceptance; nginx 499/502, TiDB Execute >=10s, exact deployed SHA, forced-cold load, and consecutive controlled windows remain separate evidence.",
        ],
    }


def format_latency(value: float | None) -> str:
    return "n/a" if value is None else f"{value:.2f} ms"


def format_rate(value: float | None) -> str:
    return "n/a" if value is None else f"{value:.2f}%"


def table_rows(items: list[dict[str, Any]], columns: list[tuple[str, str]]) -> list[str]:
    lines = ["| " + " | ".join(label for label, _ in columns) + " |"]
    lines.append("| " + " | ".join("---" if index == 0 else "---:" for index, _ in enumerate(columns)) + " |")
    for item in items:
        values = [str(item.get(key, "")) for _, key in columns]
        lines.append("| " + " | ".join(values) + " |")
    return lines


def build_markdown(report: dict[str, Any]) -> str:
    summary = report["summary"]
    collection = report.get("log_collection", {})
    lines = [
        "## Daily Production Backend Slow Request Report",
        "",
        f"- report_date: `{report['report_date']}`",
        f"- window: `{report.get('window_label', report['report_date'])}`",
        f"- timezone: `{report['timezone']}`",
        f"- local_window: `{report['local_window_start']}` -> `{report['local_window_end']}`",
        f"- utc_window: `{report['utc_window_start']}` -> `{report['utc_window_end']}`",
        f"- slow_threshold: `{format_latency(summary['slow_threshold_ms'])}`",
        f"- log_files: `{', '.join(report['log_files'])}`",
    ]
    if collection:
        lines.append(
            f"- log_collection: scanned `{collection.get('scanned_files', 0)}` file(s), "
            f"skipped `{collection.get('skipped_stale_files', 0)}` stale (pre-window) file(s)"
        )
    coverage = report.get("source_coverage")
    if coverage:
        lines.extend(["", "### Source coverage", "",
                      f"- status: **{coverage['status']}**",
                      f"- completeness: {coverage['completeness']}",
                      f"- first/last request: {summary['first_request_at']} / {summary['last_request_at']}",
                      f"- measured/missing latency: {summary['latency_sample_count']} / {summary['missing_latency_count']}"])
        lines.extend(f"- {reason}" for reason in coverage["reasons"])
        lines.extend(["", "| Source | In window | First event | Last event |",
                      "| --- | ---: | --- | --- |"])
        for source in coverage["sources"]:
            lines.append(f"| `{source['path']}` | {source['in_window']} | {source['first_event_at']} | {source['last_event_at']} |")
        lines.append("See source_coverage in JSON for inode/container/image/restart and release identities.")
    if summary.get("window_data_warning"):
        lines.extend(
            [
                "",
                "> **WARNING: window data likely incomplete.** Sources have "
                "coverage gaps, missing business samples, missing latency, or parse failures. See "
                "source coverage diagnostics. Slow-request counts below "
                "are NOT a reliable health signal for this window.",
            ]
        )
    lines.extend(
        [
            "",
            "### Summary",
            "",
            "| Metric | Value |",
            "| --- | ---: |",
            f"| Parsed HTTP requests in window | {summary['parsed_requests_in_window']} |",
            f"| HTTP status buckets | {json.dumps(summary.get('status_buckets', {}))} |",
        f"| Slow requests | {summary['slow_request_count'] if summary['slow_request_count'] is not None else 'n/a'} |",
        f"| Slow request rate | {format_rate(summary['slow_request_rate'])} |",
        f"| Interactive requests | {summary['interactive_request_count']} |",
        f"| Interactive slow requests | {summary['interactive_slow_request_count'] if summary['interactive_slow_request_count'] is not None else 'n/a'} |",
        f"| Interactive slow rate | {format_rate(summary['interactive_slow_request_rate'])} |",
        f"| Slow avg latency | {format_latency(summary['avg_latency_ms'])} |",
        f"| Slow p95 latency | {format_latency(summary['p95_latency_ms'])} |",
        f"| Slow max latency | {format_latency(summary['max_latency_ms'])} |",
        f"| Malformed HTTP log lines | {summary['malformed_http_lines']} |",
        "",
        "### Slow Request Categories",
        "",
        "| Category | Requests |",
        "| --- | ---: |",
        ]
    )
    for category, count in summary["slow_request_categories"].items():
        lines.append(f"| {category} | {count} |")

    if report["slow_routes"]:
        lines.extend(["", "### Slow Routes", ""])
        rows: list[dict[str, Any]] = []
        for item in report["slow_routes"]:
            rows.append(
                {
                    "route": f"`{item['name']}`",
                    "count": item["count"],
                    "p95": format_latency(item["p95_latency_ms"]),
                    "max": format_latency(item["max_latency_ms"]),
                    "statuses": json.dumps(item["status_counts"], ensure_ascii=False),
                    "request_id": item["sample_request_id"],
                }
            )
        lines.extend(table_rows(rows, [("Route", "route"), ("Count", "count"), ("P95", "p95"), ("Max", "max"), ("Statuses", "statuses"), ("Sample Request", "request_id")]))

    if report.get("route_latency_profile"):
        lines.extend(["", "### Top Routes by Volume (all traffic)", ""])
        rows = []
        for item in report["route_latency_profile"][:20]:
            rows.append(
                {
                    "route": f"`{item['route']}`",
                    "count": item["request_count"],
                    "slow": item["slow_count"] if item["slow_count"] is not None else "n/a",
                    "slow_rate": format_rate(item["slow_rate"]),
                    "p95": format_latency(item["p95_latency_ms"]),
                    "p99": format_latency(item["p99_latency_ms"]),
                }
            )
        lines.extend(
            table_rows(
                rows,
                [
                    ("Route", "route"),
                    ("Count", "count"),
                    ("Slow", "slow"),
                    ("Slow Rate", "slow_rate"),
                    ("P95", "p95"),
                    ("P99", "p99"),
                ],
            )
        )

    permission_profile = report.get("permission_latency_profile") or {}
    permission_gate = permission_profile.get("gate") or {}
    if permission_profile:
        lines.extend(
            [
                "",
                "### Permission Read Evidence",
                "",
                f"- backend window signal: **{permission_gate.get('status', 'INCONCLUSIVE')}** (not release acceptance)",
                f"- samples: `{permission_profile.get('sample_count', 0)}`; "
                f"HTTP-correlated: `{permission_profile.get('correlated_http_count', 0)}`; "
                f"uncorrelated: `{permission_profile.get('uncorrelated_count', 0)}`",
                f"- missing surfaces: `{', '.join(permission_gate.get('missing_surfaces', [])) or 'none'}`",
                f"- missing HTTP correlation: `{', '.join(permission_gate.get('missing_http_correlation', [])) or 'none'}`",
                f"- insufficient samples: `{json.dumps(permission_gate.get('insufficient_samples', {}), sort_keys=True)}`",
                f"- missing diagnostic fields: `{json.dumps(permission_gate.get('missing_diagnostic_fields', {}), sort_keys=True)}`",
            ]
        )
        for failure in permission_gate.get("failures", []):
            lines.append(f"- failure: {failure}")
        if permission_profile.get("surfaces"):
            rows = []
            for item in permission_profile["surfaces"]:
                rows.append(
                    {
                        "surface": item["surface"],
                        "samples": item["sample_count"],
                        "correlated": item["correlated_http_count"],
                        "calls": item["field_bulk_calls"],
                        "rows_p99": item["field_bulk_rows_p99"],
                        "p99": format_latency(item["permission_p99_ms"]),
                        "http_p99": format_latency(item["http_p99_ms"]),
                        "max": format_latency(item["permission_max_ms"]),
                        "rejected": item["field_bulk_rejected"] if item["field_bulk_rejected_available"] else "n/a",
                        "http_bad": item["http_499_or_5xx_count"] + item["http_30s_count"],
                    }
                )
            lines.extend(
                [
                    "",
                    *table_rows(
                        rows,
                        [
                            ("Surface", "surface"),
                            ("Samples", "samples"),
                            ("HTTP", "correlated"),
                            ("Bulk Calls", "calls"),
                            ("Rows P99", "rows_p99"),
                            ("Permission Component P99", "p99"),
                            ("HTTP P99", "http_p99"),
                            ("Permission Component Max", "max"),
                            ("Rejected", "rejected"),
                            ("HTTP 499/5xx/30s", "http_bad"),
                        ],
                    ),
                ]
            )
        phase_profile = permission_profile.get("effective_permission_phases") or {}
        lines.extend(
            [
                "",
                "Effective-permission phase evidence: "
                f"matched `{phase_profile.get('matched_request_count', 0)}` / "
                f"expected `{phase_profile.get('expected_request_count', 0)}`; "
                f"incomplete `{phase_profile.get('incomplete_request_count', 0)}`; "
                f"total P95 `{format_latency(phase_profile.get('total_p95_ms', 0.0))}`; "
                f"total P99 `{format_latency(phase_profile.get('total_p99_ms', 0.0))}`.",
                "",
                "Release acceptance still requires exact deployed SHA, two or three controlled five-minute windows, nginx 499/502 evidence, and TiDB Execute >=10s evidence.",
            ]
        )

    profile = report.get("search_term_profile") or {}
    if profile.get("total_search_queries"):
        lines.extend(
            [
                "",
                "### Search Term Profile",
                "",
                f"Total search queries: **{profile['total_search_queries']}**  ·  "
                f"leading-wildcard (`%term%`, full-scan): "
                f"**{profile['leading_wildcard_count']}** "
                f"({format_rate(profile['leading_wildcard_rate'])})",
                "",
                "| Input type | Count | Share | Samples |",
                "| --- | ---: | ---: | --- |",
            ]
        )
        for row in profile.get("by_input_type", []):
            samples = ", ".join(row["samples"][:5])
            lines.append(f"| {row['type']} | {row['count']} | {format_rate(row['pct'])} | {samples} |")
        if profile.get("top_field_sets"):
            lines.extend(
                [
                    "",
                    "Field-sets queried (how many columns each search OR-LIKEs over):",
                    "",
                    "| Fields | # cols | Count |",
                    "| --- | ---: | ---: |",
                ]
            )
            for fs in profile["top_field_sets"]:
                lines.append(f"| {', '.join(fs['fields'])} | {fs['field_count']} | {fs['count']} |")

    if report["slow_soql"]:
        lines.extend(["", "### Slow SOQL", ""])
        rows = []
        for item in report["slow_soql"]:
            rows.append(
                {
                    "soql": f"`{truncate(item['normalized_soql'], 140)}`",
                    "count": item["count"],
                    "p95": format_latency(item["p95_latency_ms"]),
                    "max": format_latency(item["max_latency_ms"]),
                    "statuses": json.dumps(item["status_counts"], ensure_ascii=False),
                    "request_id": item["sample_request_id"],
                }
            )
        lines.extend(table_rows(rows, [("SOQL Pattern", "soql"), ("Count", "count"), ("P95", "p95"), ("Max", "max"), ("Statuses", "statuses"), ("Sample Request", "request_id")]))

    if report["slow_samples"]:
        lines.extend(["", "### Slowest Samples", ""])
        rows = []
        for item in report["slow_samples"][:10]:
            rows.append(
                {
                    "time": item["timestamp"],
                    "method": item["method"],
                    "status": item["status"],
                    "latency": format_latency(item["latency_ms"]),
                    "route": f"`{item['route']}`",
                    "category": item["category"],
                    "request_id": item["request_id"],
                }
            )
        lines.extend(table_rows(rows, [("Time", "time"), ("Method", "method"), ("Status", "status"), ("Latency", "latency"), ("Route", "route"), ("Category", "category"), ("Request", "request_id")]))

    lines.extend(["", "### Notes", ""])
    for note in report["notes"]:
        lines.append(f"- {note}")
    return "\n".join(lines) + "\n"


def build_feishu_text(report: dict[str, Any]) -> str:
    summary = report["summary"]
    lines = [
        "NexusCRM daily prod backend slow request report",
        f"date: {report['report_date']} ({report['timezone']})",
        f"threshold: {format_latency(summary['slow_threshold_ms'])}",
        f"slow_requests: total={summary['slow_request_count'] if summary['slow_request_count'] is not None else 'n/a'} rate={format_rate(summary['slow_request_rate'])}",
        f"interactive_slow: {summary['interactive_slow_request_count'] if summary['interactive_slow_request_count'] is not None else 'n/a'}/{summary['interactive_request_count']} rate={format_rate(summary['interactive_slow_request_rate'])}",
        f"slow_latency: p95={format_latency(summary['p95_latency_ms'])} max={format_latency(summary['max_latency_ms'])}",
    ]
    if summary.get("window_data_warning"):
        lines.insert(
            1,
            "WARNING: window coverage or latency incomplete; "
            "slow counts below are not a reliable health signal.",
        )
    coverage = report.get("source_coverage")
    if coverage:
        lines.append(f"source_coverage: {coverage['status']}; " + "; ".join(coverage["reasons"]))
        lines.append(f"first/last_request: {summary['first_request_at']} / {summary['last_request_at']}")
    if report["slow_routes"]:
        lines.append("top_slow_routes:")
        for item in report["slow_routes"][:5]:
            lines.append(
                f"- {item['name']}: count={item['count']} p95={format_latency(item['p95_latency_ms'])} max={format_latency(item['max_latency_ms'])}"
            )
    if report["slow_soql"]:
        lines.append("top_slow_soql:")
        for item in report["slow_soql"][:5]:
            lines.append(
                f"- count={item['count']} p95={format_latency(item['p95_latency_ms'])} {truncate(item['normalized_soql'], 120)}"
            )
    permission_profile = report.get("permission_latency_profile") or {}
    permission_gate = permission_profile.get("gate") or {}
    if permission_profile:
        lines.append(
            f"permission_backend_window: {permission_gate.get('status', 'INCONCLUSIVE')} release_acceptance=false "
            f"samples={permission_profile.get('sample_count', 0)} "
            f"correlated={permission_profile.get('correlated_http_count', 0)} "
            f"missing={','.join(permission_gate.get('missing_surfaces', [])) or 'none'}"
        )
        for item in permission_profile.get("surfaces", []):
            lines.append(
                f"- permission/{item['surface']}: samples={item['sample_count']} "
                f"p99={format_latency(item['permission_p99_ms'])} "
                f"http_p99={format_latency(item['http_p99_ms'])} "
                f"max={format_latency(item['permission_max_ms'])} "
                f"rows_p99={item['field_bulk_rows_p99']} "
                f"rejected={item['field_bulk_rejected'] if item['field_bulk_rejected_available'] else 'n/a'}"
            )
        phase_profile = permission_profile.get("effective_permission_phases") or {}
        lines.append(
            f"- permission/effective_phases: matched={phase_profile.get('matched_request_count', 0)}/"
            f"{phase_profile.get('expected_request_count', 0)} incomplete={phase_profile.get('incomplete_request_count', 0)} "
            f"total_p99={format_latency(phase_profile.get('total_p99_ms', 0.0))}"
        )
    profile = report.get("search_term_profile") or {}
    if profile.get("total_search_queries"):
        types = " ".join(
            f"{row['type']}={row['count']}({format_rate(row['pct'])})"
            for row in profile.get("by_input_type", [])
        )
        lines.append(
            f"search_terms: total={profile['total_search_queries']} "
            f"full_scan={profile['leading_wildcard_count']}({format_rate(profile['leading_wildcard_rate'])}) {types}"
        )
    return "\n".join(lines)


def ensure_parent(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def _split_raw_values(raw_values: list[str]) -> list[str]:
    values: list[str] = []
    for raw in raw_values:
        for part in raw.split(","):
            value = part.strip()
            if value:
                values.append(value)
    return values


def resolve_explicit_files(raw_values: list[str]) -> list[Path]:
    return [Path(value).expanduser().resolve() for value in _split_raw_values(raw_values)]


def resolve_glob_files(raw_globs: list[str], window: ReportWindow) -> tuple[list[Path], list[Path]]:
    """Expand release log globs, keeping only logs still written at/after the window start.

    Backend logs are split per release directory. A release whose log stopped being written
    before the window start (mtime < window start) was already superseded and cannot hold any
    in-window line, so it is skipped as stale. Everything from the window start onward is kept
    and later filtered line-by-line by timestamp.
    """
    window_start_epoch = window.utc_start.timestamp()
    selected: list[Path] = []
    skipped_stale: list[Path] = []
    for pattern in _split_raw_values(raw_globs):
        for match in sorted(glob.glob(Path(pattern).expanduser().as_posix())):
            path = Path(match).resolve()
            if not path.is_file():
                continue
            if path.stat().st_mtime >= window_start_epoch:
                selected.append(path)
            else:
                skipped_stale.append(path)
    return selected, skipped_stale


def dedupe_paths(paths: Iterable[Path]) -> list[Path]:
    seen: dict[Path, None] = {}
    for path in paths:
        seen.setdefault(path, None)
    return list(seen)


def main() -> int:
    args = parse_args()
    window = resolve_report_window(args.report_date, args.timezone, args.window_start, args.window_end)

    if args.source == "prod-nginx":
        from prod_http_sources import (DEFAULT_NGINX_GLOBS, coverage_evidence, discover_files,
                                       finish_coverage, prod_entries)
        if args.log_file or args.log_glob:
            raise ValueError("prod-nginx cannot mix backend HTTP logs into ingress counts")
        evidence = coverage_evidence(args.prod_root)
        log_files = discover_files(args.nginx_log_glob or DEFAULT_NGINX_GLOBS)
        if not log_files:
            raise FileNotFoundError("no Prod nginx ingress logs, including retained rotations")
        entries = prod_entries(log_files, window, evidence, not args.skip_container_diagnostics)
        report = build_report(log_files, window, args.slow_threshold_ms, entries=entries)
        finish_coverage(report, evidence, not args.skip_container_diagnostics)
    else:
        if args.nginx_log_glob or args.skip_container_diagnostics:
            raise ValueError("nginx source options require --source prod-nginx")
        explicit_files = resolve_explicit_files(args.log_file)
        missing = [str(path) for path in explicit_files if not path.is_file()]
        if missing:
            raise FileNotFoundError(f"missing backend log file(s): {', '.join(missing)}")
        glob_files, skipped_stale = resolve_glob_files(args.log_glob, window)
        if not explicit_files and not args.log_glob:
            explicit_files = resolve_explicit_files([DEFAULT_BACKEND_LOG])
        log_files = dedupe_paths([*explicit_files, *glob_files])
        if not log_files:
            raise FileNotFoundError("no backend log files resolved for window")
        report = build_report(log_files, window, args.slow_threshold_ms)
        report["log_collection"] = {
            "scanned_files": len(log_files),
            "skipped_stale_files": len(skipped_stale),
            "skipped_stale_paths": [str(path) for path in skipped_stale],
        }
    markdown = build_markdown(report)
    feishu_text = build_feishu_text(report)

    output_json = Path(args.output_json).expanduser().resolve()
    output_markdown = Path(args.output_markdown).expanduser().resolve()
    output_feishu = Path(args.output_feishu_text).expanduser().resolve()
    for path in (output_json, output_markdown, output_feishu):
        ensure_parent(path)
    output_json.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    output_markdown.write_text(markdown, encoding="utf-8")
    output_feishu.write_text(feishu_text + "\n", encoding="utf-8")
    return 0


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