"""Gateway Health & Diagnostics OTLP export runtime.

This exporter emits operator-owned gateway service-health metrics plus
narrow redacted diagnostic events. It is deliberately in-process and fail-open so
it works under systemd, launchd, s6, containers, tmux, nohup, or a simple shell
without a sidecar/watchdog dependency.
"""

from __future__ import annotations

import logging
import os
import re
import threading
from dataclasses import dataclass
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)

_DEFAULT_DIAGNOSTIC_SCOPE = "hermes.gateway.diagnostics"

_RESOURCE_ATTRIBUTE_KEYS = frozenset({
    "service.name",
    "service.namespace",
    "service.version",
    "service.instance.id",
    "deployment.environment.name",
    "cloud.provider",
    "cloud.platform",
    "cloud.region",
    "telemetry.scope",
})
_DIAGNOSTIC_ATTRIBUTE_KEYS = frozenset({
    "name",
    "subsystem",
    "error_class",
    "error_code",
    "platform",
    "old_state",
    "new_state",
    "version",
    "severity",
})
_SAFE_RESOURCE_VALUE = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$")


def _redact_string(raw: Any, *, limit: int = 500) -> str:
    try:
        from agent.monitoring.redaction import redact_for_export
        return (redact_for_export(str(raw or "")) or "[redacted]")[:limit]
    except Exception:
        return "[redaction-unavailable]"


def _safe_resource_attributes(raw: Any) -> Dict[str, str]:
    """Allowlist bounded resource labels and reject values changed by redaction."""
    attrs: Dict[str, str] = {}
    if not isinstance(raw, dict):
        return attrs
    for key, value in raw.items():
        key = str(key)
        if key not in _RESOURCE_ATTRIBUTE_KEYS or value is None:
            continue
        if key == "service.instance.id":
            from agent.monitoring.gateway_health import _safe_instance_id
            attrs[key] = _safe_instance_id(value)
            continue
        text = str(value)
        if not _SAFE_RESOURCE_VALUE.fullmatch(text):
            continue
        if _redact_string(text, limit=128) != text:
            continue
        attrs[key] = text
    return attrs


def _runtime_resource_attributes(
    config: Dict[str, Any], *, telemetry_scope: str
) -> Dict[str, str]:
    """Build the safe OTLP resource shared by metrics and diagnostic logs."""
    gh = _gateway_health_config(config)
    attrs = _safe_resource_attributes(gh.get("resource_attributes"))
    from agent.monitoring.gateway_health import _safe_instance_id

    attrs["service.name"] = "hermes-gateway"
    attrs["service.instance.id"] = _safe_instance_id(_install_id(config))
    attrs["telemetry.scope"] = telemetry_scope
    return attrs


def _diagnostic_log_attributes(event: Dict[str, Any]) -> Dict[str, Any]:
    attrs: Dict[str, Any] = {}
    for key in _DIAGNOSTIC_ATTRIBUTE_KEYS:
        value = event.get(key)
        if value is None:
            continue
        attrs[f"hermes.{key}"] = _redact_string(value) if isinstance(value, str) else value
    return attrs


@dataclass(slots=True)
class GatewayHealthExportRuntime:
    enabled: bool
    reason: str = "disabled"
    streamer: Any = None
    metric_provider: Any = None
    log_handler: Any = None
    log_streamer: Any = None
    thread: Optional[threading.Thread] = None
    stop_event: Optional[threading.Event] = None

    def shutdown(self) -> None:
        if self.stop_event is not None:
            self.stop_event.set()
        if self.thread is not None:
            self.thread.join(timeout=0.25)
        if self.log_handler is not None:
            try:
                logging.getLogger().removeHandler(self.log_handler)
            except Exception:
                pass

        # All producers above are now stopped. Drain queued and in-flight
        # events before detaching subscribers so the terminal lifecycle event
        # cannot race exporter shutdown. The barrier is bounded and fail-open.
        try:
            from agent.monitoring.emitter import get_emitter
            emitter = get_emitter()
            emitter.flush(timeout=1.0)
            if self.streamer is not None:
                emitter.unsubscribe(self.streamer)
            if self.log_streamer is not None:
                emitter.unsubscribe(self.log_streamer)
        except Exception:
            pass

        # Network flush/close runs under one bounded daemon-thread deadline and
        # can never delay gateway teardown indefinitely.
        closeables = [
            item for item in (self.streamer, self.log_streamer, self.metric_provider)
            if item is not None
        ]

        def _close() -> None:
            for item in closeables:
                try:
                    item.shutdown()
                except Exception:
                    pass

        if closeables:
            worker = threading.Thread(
                target=_close,
                name="hermes-gateway-health-export-shutdown",
                daemon=True,
            )
            worker.start()
            worker.join(timeout=2.0)

        self.streamer = None
        self.log_streamer = None
        self.metric_provider = None
        self.thread = None
        self.stop_event = None


def _gateway_health_config(config: Dict[str, Any]) -> Dict[str, Any]:
    mon = (config or {}).get("monitoring") or {}
    return mon.get("gateway_health_export") or {}


def _otlp_config(config: Dict[str, Any]) -> Dict[str, Any]:
    mon = (config or {}).get("monitoring") or {}
    export = mon.get("export") or {}
    return export.get("otlp") or {}


def _enabled(config: Dict[str, Any]) -> bool:
    gh = _gateway_health_config(config)
    otlp = _otlp_config(config)
    return bool(gh.get("enabled") and otlp.get("enabled") and otlp.get("endpoint"))


def _require_metrics_sdk(*, auto_install: bool = True, prompt: bool = False) -> Dict[str, Any]:
    if auto_install:
        try:
            from tools.lazy_deps import ensure as _lazy_ensure
            _lazy_ensure("export.otlp", prompt=prompt)
        except Exception:
            pass
    try:
        from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
        from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
        from opentelemetry.metrics import Observation
        from opentelemetry.trace import INVALID_SPAN_ID, INVALID_TRACE_ID, TraceFlags
        from opentelemetry._logs import LogRecord
        from opentelemetry._logs.severity import SeverityNumber
        from opentelemetry.sdk._logs import LoggerProvider
        from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
        from opentelemetry.sdk.metrics import MeterProvider
        from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
        from opentelemetry.sdk.resources import Resource
        return {
            "OTLPLogExporter": OTLPLogExporter,
            "OTLPMetricExporter": OTLPMetricExporter,
            "Observation": Observation,
            "LogRecord": LogRecord,
            "LoggerProvider": LoggerProvider,
            "INVALID_SPAN_ID": INVALID_SPAN_ID,
            "INVALID_TRACE_ID": INVALID_TRACE_ID,
            "TraceFlags": TraceFlags,
            "SeverityNumber": SeverityNumber,
            "BatchLogRecordProcessor": BatchLogRecordProcessor,
            "MeterProvider": MeterProvider,
            "PeriodicExportingMetricReader": PeriodicExportingMetricReader,
            "Resource": Resource,
        }
    except Exception as exc:
        raise RuntimeError(f"OTLP metrics SDK unavailable: {exc}") from exc


def _resolve_headers(headers_env: Optional[Dict[str, str]]) -> Dict[str, str]:
    resolved: Dict[str, str] = {}
    for header_name, env_name in (headers_env or {}).items():
        val = os.environ.get(str(env_name))
        if val:
            resolved[str(header_name)] = val
    return resolved


def _metric_endpoint(endpoint: str) -> str:
    if endpoint.endswith("/v1/traces"):
        return endpoint[: -len("/v1/traces")] + "/v1/metrics"
    return endpoint


def _logs_endpoint(endpoint: str) -> str:
    if endpoint.endswith("/v1/traces"):
        return endpoint[: -len("/v1/traces")] + "/v1/logs"
    if endpoint.endswith("/v1/metrics"):
        return endpoint[: -len("/v1/metrics")] + "/v1/logs"
    return endpoint


def _version() -> str:
    try:
        from hermes_cli import __version__
        return str(__version__)
    except Exception:
        return "unknown"


def _profile() -> str:
    try:
        from hermes_cli.profiles import get_active_profile_name
        return str(get_active_profile_name() or "default")
    except Exception:
        return "default"


def _install_id(config: Dict[str, Any]) -> str:
    try:
        from agent.monitoring.policy import ensure_install_id
        return str(ensure_install_id(config))
    except Exception:
        return "unknown"


def _supervision_mode() -> str:
    if os.environ.get("INVOCATION_ID"):
        return "systemd"
    if os.environ.get("S6_CMD_ARG0") or os.environ.get("S6_VERSION"):
        return "s6"
    if os.environ.get("container") or os.path.exists("/.dockerenv"):
        return "container"
    if os.environ.get("LAUNCHD_SOCKET"):
        return "launchd"
    return "manual"


def _read_gateway_snapshot(config: Dict[str, Any]):
    from agent.monitoring.gateway_health import build_gateway_health_snapshot
    try:
        from gateway.status import read_runtime_status
        runtime = read_runtime_status() or {}
    except Exception:
        runtime = {}
    return build_gateway_health_snapshot(
        runtime,
        gateway_running=True,
        profile=_profile(),
        install_id=_install_id(config),
        version=_version(),
        supervision_mode=_supervision_mode(),
    )


def _read_cron_snapshot():
    from agent.monitoring.cron_health import build_cron_health_snapshot

    return build_cron_health_snapshot()


def _read_background_work_count() -> int:
    """Count live background/subagent work that ``active_agents`` does NOT include.

    ``hermes.gateway.active_agents`` counts foreground turns + in-flight cron
    jobs + API runs, but deliberately excludes backgrounded ``delegate_task``
    subagents, ``terminal(background=true)`` processes, kanban workers, and the
    runner's own background tasks (they are tracked only for the scale-to-zero
    suspend guard, ``_scale_to_zero_has_live_background_work``). Without this
    metric a peer churning through delegated subagents shows ``active_agents=0``
    on the fleet dashboard. Best-effort and content-free: a single integer,
    no job/task identity. Returns 0 if a source can't be imported.

    Delegation is counted TASK-granular (``active_task_count``): a fan-out batch
    of N subagents contributes N, not 1, so the metric reflects real concurrent
    subagent load rather than dispatch-unit/pool-slot count. This intentionally
    differs from the async pool's capacity accounting (one batch = one slot).
    """
    total = 0
    try:
        from tools.async_delegation import active_task_count

        total += max(0, int(active_task_count()))
    except Exception:
        logger.debug("background-work async-delegation count failed", exc_info=True)
    try:
        from tools.process_registry import process_registry

        total += max(0, int(process_registry.count_running()))
    except Exception:
        logger.debug("background-work process-registry count failed", exc_info=True)
    return total


def _read_background_delegations_count() -> int:
    """Count live async delegation UNITS (dispatch/pool slots).

    Complements ``_read_background_work_count`` (which is task-granular): this
    counts each ``delegate_task`` dispatch as ONE regardless of fan-out width,
    matching the async pool's capacity accounting (a batch = one slot). Together
    the two metrics let an operator see both slot pressure
    (``background_delegations``, alert vs ``max_concurrent_children``) and real
    concurrent subagent load (``background_work``). Delegations only — it does
    not include ``terminal(background)`` / kanban work, which are already folded
    into ``background_work``. Best-effort; 0 if the source can't be imported.
    """
    try:
        from tools.async_delegation import active_count

        return max(0, int(active_count()))
    except Exception:
        logger.debug("background-delegations count failed", exc_info=True)
        return 0


def _read_runtime_snapshot(config: Dict[str, Any]):
    gateway_snapshot = _read_gateway_snapshot(config)
    # Background/subagent work — a distinct metric from active_agents (which
    # never counts it). Appended to the gateway snapshot so it rides the same
    # base resource attributes (service.instance.id etc.).
    try:
        from agent.monitoring.gateway_health import GatewayMetric

        base = dict(gateway_snapshot.metrics[0].attributes) if gateway_snapshot.metrics else {}
        gateway_snapshot.metrics.append(
            GatewayMetric(
                name="hermes.gateway.background_work",
                value=_read_background_work_count(),
                attributes=base,
            )
        )
        gateway_snapshot.metrics.append(
            GatewayMetric(
                name="hermes.gateway.background_delegations",
                value=_read_background_delegations_count(),
                attributes=base,
            )
        )
    except Exception as exc:
        logger.warning(
            "background-work snapshot unavailable; metric not exported (error_type=%s)",
            type(exc).__name__,
        )
        logger.debug("background-work snapshot traceback", exc_info=True)
    try:
        cron_snapshot = _read_cron_snapshot()
    except Exception as exc:
        # Content-free visibility: cron telemetry silently dropping out is a
        # release-relevant regression, so surface it at WARNING with only the
        # exception *type* name (never the message, which could carry paths or
        # other environment detail). exc_info stays on the DEBUG record.
        logger.warning(
            "cron health snapshot unavailable; cron telemetry not exported (error_type=%s)",
            type(exc).__name__,
        )
        logger.debug("cron health snapshot traceback", exc_info=True)
        return gateway_snapshot
    gateway_snapshot.metrics.extend(cron_snapshot.metrics)
    return gateway_snapshot


def _emit_snapshot_events(config: Dict[str, Any]) -> None:
    gh = _gateway_health_config(config)
    if not gh.get("diagnostic_events_enabled", True):
        return
    try:
        from agent.monitoring import emitter
        snapshot = _read_runtime_snapshot(config)
        for event in snapshot.events:
            emitter.emit(event)
    except Exception:
        logger.debug("gateway health snapshot emit failed", exc_info=True)


def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any:
    gh = _gateway_health_config(config)
    if not gh.get("metrics_enabled", True):
        return None
    otlp = _otlp_config(config)
    endpoint = _metric_endpoint(str(otlp.get("endpoint")))
    headers = _resolve_headers(otlp.get("headers_env"))
    exporter = sdk["OTLPMetricExporter"](endpoint=endpoint, headers=headers or None)
    interval_ms = max(5, int(gh.get("export_interval_seconds", 60))) * 1000
    reader = sdk["PeriodicExportingMetricReader"](exporter, export_interval_millis=interval_ms)
    resource_attrs = _runtime_resource_attributes(
        config, telemetry_scope="gateway_health"
    )
    provider = sdk["MeterProvider"](
        metric_readers=[reader],
        resource=sdk["Resource"].create(resource_attrs),
    )
    meter = provider.get_meter("hermes.gateway.health")
    Observation = sdk["Observation"]

    metric_names = [
        "hermes.gateway.up",
        "hermes.gateway.state",
        "hermes.gateway.active_agents",
        "hermes.gateway.busy",
        "hermes.gateway.drainable",
        "hermes.gateway.restart_requested",
        "hermes.gateway.background_work",
        "hermes.gateway.background_delegations",
        "hermes.platform.up",
        "hermes.platform.degraded",
        "hermes.cron.scheduler.heartbeat_age_seconds",
        "hermes.cron.scheduler.last_success_age_seconds",
        "hermes.cron.scheduler.catch_up_occurrences",
        "hermes.cron.jobs.enabled",
        "hermes.cron.jobs.running",
        "hermes.cron.jobs.overdue",
    ]

    def callback(name: str):
        def _cb(_options=None):
            try:
                snapshot = _read_runtime_snapshot(config)
                return [Observation(m.value, m.attributes) for m in snapshot.metrics if m.name == name]
            except Exception:
                logger.debug("gateway metric callback failed", exc_info=True)
                return []
        return _cb

    for metric_name in metric_names:
        meter.create_observable_gauge(metric_name, callbacks=[callback(metric_name)])
    return provider


def _severity_number(sdk: Dict[str, Any], severity: Any) -> Any:
    SeverityNumber = sdk["SeverityNumber"]
    sev = str(severity or "warning").lower()
    if sev in {"critical", "fatal"}:
        return SeverityNumber.FATAL
    if sev == "error":
        return SeverityNumber.ERROR
    if sev in {"info", "information"}:
        return SeverityNumber.INFO
    if sev == "debug":
        return SeverityNumber.DEBUG
    return SeverityNumber.WARN


class GatewayDiagnosticLogStreamer:
    """Emitter subscriber that sends gateway diagnostic events as OTLP logs."""

    def __init__(self, config: Dict[str, Any], sdk: Dict[str, Any]):
        otlp = _otlp_config(config)
        headers = _resolve_headers(otlp.get("headers_env"))
        endpoint = _logs_endpoint(str(otlp.get("endpoint")))
        resource_attrs = _runtime_resource_attributes(
            config, telemetry_scope="gateway_diagnostics"
        )
        self._provider = sdk["LoggerProvider"](resource=sdk["Resource"].create(resource_attrs))
        self._processor = sdk["BatchLogRecordProcessor"](
            sdk["OTLPLogExporter"](endpoint=endpoint, headers=headers or None)
        )
        self._provider.add_log_record_processor(self._processor)
        self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE)
        self._LogRecord = sdk["LogRecord"]
        self._sdk = sdk
        self.exported = 0

    def __call__(self, batch: list[Dict[str, Any]]) -> None:
        from agent.monitoring.gateway_health import source_logger_for_export

        for ev in batch:
            if ev.get("event") != "gateway_diagnostic":
                continue
            attrs = _diagnostic_log_attributes(ev)
            # Preserve the source-controlled Python logger as the OTel
            # instrumentation scope. This adds precise code attribution without
            # turning a fluid module layout into a maintained subsystem enum.
            # Rendered messages stay out because they may contain arbitrary IDs,
            # names, paths, or configured strings. A future, separately gated
            # ``diagnostic_detail: redacted_message`` mode may add best-effort
            # free text when an observability plane defines that privacy policy.
            source_logger = source_logger_for_export(ev.get("source_logger"))
            otel_logger = (
                self._provider.get_logger(source_logger)
                if source_logger is not None
                else self._logger
            )
            body = "gateway diagnostic"
            record = self._LogRecord(
                timestamp=ev.get("ts_ns"),
                trace_id=self._sdk["INVALID_TRACE_ID"],
                span_id=self._sdk["INVALID_SPAN_ID"],
                trace_flags=self._sdk["TraceFlags"].DEFAULT,
                severity_text=str(ev.get("severity") or "warning").upper(),
                severity_number=_severity_number(self._sdk, ev.get("severity")),
                body=_redact_string(body),
                attributes=attrs,
            )
            otel_logger.emit(record)
            self.exported += 1

    def shutdown(self) -> None:
        try:
            from agent.monitoring.emitter import get_emitter
            get_emitter().unsubscribe(self)
        except Exception:
            pass
        try:
            self._processor.force_flush()
            self._provider.shutdown()
        except Exception:
            pass


def _start_diagnostic_log_streamer(config: Dict[str, Any], sdk: Dict[str, Any]) -> GatewayDiagnosticLogStreamer:
    from agent.monitoring.emitter import get_emitter
    streamer = GatewayDiagnosticLogStreamer(config, sdk)
    get_emitter().subscribe(streamer)
    return streamer


def _start_snapshot_thread(config: Dict[str, Any], stop_event: threading.Event) -> threading.Thread:
    interval = max(5, int(_gateway_health_config(config).get("logs_export_interval_seconds", 5)))

    def _run() -> None:
        while not stop_event.wait(interval):
            _emit_snapshot_events(config)

    thread = threading.Thread(target=_run, name="hermes-gateway-health-export", daemon=True)
    thread.start()
    return thread


def _attach_log_handler(config: Dict[str, Any]) -> Any:
    gh = _gateway_health_config(config)
    if not gh.get("diagnostic_events_enabled", True) or not gh.get("warning_error_events_enabled", True):
        return None
    from agent.monitoring.gateway_health import GatewayDiagnosticLogHandler
    handler = GatewayDiagnosticLogHandler(profile=_profile(), version=_version())
    root = logging.getLogger()
    if handler not in root.handlers:
        root.addHandler(handler)
    return handler


def _gateway_health_event(ev: Dict[str, Any]) -> bool:
    return ev.get("event") in {"gateway_health", "cron_execution"}


def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRuntime:
    """Start P0 gateway health export if configured. Never raises."""
    if not _enabled(config):
        return GatewayHealthExportRuntime(enabled=False, reason="disabled")
    gh = _gateway_health_config(config)
    runtime = GatewayHealthExportRuntime(enabled=True, reason="enabled")
    sdk: Optional[Dict[str, Any]] = None

    if gh.get("metrics_enabled", True) or gh.get("diagnostic_events_enabled", True):
        try:
            sdk = _require_metrics_sdk(prompt=False)
        except Exception:
            logger.warning(
                "monitoring.gateway_health_export.enabled but OTLP SDK is unavailable; "
                "install 'hermes-agent[otlp]'",
                exc_info=True,
            )
            return GatewayHealthExportRuntime(enabled=False, reason="otlp_unavailable")

    if gh.get("metrics_enabled", True) and sdk is not None:
        try:
            runtime.metric_provider = _start_metric_provider(config, sdk)
        except Exception:
            logger.warning("gateway health OTLP metrics failed to start", exc_info=True)
            runtime.shutdown()
            return GatewayHealthExportRuntime(enabled=False, reason="metrics_start_failed")

    if gh.get("diagnostic_events_enabled", True) and sdk is not None:
        try:
            from agent.monitoring import otlp_exporter
            runtime.streamer = otlp_exporter.start_streaming(config, event_filter=_gateway_health_event)
            if runtime.streamer is None:
                raise RuntimeError("gateway health span streamer did not start")
            runtime.log_streamer = _start_diagnostic_log_streamer(config, sdk)
        except Exception:
            logger.debug("gateway diagnostic OTLP export failed to start", exc_info=True)
            runtime.shutdown()
            return GatewayHealthExportRuntime(enabled=False, reason="diagnostics_start_failed")

    try:
        runtime.log_handler = _attach_log_handler(config)
    except Exception:
        logger.debug("gateway diagnostic log handler failed to attach", exc_info=True)
    if gh.get("diagnostic_events_enabled", True):
        try:
            _emit_snapshot_events(config)
            runtime.stop_event = threading.Event()
            runtime.thread = _start_snapshot_thread(config, runtime.stop_event)
        except Exception:
            logger.debug("gateway health snapshot thread failed to start", exc_info=True)
    return runtime


__all__ = [
    "GatewayHealthExportRuntime",
    "start_gateway_health_export",
]
