GTI_GERENCIADOR_FISCAL/fiscal/central/app/control_plane.py

"""Bounded control-panel queries backed by installation snapshots."""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
from urllib.parse import urlencode

from sqlalchemy import Numeric, and_, case, cast, func, or_, select

from .config import settings
from .events import list_events
from .installation_identity import compute_agent_health, compute_version_drift, outdated_versions, reference_agent_version
from .models import Agent, AgentEnrollment, AgentEvent, AgentSnapshot, AuditEvent, BlockedProduct, Installation, Job, Tenant
from .panel_presentation import health_reason_label, blocker_reason_label, blocker_type_label, fiscal_status_label, format_age, format_duration, format_ui_datetime, health_label, job_kind_label, job_status_label, operational_priority, priority_label


HEALTH_STATES = ("ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN")
BLOCKER_FIELDS = ("NCM", "CEST", "CFOP", "ST", "ICMS", "CSOSN", "PIS", "COFINS", "RTC", "OTHER")
ATTENTION_CANDIDATE_LIMIT = 10
INACTIVE_ENROLLMENT_STATES = ("REVOKED", "FAILED", "EXPIRED")


def operational_installation_predicate():
    """Only installations eligible for current operational indicators.

    Legacy Agents without an enrollment remain eligible. Disabling job
    processing is not the same as deactivating an installation.
    """
    inactive_enrollment = select(AgentEnrollment.id).where(
        AgentEnrollment.agent_id == Agent.id,
        AgentEnrollment.status.in_(INACTIVE_ENROLLMENT_STATES),
    ).exists()
    return and_(Tenant.active.is_(True), Agent.enabled.is_(True), ~inactive_enrollment)


def _page(page: int, page_size: int):
    return max(1, page), min(100, max(1, page_size))


def _iso(value):
    if value is None:
        return None
    value = value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
    return value.isoformat()


def _utc_now(value=None):
    value = value or datetime.now(timezone.utc)
    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value


def _health_expression(now):
    offline_after = settings.agent_offline_after_seconds or settings.agent_heartbeat_interval_seconds * 3
    cutoff = _utc_now(now) - timedelta(seconds=offline_after)
    return case(
        (Installation.last_seen_at.is_(None), "UNKNOWN"),
        (or_(AgentSnapshot.agent_health == "FAILED", Installation.last_status == "FAILED"), "FAILED"),
        (Installation.last_seen_at < cutoff, "OFFLINE"),
        (or_(AgentSnapshot.failed_products > 0, AgentSnapshot.last_job_status == "FAILED"), "DEGRADED"),
        else_="ONLINE",
    )


def _fiscal_completion_expression():
    total = func.coalesce(AgentSnapshot.total_products, 0)
    complete = func.coalesce(AgentSnapshot.fiscally_complete, 0)
    numeric_total = cast(total, Numeric(18, 6))
    numeric_complete = cast(complete * 100, Numeric(18, 6))
    raw = case((total > 0, numeric_complete / numeric_total), else_=cast(func.coalesce(AgentSnapshot.coverage_percent, 0), Numeric(18, 6)))
    return func.round(case((raw < 0, 0), (raw > 100, 100), else_=raw), 1)


def _fiscal_completion(snapshot) -> float:
    total = int(getattr(snapshot, "total_products", 0) or 0)
    complete = int(getattr(snapshot, "fiscally_complete", 0) or 0)
    if total:
        value = Decimal(max(0, min(total, complete)) * 100) / Decimal(total)
    else:
        value = Decimal(str(max(0, min(100, float(getattr(snapshot, "coverage_percent", 0) or 0)))))
    return float(value.quantize(Decimal("0.1"), rounding=ROUND_HALF_UP))


def _fiscal_status(snapshot) -> str:
    if int(getattr(snapshot, "safe_blocked", 0) or 0):
        return "BLOCKED"
    total = int(getattr(snapshot, "total_products", 0) or 0)
    complete = int(getattr(snapshot, "fiscally_complete", 0) or 0)
    if total and complete >= total:
        return "COMPLETE"
    return "INCOMPLETE"


def _coverage_partition(snapshot) -> dict:
    """Expose only explicit snapshot states and label the remaining count honestly."""
    total = int(getattr(snapshot, "total_products", 0) or 0)
    complete = int(getattr(snapshot, "fiscally_complete", 0) or 0)
    safe_blocked = int(getattr(snapshot, "safe_blocked", 0) or 0)
    failed = int(getattr(snapshot, "failed_products", 0) or 0)
    explicit = complete + safe_blocked + failed
    residual = total - explicit
    invalid_counts = min(total, complete, safe_blocked, failed) < 0
    overcount = max(0, explicit - total)
    return {
        "unclassified_products": max(0, residual) if not invalid_counts else 0,
        "coverage_overcount_products": overcount,
        "coverage_invalid_counts": invalid_counts,
        "coverage_invariant_pass": not invalid_counts and overcount == 0,
    }


def _technical_timestamp(item):
    if item["technical_health"] in {"FAILED", "DEGRADED"}:
        return item.get("last_job_at") or item.get("last_seen_at")
    return item.get("last_seen_at")


def _attention_item(item, problem, timestamp, *, rank, link=None, now=None):
    timestamp_value = datetime.fromisoformat(timestamp) if isinstance(timestamp, str) else timestamp
    return {
        "client": item.get("client"),
        "display_name": item.get("display_name"),
        "hostname": item.get("hostname"),
        "problem": problem,
        "timestamp": timestamp if isinstance(timestamp, str) else _iso(timestamp_value),
        "timestamp_display": format_ui_datetime(timestamp_value),
        "age": format_age(timestamp_value, now),
        "link": link or f"/agents/{item['agent_id']}",
        "technical_health": item["technical_health"],
        "rank": rank,
    }


def installation_view(installation, agent, tenant, snapshot=None, *, now=None, reference=None):
    has_snapshot = snapshot is not None
    snapshot = snapshot or AgentSnapshot(installation_id=installation.id)
    health, reason = compute_agent_health(installation, now=now, supplemental=snapshot)
    # Always recomputed against the current reference release: the value stored on the snapshot was taken at the
    # last heartbeat (against whatever was "current" then) and goes stale as soon as a newer release is published.
    version_drift = compute_version_drift(agent.agent_version, reference)
    fiscal_status = _fiscal_status(snapshot)
    coverage_partition = _coverage_partition(snapshot)
    if not has_snapshot:
        # No heartbeat/snapshot yet (e.g. an enrollment-only E2E row): expose
        # no counts at all rather than fabricated zeros — the UI shows
        # "Não informado" instead of None/0.
        coverage_partition = {"unclassified_products": None, "coverage_overcount_products": 0, "coverage_invalid_counts": False, "coverage_invariant_pass": True}
    version_drift_label = {"CURRENT": "Atualizada", "OUTDATED": "Desatualizada", "UNKNOWN": "Desconhecida"}.get(version_drift, str(version_drift))
    agent_version_display = f"{agent.agent_version} · {version_drift_label}" if agent.agent_version else "Desconhecida"
    return {
        "agent_id": agent.id,
        "installation_id": installation.installation_id,
        "installation_db_id": installation.id,
        "client": tenant.name if tenant else None,
        "tenant_id": installation.tenant_id,
        "display_name": installation.display_name,
        "hostname": installation.hostname,
        "company_name": installation.company_name,
        "company_code": installation.company_code,
        "company_cnpj": installation.company_cnpj_masked,
        "company_uf": installation.company_uf,
        "company_crt": installation.company_crt,
        "agent_version": agent.agent_version,
        "agent_version_display": agent_version_display,
        "version_drift": version_drift,
        "version_drift_label": version_drift_label,
        "health": health,
        "health_reason": reason,
        "technical_health": health,
        "health_label": health_label(health),
        "technical_health_label": health_label(health),
        "last_seen_at": _iso(installation.last_seen_at),
        "last_seen_display": format_ui_datetime(installation.last_seen_at),
        "last_seen_age": format_age(installation.last_seen_at, now),
        "started_at": _iso(installation.started_at),
        "last_status": installation.last_status,
        "total_products": snapshot.total_products if has_snapshot else None,
        "fiscally_complete": snapshot.fiscally_complete if has_snapshot else None,
        "safe_blocked": snapshot.safe_blocked if has_snapshot else None,
        "failed_products": snapshot.failed_products if has_snapshot else None,
        **coverage_partition,
        "coverage_percent": snapshot.coverage_percent if has_snapshot else None,
        "fiscal_completion_percent": _fiscal_completion(snapshot) if has_snapshot else None,
        "fiscal_status": fiscal_status if has_snapshot else "UNKNOWN",
        "fiscal_status_label": fiscal_status_label(fiscal_status) if has_snapshot else "Não informado",
        "blockers": snapshot.blockers or {},
        "last_job_status": snapshot.last_job_status,
        "last_job_at": _iso(snapshot.last_job_at),
        "last_successful_job_at": _iso(snapshot.last_successful_job_at),
        "firebird_version": installation.firebird_version,
        "firebird_ods": installation.firebird_ods,
        "database_identifier": installation.database_identifier,
        "capabilities": installation.capabilities or {},
        "fiscal_engine_version": installation.fiscal_engine_version,
        "dataset_versions": installation.dataset_versions or {},
        "technical_ids": {"agent_id": agent.id, "installation_db_id": installation.id, "tenant_id": installation.tenant_id},
        "link": f"/agents/{agent.id}",
    }


def _installation_query_filters(query, *, client=None, version=None, search=None, last_seen_before=None):
    if client:
        query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%")))
    if version and version != "outdated":
        query = query.where(Agent.agent_version == version)
    if search:
        query = query.where(or_(Installation.display_name.ilike(f"%{search}%"), Installation.hostname.ilike(f"%{search}%"), Tenant.name.ilike(f"%{search}%")))
    if last_seen_before:
        query = query.where(or_(Installation.last_seen_at.is_(None), Installation.last_seen_at < last_seen_before))
    return query


def list_installations(session, *, page=1, page_size=50, client=None, health=None, version=None, search=None, last_seen_before=None, sort="display_name", scope="active", now=None):
    page, page_size = _page(page, page_size)
    now = _utc_now(now)
    reference = reference_agent_version(session)
    query = select(Installation, Agent, Tenant, AgentSnapshot, operational_installation_predicate().label("operational_active")).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id)
    if scope == "active":
        query = query.where(operational_installation_predicate())
    elif scope != "all":
        raise ValueError("scope must be active or all")
    query = _installation_query_filters(query, client=client, version=version, search=search, last_seen_before=last_seen_before)
    if version == "outdated":
        query = query.where(Agent.agent_version.in_(outdated_versions(session, reference)))
    health_expression = _health_expression(now)
    if health:
        query = query.where(health_expression == health)
    if sort in {"worst_coverage", "coverage", "coverage_asc", "fiscal_completion_percent"}:
        ordered = query.order_by(_fiscal_completion_expression().asc(), Installation.display_name.asc(), Installation.id.asc())
    else:
        ordered = query.order_by(Installation.display_name.asc(), Installation.id.asc())
    rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all()
    items = [{**installation_view(i, a, t, s, now=now, reference=reference), "operational_active": bool(active)} for i, a, t, s, active in rows]
    count_query = select(func.count()).select_from(query.with_only_columns(Installation.id).order_by(None).subquery())
    total = session.scalar(count_query) or 0
    return {"page": page, "page_size": page_size, "items": items, "total": total, "health_filter_applied": health}


def _blocker_item(row, tenant, installation=None, agent=None, *, now=None):
    return {
        "id": row.id,
        "tenant_id": row.tenant_id,
        "client": tenant.name if tenant else row.company,
        "product_code": row.product_code,
        "description": row.description,
        "field": row.field,
        "blocker_type": row.blocker_type,
        "blocker_type_label": blocker_type_label(row.blocker_type),
        "reason": row.reason_code,
        "reason_code": row.reason_code,
        "reason_label": blocker_reason_label(row.reason_code),
        "evidence_summary": row.evidence_summary,
        "first_seen_at": _iso(row.first_seen_at),
        "last_seen_at": _iso(row.last_seen_at),
        "first_seen_display": format_ui_datetime(row.first_seen_at),
        "last_seen_display": format_ui_datetime(row.last_seen_at),
        "age": format_age(row.last_seen_at, now),
        "retry_count": row.retry_count,
        "resolved_at": _iso(row.resolved_at),
        "resolution_source": row.resolution_source,
        "priority": operational_priority(row),
        "priority_label": priority_label(operational_priority(row)),
        "installation_id": installation.installation_id if installation else None,
        "installation_db_id": installation.id if installation else None,
        "display_name": installation.display_name if installation else None,
        "hostname": installation.hostname if installation else None,
        "agent_id": agent.id if agent else None,
        "link": f"/blockers/{row.id}",
    }


def _parse_age(value):
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return timedelta(days=float(value))
    text = str(value).strip().lower()
    try:
        if text.endswith("h"):
            return timedelta(hours=float(text[:-1]))
        if text.endswith("d"):
            return timedelta(days=float(text[:-1]))
        return timedelta(days=float(text))
    except ValueError:
        return None


def _blockers_query(session, *, tenant_id=None, agent_id=None, product_code=None, blocker_type=None, field=None, reason_code=None, status="OPEN", client=None, installation=None, search=None, age=None, scope="active"):
    query = select(BlockedProduct, Tenant, Installation, Agent).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == BlockedProduct.installation_id).outerjoin(Agent, Agent.id == Installation.agent_id)
    if scope == "active":
        query = query.where(operational_installation_predicate(), Installation.id.is_not(None))
    elif scope != "all":
        raise ValueError("scope must be active or all")
    status = (status or "OPEN").upper()
    if status == "OPEN":
        query = query.where(BlockedProduct.resolved_at.is_(None))
    elif status in {"RESOLVED", "CLOSED"}:
        query = query.where(BlockedProduct.resolved_at.is_not(None))
    if tenant_id is not None:
        query = query.where(BlockedProduct.tenant_id == tenant_id)
    if agent_id is not None:
        query = query.where(Agent.id == agent_id)
    if product_code is not None:
        query = query.where(BlockedProduct.product_code == product_code)
    if blocker_type:
        query = query.where(BlockedProduct.blocker_type == blocker_type)
    if field:
        query = query.where(func.upper(BlockedProduct.field) == field.upper())
    if reason_code:
        query = query.where(BlockedProduct.reason_code == reason_code)
    if client:
        query = query.where(or_(Tenant.name.ilike(f"%{client}%"), BlockedProduct.company.ilike(f"%{client}%")))
    if installation:
        query = query.where(or_(Installation.installation_id.ilike(f"%{installation}%"), Installation.display_name.ilike(f"%{installation}%"), Installation.hostname.ilike(f"%{installation}%")))
    if search:
        term = f"%{search}%"
        query = query.where(or_(BlockedProduct.product_code.ilike(term), BlockedProduct.description.ilike(term), BlockedProduct.field.ilike(term), BlockedProduct.blocker_type.ilike(term), BlockedProduct.reason_code.ilike(term), Tenant.name.ilike(term)))
    age_delta = _parse_age(age)
    if age_delta:
        query = query.where(BlockedProduct.last_seen_at <= datetime.now(timezone.utc) - age_delta)
    return query


def blocker_filter_summary(session, *, tenant_id=None, agent_id=None, product_code=None, blocker_type=None, field=None, reason_code=None, status="OPEN", client=None, installation=None, age=None, search=None, scope="active"):
    query = _blockers_query(session, tenant_id=tenant_id, agent_id=agent_id, product_code=product_code, blocker_type=blocker_type, field=field, reason_code=reason_code, status=status, client=client, installation=installation, search=search, age=age, scope=scope)
    open_count = func.sum(case((BlockedProduct.resolved_at.is_(None), 1), else_=0))
    open_count, = session.execute(query.with_only_columns(open_count).order_by(None)).one()
    rows = session.execute(_blocker_breakdown_query(query)).all()
    breakdown = {field: 0 for field in BLOCKER_FIELDS}
    for field_name, count in rows:
        key = str(field_name or "").upper()
        breakdown[key if key in breakdown else "OTHER"] += int(count)
    pair_query = query.with_only_columns(BlockedProduct.tenant_id, BlockedProduct.product_code).distinct().subquery()
    client_query = query.with_only_columns(BlockedProduct.tenant_id).distinct().subquery()
    return {
        "open_blockers": int(open_count or 0),
        "affected_products": int(session.scalar(select(func.count()).select_from(pair_query)) or 0),
        "affected_clients": int(session.scalar(select(func.count()).select_from(client_query)) or 0),
        "blocker_breakdown": breakdown,
    }


def rtc_blocker_summary(session, agent_id: int) -> dict:
    """Read-only, installation-scoped RTC blocker aggregates for the cClassTrib page."""
    query = _blockers_query(session, agent_id=agent_id, field="RTC", status="OPEN", scope="all")
    blockers = int(session.scalar(query.with_only_columns(func.count(BlockedProduct.id)).order_by(None)) or 0)
    products = int(session.scalar(query.with_only_columns(func.count(func.distinct(BlockedProduct.product_code))).order_by(None)) or 0)
    reason = func.coalesce(BlockedProduct.evidence_summary["reason"].as_string(), BlockedProduct.reason_code)
    reasons = session.execute(
        query.with_only_columns(reason, func.count(func.distinct(BlockedProduct.product_code)))
        .group_by(reason).order_by(None)
    ).all()
    return {"open_blockers": blockers, "products_affected": products, "reason_counts": {str(code or ""): int(count) for code, count in reasons}}


def _blocker_breakdown_query(query):
    """Group on the source column itself for PostgreSQL-safe SQL semantics."""
    return query.with_only_columns(BlockedProduct.field, func.count(BlockedProduct.id)).group_by(BlockedProduct.field).order_by(None)


def list_blockers(session, *, tenant_id=None, agent_id=None, product_code=None, page=1, page_size=50, blocker_type=None, field=None, reason_code=None, resolved=False, client=None, installation=None, status="OPEN", age=None, search=None, scope="active"):
    page, page_size = _page(page, page_size)
    if resolved and status == "OPEN":
        status = "RESOLVED"
    query = _blockers_query(session, tenant_id=tenant_id, agent_id=agent_id, product_code=product_code, blocker_type=blocker_type, field=field, reason_code=reason_code, status=status, client=client, installation=installation, search=search, age=age, scope=scope)
    ordered = query.order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc())
    rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all()
    items = [_blocker_item(blocker, tenant, installation_row, agent, now=datetime.now(timezone.utc)) for blocker, tenant, installation_row, agent in rows]
    count_query = query.with_only_columns(func.count(BlockedProduct.id)).order_by(None)
    total = session.scalar(count_query) or 0
    return {"page": page, "page_size": page_size, "items": items, "total": total}


def _job_item(row, agent, installation, tenant, *, now=None):
    return {
        "job_id": row.id,
        "agent_id": row.agent_id,
        "client": tenant.name if tenant else None,
        "display_name": installation.display_name if installation else None,
        "hostname": installation.hostname if installation else None,
        "kind": row.kind,
        "kind_label": job_kind_label(row.kind),
        "status": row.status,
        "status_label": job_status_label(row.status),
        "created_at": _iso(row.created_at),
        "started_at": _iso(row.started_at),
        "finished_at": _iso(row.finished_at),
        "created_display": format_ui_datetime(row.created_at),
        "started_display": format_ui_datetime(row.started_at),
        "finished_display": format_ui_datetime(row.finished_at),
        "duration": format_duration(row.started_at, row.finished_at, now),
        "attempts": row.attempts,
        "result_summary": row.result_summary,
        "error": row.error,
    }


def list_jobs(session, *, agent_id=None, job_id=None, page=1, page_size=50, status=None, kind=None, since=None, until=None, client=None, search=None, scope="active"):
    page, page_size = _page(page, page_size)
    query = select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id)
    if scope == "active":
        query = query.where(operational_installation_predicate())
    elif scope != "all":
        raise ValueError("scope must be active or all")
    if agent_id is not None:
        query = query.where(Job.agent_id == agent_id)
    if job_id is not None:
        query = query.where(Job.id == job_id)
    if status:
        query = query.where(Job.status == status)
    if kind:
        query = query.where(Job.kind == kind)
    if since:
        query = query.where(Job.created_at >= since)
    if until:
        query = query.where(Job.created_at <= until)
    if client:
        query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%")))
    if search:
        term = f"%{search}%"
        query = query.where(or_(Tenant.name.ilike(term), Installation.display_name.ilike(term), Installation.hostname.ilike(term), Job.kind.ilike(term), Job.status.ilike(term)))
    ordered = query.order_by(Job.created_at.desc(), Job.id.desc())
    rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all()
    count_query = query.with_only_columns(func.count(Job.id)).order_by(None)
    total = session.scalar(count_query) or 0
    return {"page": page, "page_size": page_size, "items": [_job_item(row, agent, installation, tenant) for row, agent, installation, tenant in rows], "total": total}


def jobs_status_summary(session, *, scope="active"):
    """Bounded job counts for the current operational scope."""
    query = (select(Job.status, func.count(Job.id)).join(Agent, Agent.id == Job.agent_id)
             .join(Installation, Installation.agent_id == Agent.id)
             .join(Tenant, Tenant.id == Installation.tenant_id))
    if scope == "active":
        query = query.where(operational_installation_predicate())
    elif scope != "all":
        raise ValueError("scope must be active or all")
    counts = {status: 0 for status in ("PENDING", "RUNNING", "COMPLETED", "FAILED")}
    for status, count in session.execute(query.group_by(Job.status)):
        if status in counts:
            counts[status] = int(count)
    return counts


def dashboard_summary(session, now=None):
    now = _utc_now(now)
    reference = reference_agent_version(session)
    active = operational_installation_predicate()
    installation_query = select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(active)
    health_expression = _health_expression(now)
    health_rows = session.execute(
        select(health_expression.label("health"), func.count(Installation.id))
        .select_from(Installation)
        .join(Agent, Agent.id == Installation.agent_id)
        .join(Tenant, Tenant.id == Installation.tenant_id)
        .outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id)
        .where(active)
        .group_by(health_expression)
    ).all()
    counts = {state: 0 for state in HEALTH_STATES}
    for state, count in health_rows:
        counts[state] = int(count)

    snapshot_total = func.coalesce(AgentSnapshot.total_products, 0)
    snapshot_complete = func.coalesce(AgentSnapshot.fiscally_complete, 0)
    snapshot_blocked = func.coalesce(AgentSnapshot.safe_blocked, 0)
    snapshot_failed = func.coalesce(AgentSnapshot.failed_products, 0)
    snapshot_explicit = snapshot_complete + snapshot_blocked + snapshot_failed
    snapshot_has_negative = or_(snapshot_total < 0, snapshot_complete < 0, snapshot_blocked < 0, snapshot_failed < 0)
    total_row = session.execute(
        select(
            func.coalesce(func.sum(AgentSnapshot.total_products), 0),
            func.coalesce(func.sum(AgentSnapshot.fiscally_complete), 0),
            func.coalesce(func.sum(AgentSnapshot.safe_blocked), 0),
            func.coalesce(func.sum(AgentSnapshot.failed_products), 0),
            func.coalesce(func.sum(case((and_(~snapshot_has_negative, snapshot_explicit <= snapshot_total), snapshot_total - snapshot_explicit), else_=0)), 0),
            func.coalesce(func.sum(case((and_(~snapshot_has_negative, snapshot_explicit > snapshot_total), snapshot_explicit - snapshot_total), else_=0)), 0),
            func.coalesce(func.sum(case((snapshot_has_negative, 1), else_=0)), 0),
        ).select_from(Installation).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(active)
    ).one()
    totals = dict(zip(("total_products", "fiscally_complete", "safe_blocked", "failed_products", "unclassified_products", "coverage_overcount_products", "coverage_invalid_snapshots"), (int(value or 0) for value in total_row)))
    totals["coverage_invariant_pass"] = totals["coverage_overcount_products"] == 0 and totals["coverage_invalid_snapshots"] == 0

    open_filter = BlockedProduct.resolved_at.is_(None)
    active_blockers = (select(BlockedProduct).join(Installation, Installation.id == BlockedProduct.installation_id)
                       .join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == BlockedProduct.tenant_id)
                       .where(open_filter, active))
    open_blockers = session.scalar(active_blockers.with_only_columns(func.count(BlockedProduct.id)).order_by(None)) or 0
    breakdown_expression = func.coalesce(BlockedProduct.field, "OTHER")
    breakdown_rows = session.execute(active_blockers.with_only_columns(breakdown_expression, func.count(BlockedProduct.id)).group_by(breakdown_expression).order_by(None)).all()
    open_blocker_breakdown = {field: 0 for field in BLOCKER_FIELDS}
    for field, count in breakdown_rows:
        key = str(field or "").upper()
        open_blocker_breakdown[key if key in open_blocker_breakdown else "OTHER"] += int(count)
    product_pairs = active_blockers.with_only_columns(BlockedProduct.tenant_id, BlockedProduct.product_code).distinct().subquery()
    safe_blocked_products = session.scalar(select(func.count()).select_from(product_pairs)) or 0
    blocker_counts = {field: count for field, count in open_blocker_breakdown.items() if count}

    def installation_candidates(health):
        query = installation_query.where(health_expression == health).order_by(Installation.id.asc()).limit(ATTENTION_CANDIDATE_LIMIT)
        return [installation_view(i, a, t, s, now=now, reference=reference) for i, a, t, s in session.execute(query).all()]

    health_candidates = {health: installation_candidates(health) for health in ("FAILED", "DEGRADED", "OFFLINE", "UNKNOWN")}
    version_filter = Agent.agent_version.in_(outdated_versions(session, reference))
    outdated_query = installation_query.where(version_filter).order_by(Installation.id.asc()).limit(ATTENTION_CANDIDATE_LIMIT)
    outdated_agents = [installation_view(i, a, t, s, now=now, reference=reference) for i, a, t, s in session.execute(outdated_query).all()]

    group_field = func.upper(func.coalesce(BlockedProduct.field, "OTHER"))
    fiscal_priority = case(
        (func.upper(BlockedProduct.blocker_type).in_(("TECHNICAL_FAILURE", "CONNECTION_ERROR", "AGENT_OFFLINE", "JOB_FAILED")), 0),
        (func.upper(BlockedProduct.blocker_type).in_(("BLOCKED_MISSING_EVIDENCE", "MISSING_EVIDENCE")), 1),
        (group_field.in_(BLOCKER_FIELDS[:-1]), 2),
        else_=3,
    )
    blocker_groups = session.execute(
        active_blockers.with_only_columns(
            Tenant.name, Installation.display_name, Installation.hostname, Agent.id,
            group_field, BlockedProduct.blocker_type, BlockedProduct.reason_code,
            func.count(BlockedProduct.id), func.count(func.distinct(BlockedProduct.product_code)),
            func.max(BlockedProduct.last_seen_at),
        ).group_by(Tenant.name, Installation.id, Installation.display_name, Installation.hostname,
                   Agent.id, group_field, BlockedProduct.blocker_type, BlockedProduct.reason_code)
        .order_by(fiscal_priority.asc(), func.max(BlockedProduct.last_seen_at).desc(), Agent.id.asc())
        .limit(ATTENTION_CANDIDATE_LIMIT)
    ).all()

    attention = []
    for health in ("FAILED", "DEGRADED", "OFFLINE", "UNKNOWN"):
        rank = {"FAILED": 0, "DEGRADED": 1, "OFFLINE": 2, "UNKNOWN": 3}[health]
        attention.extend((_attention_item(item, health_reason_label(item["health_reason"]), _technical_timestamp(item), rank=rank, now=now), rank, item["agent_id"]) for item in health_candidates[health])
    for client_name, display_name, hostname, agent_id, field, blocker_type, reason, count, products, last_seen, in blocker_groups:
        item = {"client": client_name, "display_name": display_name, "hostname": hostname, "agent_id": agent_id, "technical_health": "BLOCKED"}
        link = "/blockers?" + urlencode({"scope": "active", "status": "OPEN", "agent_id": agent_id, "field": field, "blocker_type": blocker_type, "reason_code": reason})
        priority = operational_priority({"field": field, "blocker_type": blocker_type, "reason_code": reason})
        rank = {"CRITICAL": 3, "HIGH": 4, "NORMAL": 5, "LOW": 6}[priority]
        entry = _attention_item(item, f"{field}: {blocker_reason_label(reason)}", last_seen, rank=rank, link=link, now=now)
        entry.update(group_count=count, products_affected=products, field=field, priority=priority)
        attention.append((entry, rank, agent_id))
    for item in outdated_agents:
        attention.append((_attention_item(item, "Versão do agent desatualizada", item.get("last_seen_at"), rank=7, now=now), 7, item["agent_id"]))
    attention.sort(key=lambda entry: (entry[1], -(entry[0]["timestamp"] is not None), entry[2]))

    active_jobs = select(Job).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id).where(active)
    failed_jobs = session.scalar(active_jobs.with_only_columns(func.count(Job.id)).where(Job.status == "FAILED").order_by(None)) or 0
    failed_job_rows = session.execute(
        select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id)
        .join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id)
        .where(Job.status == "FAILED", active).order_by(Job.created_at.desc(), Job.id.desc()).limit(10)
    ).all()
    recent_failed_jobs = []
    for job, agent, installation, tenant in failed_job_rows:
        item = _job_item(job, agent, installation, tenant, now=now)
        item["error"] = (item["error"] or "")[:240]
        item["created_age"] = format_age(job.created_at, now)
        item["finished_age"] = format_age(job.finished_at, now)
        item["link"] = f"/jobs?job_id={job.id}"
        recent_failed_jobs.append(item)
    return {
        "agents_total": sum(counts.values()),
        **{key.lower(): value for key, value in counts.items()},
        "health_counts": counts,
        **totals,
        "blockers": blocker_counts,
        "stale_agents": health_candidates["OFFLINE"],
        "outdated_agents": outdated_agents,
        "safe_blocked_products": int(safe_blocked_products),
        "open_blockers": int(open_blockers),
        "open_blocker_breakdown": open_blocker_breakdown,
        "attention_items": [entry[0] for entry in attention[:10]],
        "failed_jobs": failed_jobs,
        "recent_failed_jobs": recent_failed_jobs,
    }


def blocker_detail(session, blocker_id):
    row = session.execute(_blockers_query(session, status="ALL", scope="all").where(BlockedProduct.id == blocker_id)).first()
    if not row:
        return None
    blocker, tenant, installation, agent = row
    item = _blocker_item(blocker, tenant, installation, agent)
    events = session.scalars(select(AgentEvent).where(AgentEvent.blocker_id == blocker_id).order_by(AgentEvent.occurred_at.desc(), AgentEvent.id.desc()).limit(100)).all()
    event_views = [{"id": event.id, "severity": event.severity, "event_type": event.event_type, "message": event.message, "context": event.structured_context, "job_id": event.job_id, "product_code": event.product_code, "occurred_at": _iso(event.occurred_at)} for event in events]
    job_ids = {event.job_id for event in events if event.job_id is not None}
    related_jobs = []
    if job_ids:
        job_query = select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id).where(Job.id.in_(job_ids)).order_by(Job.created_at.desc(), Job.id.desc()).limit(100)
        related_jobs = [_job_item(job, related_agent, related_installation, related_tenant) for job, related_agent, related_installation, related_tenant in session.execute(job_query).all()]
    audits = session.scalars(select(AuditEvent).where(AuditEvent.target_type == "blocker", AuditEvent.target_id == str(blocker_id)).order_by(AuditEvent.occurred_at.desc(), AuditEvent.id.desc()).limit(100)).all()
    audit_views = [{"id": audit.id, "admin_user_id": audit.admin_user_id, "action": audit.action, "target_id": audit.target_id, "context": audit.context, "severity": audit.severity, "occurred_at": _iso(audit.occurred_at)} for audit in audits]
    server = {
        "hostname": installation.hostname if installation else None,
        "company_name": installation.company_name if installation else None,
        "company_code": installation.company_code if installation else None,
        "company_uf": installation.company_uf if installation else None,
        "company_crt": installation.company_crt if installation else None,
        "firebird_version": installation.firebird_version if installation else None,
        "firebird_ods": installation.firebird_ods if installation else None,
        "database_identifier": installation.database_identifier if installation else None,
    }
    return {
        **item,
        "evidence": blocker.evidence_summary or {},
        "installation": {"installation_id": item["installation_id"], "installation_db_id": item["installation_db_id"], "display_name": item["display_name"], "hostname": item["hostname"], "agent_id": item["agent_id"]},
        "server": server,
        "related_events": event_views,
        "related_jobs": related_jobs,
        "resolution_audit": audit_views,
    }


def list_panel_events(session, installation_db_id, *, include_more=False, **filters):
    rows = list_events(session, installation_db_id, fetch_extra=include_more, **filters)
    return [{"id": row.id, "severity": row.severity, "event_type": row.event_type, "message": row.message, "context": row.structured_context, "job_id": row.job_id, "product_code": row.product_code, "occurred_at": _iso(row.occurred_at)} for row in rows]