scripts/operational_telemetry.py

#!/usr/bin/env python3
"""Privacy-safe operational telemetry primitives for Synderesis.

Only aggregate counters with closed, content-free dimensions are persisted.
The module deliberately has no fields for account, user, device, session,
request, URL, prompt, answer, file, repository, citation, or exception data.
"""

from __future__ import annotations

import json
import re
import sqlite3
from collections.abc import Iterable, Mapping
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any


SCHEMA_VERSION = 1
CONSENT_VERSION = 2
RETENTION_DAYS = 30
MAX_BATCH_EVENTS = 20
MAX_EXPORT_ROWS = 5_000

SOURCES = frozenset({"client", "server"})
SURFACES = frozenset(
    {
        "account",
        "article",
        "chat",
        "health",
        "home",
        "other",
        "platform",
        "privacy",
        "resources",
        "share",
        "solutions",
    }
)
EVENT_NAMES = frozenset(
    {
        "api_request",
        "client_fault",
        "control_activate",
        "control_change",
        "form_submit",
        "navigation",
        "network_outcome",
        "page_view",
    }
)
OUTCOMES = frozenset(
    {
        "attempt",
        "cancelled",
        "client_error",
        "disabled",
        "enabled",
        "network_error",
        "observed",
        "rejected",
        "server_error",
        "success",
    }
)
STATUS_BUCKETS = frozenset({"none", "2xx", "3xx", "4xx", "5xx", "network_error"})
DURATION_BUCKETS = frozenset(
    {"none", "lt_100ms", "100_499ms", "500_1999ms", "2_9s", "10s_plus"}
)
ERROR_CODES = frozenset(
    {
        "none",
        "api_error",
        "http_error",
        "internal_error",
        "invalid_request",
        "js_error",
        "model_generation_failed",
        "network_error",
        "rate_limited",
        "unhandled_rejection",
    }
)

# Client controls and server route groups share one bounded component dimension.
# These names are developer-authored taxonomy, never copied from the DOM or a URL.
COMPONENTS = frozenset(
    {
        "account_api_key_copy",
        "account_api_key_create",
        "account_api_key_revoke",
        "account_billing_checkout",
        "account_billing_portal",
        "account_delete_cancel",
        "account_delete_confirm",
        "account_delete_open",
        "account_extension_approve",
        "account_extension_cancel",
        "account_invite_claim",
        "account_login",
        "account_login_tab",
        "account_logout",
        "account_register",
        "account_register_tab",
        "account_waitlist_join",
        "account_workspace_open",
        "api_account",
        "api_admin",
        "api_auth",
        "api_billing",
        "api_browser_chat",
        "api_demo",
        "api_device",
        "api_health",
        "api_models",
        "api_other",
        "api_shares",
        "api_sources",
        "api_telemetry",
        "api_usage",
        "chat_attachment_add",
        "chat_attachment_remove",
        "chat_byok_delete",
        "chat_byok_enable",
        "chat_byok_remember",
        "chat_byok_save",
        "chat_citation_open",
        "chat_connections",
        "chat_dialog_close",
        "chat_download",
        "chat_download_format",
        "chat_download_selected",
        "chat_download_select_all",
        "chat_ghost",
        "chat_github_connect",
        "chat_github_context",
        "chat_github_disconnect",
        "chat_github_refresh",
        "chat_github_repository",
        "chat_history_delete",
        "chat_history_open",
        "chat_history_pin",
        "chat_history_rename",
        "chat_memory_add",
        "chat_memory_clear",
        "chat_memory_delete",
        "chat_memory_enable",
        "chat_memory_settings",
        "chat_model_tier",
        "chat_new",
        "chat_reading_list",
        "chat_reading_list_download",
        "chat_reading_list_format",
        "chat_reading_list_select_all",
        "chat_reasoning_effort",
        "chat_retry",
        "chat_share",
        "chat_share_copy",
        "chat_share_create",
        "chat_share_revoke",
        "chat_source_mode",
        "chat_sources",
        "chat_starter_prompt",
        "chat_stop",
        "chat_submit",
        "chat_theme",
        "chat_upload_rules",
        "chat_web_review",
        "chat_web_search",
        "client_runtime",
        "cookie_preferences",
        "navigation_external",
        "navigation_internal",
        "page",
        "privacy_preferences",
        "site_menu",
        "unclassified_control",
    }
)

ROUTE_COMPONENTS = (
    ("/health", "api_health"),
    ("/v1/account", "api_account"),
    ("/v1/auth", "api_auth"),
    ("/v1/billing", "api_billing"),
    ("/v1/browser/chat", "api_browser_chat"),
    ("/v1/demo", "api_demo"),
    ("/v1/device", "api_device"),
    ("/v1/models", "api_models"),
    ("/v1/shares", "api_shares"),
    ("/v1/sources", "api_sources"),
    ("/v1/telemetry", "api_telemetry"),
    ("/v1/usage", "api_usage"),
    ("/v1/admin", "api_admin"),
)

SAFE_RELEASE_RE = re.compile(r"^[a-z][a-z0-9_]{0,79}$")
MARKER_PREFIX = "SYNDERESIS_OPERATIONAL_EVENT "


class TelemetryContractError(ValueError):
    """A telemetry event did not match the closed aggregate contract."""


def utc_hour(value: datetime | None = None) -> str:
    """Return the server-stamped UTC hour bucket."""
    current = value or datetime.now(UTC)
    if current.tzinfo is None:
        current = current.replace(tzinfo=UTC)
    current = current.astimezone(UTC).replace(minute=0, second=0, microsecond=0)
    return current.strftime("%Y-%m-%dT%H:00:00Z")


def release_label(deployment: Mapping[str, Any] | None, api_version: str) -> str:
    """Build a bounded server-owned release label."""
    values = dict(deployment or {})
    for key in ("git_commit", "git_sha", "sha", "commit_sha", "version"):
        raw = str(values.get(key, "") or "").strip().lower()
        safe = re.sub(r"[^a-z0-9]+", "_", raw).strip("_")[:48]
        if safe:
            candidate = f"release_{safe}"[:80]
            if SAFE_RELEASE_RE.fullmatch(candidate):
                return candidate
    safe_version = re.sub(r"[^a-z0-9]+", "_", api_version.lower()).strip("_")
    candidate = f"api_{safe_version}"[:80] if safe_version else "unknown"
    return candidate if SAFE_RELEASE_RE.fullmatch(candidate) else "unknown"


def route_component(path: str) -> str | None:
    """Map a request path to a closed route group without returning the path."""
    normalized = str(path or "")
    for prefix, component in ROUTE_COMPONENTS:
        if normalized == prefix or normalized.startswith(f"{prefix}/"):
            return component
    return "api_other" if normalized.startswith("/v1/") else None


def surface_for_component(component: str) -> str:
    """Map a route/control component to a coarse surface."""
    if component == "api_health":
        return "health"
    if component.startswith("api_account") or component.startswith("account_"):
        return "account"
    if component == "api_browser_chat" or component.startswith("chat_"):
        return "chat"
    return "other"


def status_bucket(status_code: int | None) -> str:
    """Bucket an HTTP status without retaining its exact value."""
    if status_code is None:
        return "network_error"
    if 200 <= status_code < 300:
        return "2xx"
    if 300 <= status_code < 400:
        return "3xx"
    if 400 <= status_code < 500:
        return "4xx"
    if 500 <= status_code < 600:
        return "5xx"
    return "none"


def duration_bucket(duration_ms: float | int | None) -> str:
    """Bucket request duration into fixed low-cardinality bands."""
    if duration_ms is None:
        return "none"
    value = max(0.0, float(duration_ms))
    if value < 100:
        return "lt_100ms"
    if value < 500:
        return "100_499ms"
    if value < 2_000:
        return "500_1999ms"
    if value < 10_000:
        return "2_9s"
    return "10s_plus"


def outcome_for_status(status_code: int | None) -> str:
    """Return a fixed outcome for an HTTP result."""
    if status_code is None:
        return "network_error"
    if status_code < 400:
        return "success"
    if status_code < 500:
        return "client_error"
    return "server_error"


def error_code_for_status(status_code: int | None, fixed_code: str = "") -> str:
    """Return only an allowlisted error family."""
    if fixed_code in ERROR_CODES:
        return fixed_code
    if status_code is None:
        return "network_error"
    if status_code == 422:
        return "invalid_request"
    if status_code == 429:
        return "rate_limited"
    if status_code >= 500:
        return "internal_error"
    if status_code >= 400:
        return "http_error"
    return "none"


def _closed(value: Any, allowed: frozenset[str], field: str) -> str:
    normalized = str(value or "")
    if normalized not in allowed:
        raise TelemetryContractError(f"invalid {field}")
    return normalized


def normalize_event(event: Mapping[str, Any]) -> dict[str, str]:
    """Validate an aggregate event and return exactly the persisted dimensions."""
    if not isinstance(event, Mapping):
        raise TelemetryContractError("event must be an object")
    expected = {
        "source",
        "surface",
        "event_name",
        "component",
        "outcome",
        "status_bucket",
        "duration_bucket",
        "error_code",
        "release",
    }
    if set(event) != expected:
        raise TelemetryContractError("event fields do not match the contract")
    release = str(event.get("release") or "")
    if not SAFE_RELEASE_RE.fullmatch(release):
        raise TelemetryContractError("invalid release")
    return {
        "source": _closed(event.get("source"), SOURCES, "source"),
        "surface": _closed(event.get("surface"), SURFACES, "surface"),
        "event_name": _closed(event.get("event_name"), EVENT_NAMES, "event_name"),
        "component": _closed(event.get("component"), COMPONENTS, "component"),
        "outcome": _closed(event.get("outcome"), OUTCOMES, "outcome"),
        "status_bucket": _closed(
            event.get("status_bucket"), STATUS_BUCKETS, "status_bucket"
        ),
        "duration_bucket": _closed(
            event.get("duration_bucket"), DURATION_BUCKETS, "duration_bucket"
        ),
        "error_code": _closed(event.get("error_code"), ERROR_CODES, "error_code"),
        "release": release,
    }


def normalize_client_event(
    event: Mapping[str, Any],
    *,
    release: str,
) -> dict[str, str]:
    """Validate a client event; source and release are always server-owned."""
    expected = {
        "surface",
        "event_name",
        "component",
        "outcome",
        "status_bucket",
        "duration_bucket",
        "error_code",
    }
    if not isinstance(event, Mapping) or set(event) != expected:
        raise TelemetryContractError("client event fields do not match the contract")
    return normalize_event({"source": "client", "release": release, **dict(event)})


def ensure_schema(connection: sqlite3.Connection) -> None:
    """Create the aggregate-only telemetry table."""
    connection.executescript(
        """
        CREATE TABLE IF NOT EXISTS operational_event_rollups (
            bucket_start TEXT NOT NULL,
            schema_version INTEGER NOT NULL,
            source TEXT NOT NULL,
            surface TEXT NOT NULL,
            event_name TEXT NOT NULL,
            component TEXT NOT NULL,
            outcome TEXT NOT NULL,
            status_bucket TEXT NOT NULL,
            duration_bucket TEXT NOT NULL,
            error_code TEXT NOT NULL,
            release TEXT NOT NULL,
            event_count INTEGER NOT NULL CHECK (event_count > 0),
            updated_at TEXT NOT NULL,
            PRIMARY KEY (
                bucket_start, schema_version, source, surface, event_name,
                component, outcome, status_bucket, duration_bucket,
                error_code, release
            )
        );
        CREATE INDEX IF NOT EXISTS idx_operational_event_rollups_bucket
            ON operational_event_rollups(bucket_start);
        """
    )


def increment_rollup(
    connection: sqlite3.Connection,
    event: Mapping[str, Any],
    *,
    count: int = 1,
    now: datetime | None = None,
) -> None:
    """Atomically increment one validated aggregate counter."""
    normalized = normalize_event(event)
    if (
        isinstance(count, bool)
        or not isinstance(count, int)
        or not 1 <= count <= 1_000_000
    ):
        raise TelemetryContractError("invalid count")
    bucket = utc_hour(now)
    updated_at = (
        (now or datetime.now(UTC)).astimezone(UTC).isoformat(timespec="seconds")
    )
    connection.execute(
        """
        INSERT INTO operational_event_rollups (
            bucket_start, schema_version, source, surface, event_name,
            component, outcome, status_bucket, duration_bucket, error_code,
            release, event_count, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT (
            bucket_start, schema_version, source, surface, event_name,
            component, outcome, status_bucket, duration_bucket, error_code,
            release
        ) DO UPDATE SET
            event_count = operational_event_rollups.event_count + excluded.event_count,
            updated_at = excluded.updated_at
        """,
        (
            bucket,
            SCHEMA_VERSION,
            normalized["source"],
            normalized["surface"],
            normalized["event_name"],
            normalized["component"],
            normalized["outcome"],
            normalized["status_bucket"],
            normalized["duration_bucket"],
            normalized["error_code"],
            normalized["release"],
            count,
            updated_at,
        ),
    )


def purge_expired(
    connection: sqlite3.Connection,
    *,
    now: datetime | None = None,
) -> int:
    """Delete aggregate buckets older than the fixed retention period."""
    current = now or datetime.now(UTC)
    cutoff = utc_hour(current - timedelta(days=RETENTION_DAYS))
    cursor = connection.execute(
        "DELETE FROM operational_event_rollups WHERE bucket_start < ?",
        (cutoff,),
    )
    return max(0, int(cursor.rowcount or 0))


def record_events(
    db_path: Path,
    events: Iterable[Mapping[str, Any]],
    *,
    now: datetime | None = None,
) -> int:
    """Persist a bounded batch of already-normalized aggregate events."""
    rows = list(events)
    if not 1 <= len(rows) <= MAX_BATCH_EVENTS:
        raise TelemetryContractError("invalid batch size")
    connection = sqlite3.connect(db_path, timeout=5)
    try:
        ensure_schema(connection)
        for event in rows:
            increment_rollup(connection, event, now=now)
        purge_expired(connection, now=now)
        connection.commit()
    finally:
        connection.close()
    return len(rows)


def export_rollups(
    db_path: Path,
    *,
    hours: int = 24,
    limit: int = MAX_EXPORT_ROWS,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Return a deterministic, identifier-free operational export."""
    if isinstance(hours, bool) or not isinstance(hours, int) or not 1 <= hours <= 720:
        raise TelemetryContractError("invalid hours")
    if (
        isinstance(limit, bool)
        or not isinstance(limit, int)
        or not 1 <= limit <= 20_000
    ):
        raise TelemetryContractError("invalid limit")
    current = (now or datetime.now(UTC)).astimezone(UTC)
    # Include exactly ``hours`` hourly buckets: the current bucket plus the
    # preceding ``hours - 1``. Daily 24-hour exports therefore do not share a
    # boundary bucket merely because the comparison is inclusive.
    cutoff = utc_hour(current - timedelta(hours=hours - 1))
    connection = sqlite3.connect(db_path, timeout=5)
    connection.row_factory = sqlite3.Row
    try:
        ensure_schema(connection)
        rows = connection.execute(
            """
            SELECT
                source, surface, event_name, component, outcome,
                status_bucket, duration_bucket, error_code, release,
                SUM(event_count) AS event_count
            FROM operational_event_rollups
            WHERE bucket_start >= ?
            GROUP BY
                source, surface, event_name, component, outcome,
                status_bucket, duration_bucket, error_code, release
            ORDER BY
                source, surface, event_name, component, outcome,
                status_bucket, duration_bucket, error_code, release
            LIMIT ?
            """,
            (cutoff, limit + 1),
        ).fetchall()
    finally:
        connection.close()
    truncated = len(rows) > limit
    rows = rows[:limit]
    return {
        "object": "synderesis.operational_events.export",
        "schema_version": SCHEMA_VERSION,
        "generated_at": current.isoformat(timespec="seconds"),
        "window_hours": hours,
        "retention_days": RETENTION_DAYS,
        "truncated": truncated,
        "rollups": [
            {
                "source": str(row["source"]),
                "surface": str(row["surface"]),
                "event_name": str(row["event_name"]),
                "component": str(row["component"]),
                "outcome": str(row["outcome"]),
                "status_bucket": str(row["status_bucket"]),
                "duration_bucket": str(row["duration_bucket"]),
                "error_code": str(row["error_code"]),
                "release": str(row["release"]),
                "event_count": int(row["event_count"]),
            }
            for row in rows
        ],
    }


def marker(event: Mapping[str, Any], *, count: int = 1) -> str:
    """Serialize a single safe aggregate marker for log fallback discovery."""
    normalized = normalize_event(event)
    payload = {
        "object": "synderesis.operational_event",
        "schema_version": SCHEMA_VERSION,
        **normalized,
        "event_count": count,
    }
    return MARKER_PREFIX + json.dumps(payload, sort_keys=True, separators=(",", ":"))